tihao-ai/claude-code-tihao-sourcing/scripts/export-software-dedup-table.py

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()