import csv from collections import Counter, defaultdict from pathlib import Path from openpyxl import Workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side ROOT = Path(__file__).resolve().parents[1] OUTPUT_ROOT = ROOT / "outputs" / "overnight-quality-1780682025651" SOURCE = OUTPUT_ROOT / "manual-review-sample.csv" CSV_OUT = OUTPUT_ROOT / "software-client-list-dedup.csv" XLSX_OUT = OUTPUT_ROOT / "software-client-list-dedup.xlsx" SUMMARY_OUT = OUTPUT_ROOT / "software-client-list-dedup-summary.md" HEADERS = [ "brief编号", "策略", "排名", "平台", "博主名称", "综合分", "brief匹配分", "参考风格分", "证据加分", "证据风险扣分", "推荐理由", "风险提示", "主页链接", "人工复核标签", ] STRATEGY_PRIORITY = { "result-first-broad": 50, "result-first": 40, "video-enhanced": 30, "result-first-risk": 20, "baseline-live": 10, } def to_num(row, key): try: return float(str(row.get(key, "")).strip() or 0) except ValueError: return 0.0 def norm(text): return " ".join(str(text or "").strip().lower().split()) def dedupe_key(row): link = norm(row.get("主页链接")) if link: return (norm(row.get("brief编号")), norm(row.get("平台")), link) return (norm(row.get("brief编号")), norm(row.get("平台")), norm(row.get("博主名称"))) def quality_key(row): return ( STRATEGY_PRIORITY.get(row.get("策略"), 0), to_num(row, "综合分"), to_num(row, "brief匹配分"), to_num(row, "参考风格分"), to_num(row, "证据加分"), -to_num(row, "证据风险扣分"), -to_num(row, "排名"), ) def load_rows(): with SOURCE.open("r", encoding="utf-8-sig", newline="") as handle: rows = list(csv.DictReader(handle)) for row in rows: for header in HEADERS: row.setdefault(header, "") return rows def dedupe_rows(rows): kept = {} all_by_key = defaultdict(list) for row in rows: key = dedupe_key(row) all_by_key[key].append(row) if key not in kept or quality_key(row) > quality_key(kept[key]): kept[key] = dict(row) unique_rows = list(kept.values()) unique_rows.sort( key=lambda row: ( row.get("brief编号", ""), -to_num(row, "综合分"), -to_num(row, "brief匹配分"), -to_num(row, "参考风格分"), -to_num(row, "证据加分"), to_num(row, "证据风险扣分"), row.get("平台", ""), row.get("博主名称", ""), ) ) rank_by_brief = defaultdict(int) for row in unique_rows: brief = row.get("brief编号", "") rank_by_brief[brief] += 1 row["排名"] = str(rank_by_brief[brief]) row["人工复核标签"] = "" return unique_rows, all_by_key def write_csv(rows): with CSV_OUT.open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=HEADERS, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def write_xlsx(rows, original_count, duplicate_key_count): wb = Workbook() ws = wb.active ws.title = "去重名单" ws.append(HEADERS) for row in rows: ws.append([row.get(header, "") for header in HEADERS]) header_fill = PatternFill("solid", fgColor="1F4E78") header_font = Font(color="FFFFFF", bold=True) thin = Side(style="thin", color="D9E2F3") for cell in ws[1]: cell.fill = header_fill cell.font = header_font cell.alignment = Alignment(horizontal="center", vertical="center") cell.border = Border(bottom=thin) widths = { "A": 20, "B": 18, "C": 8, "D": 14, "E": 24, "F": 10, "G": 12, "H": 12, "I": 10, "J": 12, "K": 110, "L": 85, "M": 56, "N": 16, } for col, width in widths.items(): ws.column_dimensions[col].width = width for row in ws.iter_rows(min_row=2): for cell in row: cell.alignment = Alignment(vertical="top", wrap_text=True) for idx in [3, 6, 7, 8, 9, 10]: row[idx - 1].alignment = Alignment(horizontal="center", vertical="top", wrap_text=True) ws.freeze_panes = "A2" ws.auto_filter.ref = ws.dimensions summary = wb.create_sheet("去重说明") summary_rows = [ ["项目", "值"], ["来源文件", str(SOURCE)], ["原始行数", original_count], ["去重后行数", len(rows)], ["删除重复行数", original_count - len(rows)], ["存在重复的博主键数量", duplicate_key_count], ["去重口径", "同一 brief编号 + 平台 + 主页链接；没有主页链接时使用 brief编号 + 平台 + 博主名称"], ["保留规则", "优先 result-first-broad，其次 result-first、video-enhanced、result-first-risk、baseline-live；同优先级下保留综合分/brief匹配/参考风格/证据加分更高且风险扣分更低的一条"], ["排名规则", "去重后按每个 brief 重新从 1 排名"], ] for item in summary_rows: summary.append(item) summary.column_dimensions["A"].width = 22 summary.column_dimensions["B"].width = 130 for cell in summary[1]: cell.fill = header_fill cell.font = header_font for row in summary.iter_rows(): for cell in row: cell.alignment = Alignment(vertical="top", wrap_text=True) wb.save(XLSX_OUT) def write_summary(rows, original_count, duplicate_key_count): brief_counts = Counter(row["brief编号"] for row in rows) strategy_counts = Counter(row["策略"] for row in rows) lines = [ "# 软件端去重名单导出说明", "", f"- 来源：`{SOURCE}`", f"- 原始行数：{original_count}", f"- 去重后行数：{len(rows)}", f"- 删除重复行数：{original_count - len(rows)}", f"- 存在重复的博主键数量：{duplicate_key_count}", f"- CSV：`{CSV_OUT}`", f"- XLSX：`{XLSX_OUT}`", "", "## 各 brief 行数", "", ] for brief, count in sorted(brief_counts.items()): lines.append(f"- {brief}: {count}") lines.extend(["", "## 保留策略分布", ""]) for strategy, count in sorted(strategy_counts.items()): lines.append(f"- {strategy}: {count}") lines.extend( [ "", "## 去重与排序口径", "", "- 去重口径：同一 `brief编号 + 平台 + 主页链接` 视为同一博主；没有主页链接时用 `brief编号 + 平台 + 博主名称`。", "- 保留规则：优先发布策略 `result-first-broad`、`result-first`，再保留诊断策略；同优先级下看综合分、brief 匹配、参考风格、证据加分和风险扣分。", "- 排名规则：去重后按每个 brief 重新排序。", ] ) SUMMARY_OUT.write_text("\n".join(lines), encoding="utf-8-sig") def main(): rows = load_rows() unique_rows, all_by_key = dedupe_rows(rows) duplicate_key_count = sum(1 for values in all_by_key.values() if len(values) > 1) write_csv(unique_rows) write_xlsx(unique_rows, len(rows), duplicate_key_count) write_summary(unique_rows, len(rows), duplicate_key_count) print(f"rows_in={len(rows)}") print(f"rows_out={len(unique_rows)}") print(f"dropped={len(rows) - len(unique_rows)}") print(f"duplicate_keys={duplicate_key_count}") print(f"csv={CSV_OUT}") print(f"xlsx={XLSX_OUT}") print(f"summary={SUMMARY_OUT}") if __name__ == "__main__": main()