project/tbsg/check_cross.py

# -*- coding: utf-8 -*-
"""
交叉验证效果核对 — 用来源截图独立重读当"真值"，逐条评判交叉验证的每个决定
=============================================================================
用法: python check_cross.py            # 核对 logs/cross_validate.jsonl 里全部记录
     python check_cross.py --max 20   # 只核对最近20条有分歧的记录(PP-OCR较慢)

评判口径(真值 = PP-OCR+几何规则在来源截图上枚举出的完整卡片):
  分歧候选卡(仲裁对象):
    保留+图上真有完整卡 → 正确保留        保留+图上没有 → 错误保留(脏数据)
    丢弃+图上真没有     → 正确丢弃(残卡)   丢弃+图上有   → 误杀(漏卡)
  错误率对比(同一屏):
    单次识别错误数(run1: 多余+漏卡) vs 交叉最终错误数 → 错误是否减少
"""
import json
import sys
from pathlib import Path

ROOT = Path(__file__).parent
sys.path.insert(0, str(ROOT))

from audit_ocr import enumerate_cards, price_num, shop_match
from ai_helper_vision1 import VisionParser1

LOG = ROOT / "logs" / "cross_validate.jsonl"


def title_ov(a, b, min_len=6):
    a, b = str(a or ""), str(b or "")
    best, prev = 0, [0] * (len(b) + 1)
    for ca in a:
        cur = [0]
        for j, cb in enumerate(b, 1):
            cur.append(prev[j - 1] + 1 if ca == cb else 0)
            best = max(best, cur[-1])
        prev = cur
    return best >= min_len


def cand_matches_truth(cand, truth_cards):
    """分歧候选卡是否与图上某张完整卡配得上(价格+店名+标题)"""
    pn = price_num(cand.get("price"))
    if not pn:
        return False
    for t in truth_cards:
        if t["price"] == pn and shop_match(cand.get("shop"), t["shop"]) \
                and title_ov(cand.get("title"), t["title"]):
            return True
    return False


def card_errors(cards, truth_cards):
    """(多余数, 漏卡数): cards vs 图上真值"""
    extras = sum(0 if cand_matches_truth(c, truth_cards) else 1 for c in cards)
    used = set()
    misses = 0
    for t in truth_cards:
        hit = False
        for i, c in enumerate(cards):
            if i in used:
                continue
            if price_num(c.get("price")) == t["price"] and shop_match(c.get("shop"), t["shop"]) \
                    and title_ov(c.get("title"), t["title"]):
                hit = True
                used.add(i)
                break
        if not hit:
            misses += 1
    return extras, misses


def main():
    max_n = 0
    if "--max" in sys.argv:
        try:
            max_n = int(sys.argv[sys.argv.index("--max") + 1])
        except (IndexError, ValueError):
            pass
    if not LOG.exists():
        print("没有日志: logs/cross_validate.jsonl (交叉版跑过采集后生成)")
        sys.exit(1)
    recs = [json.loads(ln) for ln in LOG.read_text(encoding="utf-8").splitlines() if ln.strip()]
    disputed = [r for r in recs if (r.get("a_only") or r.get("b_only")) and r.get("image")]
    if max_n:
        disputed = disputed[-max_n:]
    print(f"日志共 {len(recs)} 条, 有分歧需核对的 {len(disputed)} 条\n")
    if not disputed:
        print("抽样核对的记录里没有分歧卡 —— 交叉版与单次版无差异")
        return

    parser = VisionParser1()
    truth_cache = {}

    def truth_of(image):
        if image not in truth_cache:
            try:
                truth_cache[image] = enumerate_cards(parser._ppocr_blocks(image, 0))
            except Exception as e:
                print(f"  [真值枚举失败] {image}: {e}")
                truth_cache[image] = None
        return truth_cache[image]

    # ── 仲裁决定逐条评判 ──
    kept_ok = kept_bad = drop_ok = drop_bad = 0
    bad_detail = []
    for r in disputed:
        truth = truth_of(r["image"])
        if truth is None:
            continue
        cands = r.get("a_only", []) + r.get("b_only", [])
        if not cands:
            continue
        for c in cands:
            exists = cand_matches_truth(c, truth)
            was_kept = any(price_num(k.get("price")) == price_num(c.get("price"))
                           and shop_match(k.get("shop"), c.get("shop"))
                           and title_ov(k.get("title"), c.get("title"))
                           for k in r.get("kept", []))
            if was_kept:
                if exists:
                    kept_ok += 1
                else:
                    kept_bad += 1
                    bad_detail.append(f"    [错误保留→脏数据] ¥{c.get('price')} {c.get('shop')} {str(c.get('title'))[:24]} @ {Path(r['image']).name}")
            else:
                if exists:
                    drop_bad += 1
                    bad_detail.append(f"    [误杀→漏卡] ¥{c.get('price')} {c.get('shop')} {str(c.get('title'))[:24]} @ {Path(r['image']).name}")
                else:
                    drop_ok += 1
    total_decisions = kept_ok + kept_bad + drop_ok + drop_bad
    print("=" * 78)
    print(f"口径一 仲裁决定正确率（共 {total_decisions} 个分歧决定）")
    print(f"  正确保留 {kept_ok}   错误保留 {kept_bad}   正确丢弃 {drop_ok}   误杀 {drop_bad}")
    if total_decisions:
        print(f"  → 仲裁决定正确率 {(kept_ok + drop_ok) / total_decisions * 100:.1f}%")
    for d in bad_detail:
        print(d)

    # ── 错误率对比: 单次(run1) vs 交叉(final) ──
    n = 0
    err_run1 = err_final = 0
    extra1 = miss1 = extraf = missf = 0
    for r in recs:
        if r.get("status") != "ok" or not r.get("image") or not r.get("final_cards") and not r.get("run1"):
            continue
        truth = truth_of(r["image"])
        if truth is None:
            continue
        n += 1
        e1x, e1m = card_errors(r.get("run1", []), truth)
        efx, efm = card_errors(r.get("final_cards", []), truth)
        extra1 += e1x; miss1 += e1m; extraf += efx; missf += efm
        err_run1 += e1x + e1m
        err_final += efx + efm
    print("=" * 78)
    print(f"口径二 错误率对比（{n} 屏有真值）")
    print(f"  单次识别(run1) 错误总数: 多余{extra1} + 漏卡{miss1} = {err_run1}  ({err_run1/n:.2f}/屏)")
    print(f"  交叉最终错误总数:        多余{extraf} + 漏卡{missf} = {err_final}  ({err_final/n:.2f}/屏)")
    if err_run1:
        print(f"  → 错误减少 {(1 - err_final / err_run1) * 100:.0f}%")
    print("=" * 78)


if __name__ == "__main__":
    main()