tps-dashboard/ai-labeling/scripts/summarize.py

"""
summarize.py - 汇总打标结果命中情况
用法: python summarize.py --model mimo-v2.5-pro
"""

import sys
sys.stdout.reconfigure(encoding='utf-8')
sys.stderr.reconfigure(encoding='utf-8')

import json
import glob
import argparse
from pathlib import Path

BASE_DIR = Path(__file__).parent.parent
EXPERIMENTS_DIR = BASE_DIR / "experiments"


def load_json(path):
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return None


def latest_per_ep(files):
    """同 ep 有多个文件时取时间戳最新的。"""
    latest = {}
    for f in files:
        name = f.name
        # 文件名格式: 20260611_154037_model_field_ep003.json
        parts = name.replace(".json", "").split("_")
        if len(parts) >= 4:
            ts = parts[0] + parts[1]  # yyyymmddHHMMSS
            ep = int(parts[-1].replace("ep", ""))
            key = ep
            if key not in latest or ts > latest[key][1]:
                latest[key] = (f, ts)
    return {ep: info[0] for ep, info in latest.items()}


def run(model, field="narrative"):
    # 文件匹配：新格式有 field 标记，老格式（无 field）向后兼容
    pattern_new = str(EXPERIMENTS_DIR / f"*_{model}_{field}_*.json")
    files_new = sorted(Path(p) for p in glob.glob(pattern_new))
    # 如果是 narrative 且新格式文件为空，尝试匹配老格式（无 field 标记）
    if field == "narrative" and not files_new:
        pattern_old = str(EXPERIMENTS_DIR / f"*_{model}_ep*.json")
        files_old = [f for f in sorted(Path(p) for p in glob.glob(pattern_old))
                     if len(f.name.replace(".json", "").split("_")) == 3]  # 老格式: ts_model_epNN.json
        files = files_old
    else:
        files = files_new

    if not files:
        print(f"未找到 {pattern_new}")
        return

    ep_files = latest_per_ep(files)
    rows = []
    parse_fail = 0

    for ep in sorted(ep_files.keys()):
        data = load_json(ep_files[ep])
        if data is None:
            parse_fail += 1
            rows.append({"ep": ep, "title": "?", "gt": "?", "pred": "解析失败", "hit": False, "conf": "?"})
            continue

        gt = data.get("ground_truth", {})
        result = data.get("result")

        title = gt.get("title", "?")

        if field == "narrative":
            gt_val = gt.get("narrative_structure", "?")
            pred_val = result.get("narrative_structure") if result else None
            conf = result.get("confidence", "?") if result else "?"
            hit = pred_val == gt_val if pred_val is not None else False
            rows.append({"ep": ep, "title": title, "gt": gt_val, "pred": pred_val or "解析失败", "hit": hit, "conf": conf})
        elif field == "classification":
            conf = result.get("confidence", "?") if result else "?"
            sub = {}
            for key in ("program_format", "equipment_domain", "scene_tags", "tech_tags"):
                gt_val = gt.get(key)
                pred_val = result.get(key) if result else None
                if gt_val is None or pred_val is None:
                    sub[key] = None
                elif isinstance(gt_val, list):
                    sub[key] = set(gt_val) == set(pred_val) if isinstance(pred_val, list) else False
                else:
                    sub[key] = pred_val == gt_val
            all_hit = all(v is True for v in sub.values() if v is not None)
            rows.append({"ep": ep, "title": title, "sub": sub, "all_hit": all_hit, "conf": conf})

    if field == "narrative":
        # 打印每行
        for r in rows:
            mark = "✓" if r["hit"] else "✗"
            conf_str = f'置信度:{r["conf"]}' if r["conf"] != "?" else ""
            print(f'  ep{r["ep"]:02d} {r["title"]:<10} | 标准:{r["gt"]:<8} | {model}:{r["pred"]:<8} | {mark} {conf_str}')

        # 汇总
        total = len(rows)
        hits = sum(1 for r in rows if r["hit"])
        hi_conf = [r for r in rows if r["conf"] == "高"]
        mid_low = [r for r in rows if r["conf"] in ("中", "低")]
        hi_hit = sum(1 for r in hi_conf if r["hit"])
        ml_hit = sum(1 for r in mid_low if r["hit"])

        print(f"\n  ===== {model} 命中情况 =====")
        print(f"  narrative_structure 命中: {hits}/{total} = {hits*100//total}%")
        print(f"  自评\"高\"置信的命中率: {hi_hit}/{len(hi_conf)}")
        print(f"  自评\"中/低\"置信的命中率: {ml_hit}/{len(mid_low)}")
        print(f"  解析失败: {parse_fail} 期")
    elif field == "classification":
        field_names = ["program_format", "equipment_domain", "scene_tags", "tech_tags"]
        short = {"program_format": "题材", "equipment_domain": "装备域", "scene_tags": "场景", "tech_tags": "技术"}
        for r in rows:
            marks = ""
            for fn in field_names:
                v = r["sub"].get(fn)
                marks += "✓" if v is True else ("✗" if v is False else "-")
            all_mark = "✓" if r["all_hit"] else "✗"
            print(f'  ep{r["ep"]:03d} {r["title"]:<12} | {"/".join(short[f] for f in field_names)}={marks} | 全对:{all_mark} 置信度:{r["conf"]}')

        total = len(rows)
        for fn in field_names:
            scored = [r for r in rows if r["sub"].get(fn) is not None]
            hits = sum(1 for r in scored if r["sub"][fn])
            pct = hits * 100 // len(scored) if scored else 0
            print(f"  {short[fn]}命中: {hits}/{len(scored)} = {pct}%")
        all_scored = [r for r in rows if any(v is not None for v in r["sub"].values())]
        all_hits = sum(1 for r in all_scored if r["all_hit"])
        print(f"  4字段全对: {all_hits}/{len(all_scored)} = {all_hits*100//len(all_scored) if all_scored else 0}%")
        print(f"  解析失败: {parse_fail} 期")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--model", required=True, help="模型键名，如 mimo-v2.5-pro / deepseek-v4-pro")
    parser.add_argument("--field", default="narrative", choices=["narrative", "classification"],
                        help="打标字段: narrative(叙事结构) / classification(4分类)")
    args = parser.parse_args()
    run(args.model, args.field)