project/tbsg/_tmp_pairing_audit.py

# -*- coding: utf-8 -*-
"""
配对审计 — 验证 Excel 每行的 (店名, 药品标题, 价格) 是否真的来自同一张卡片
=============================================================================
方法: 取行的来源截图 → PP-OCR 独立重读 → 几何枚举卡片(标题/价格/店铺) →
     行的三元组与图上卡片逐一配对(价格精确 + 店名容错 + 标题重叠)。
     当前图配不上时,依次尝试该批次的时间戳备份图(防批次覆盖)。
用法: python _tmp_pairing_audit.py [样本行数,默认60]
"""
import random
import re
import sys
from pathlib import Path

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

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


def title_overlap(a, b, min_len=6):
    """最长公共子串 >= min_len 视为同一商品标题(容忍前后缀差异)"""
    a, b = norm(a), norm(b)
    best = 0
    prev = [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 main():
    n_sample = int(sys.argv[1]) if len(sys.argv) > 1 else 60
    rows = load_excel(ROOT / "采集数据" / "20260909.xlsx")
    src_rows = [r for r in rows if r["src"]]
    # 分层抽样: 按来源排序后等距取,保证4台设备/各药品都覆盖
    src_rows_sorted = sorted(src_rows, key=lambda r: r["src"])
    step = max(1, len(src_rows_sorted) // n_sample)
    sample = src_rows_sorted[::step][:n_sample]
    print(f"可追溯行 {len(src_rows)}, 抽样 {len(sample)} 行进行配对审计\n")

    parser = VisionParser1()
    card_cache = {}      # 图片路径 -> 枚举卡片
    confirmed = unconfirmed = 0
    fail_rows = []

    def cards_of(dev, batch):
        """该批次所有候选图(当前图+时间戳副本)的枚举卡片(带缓存)"""
        out = []
        sdir = ROOT / "screenshots" / dev / "step3"
        cands = [sdir / f"step3_b{batch}.png"]
        cands += sorted(sdir.glob(f"step3_b{batch}_*.png"), key=lambda p: p.stat().st_mtime)
        for img in cands:
            if not img.exists():
                continue
            key = str(img)
            if key not in card_cache:
                try:
                    card_cache[key] = enumerate_cards(parser._ppocr_blocks(key, 0))
                except Exception as e:
                    print(f"    [OCR失败] {img.name}: {e}")
                    card_cache[key] = []
            out.append((key, card_cache[key]))
        return out

    for idx, r in enumerate(sample, 1):
        dev, batch = r["src"].split("/step3_b")
        pn = price_num(r["price"])
        print(f"[{idx}/{len(sample)}] {r['src']} ¥{pn} {r['shop'][:12]} | {r['title'][:22]}", flush=True)
        hit = False
        tried = 0
        for key, cards in cards_of(dev, batch):
            tried += 1
            for c in cards:
                if (c["price"] == pn and shop_match(c["shop"], r["shop"])
                        and title_overlap(c["title"], r["title"])):
                    hit = True
                    break
            if hit:
                break
        if hit:
            confirmed += 1
            print(f"    ✓ 配对确认 (试了{tried}张候选图)")
        else:
            unconfirmed += 1
            fail_rows.append(r)
            print(f"    ✗ 未能在来源图上找到配对卡片")
        if idx % 10 == 0:
            print(f"    --- 进度: 确认{confirmed} 未确认{unconfirmed} ---", flush=True)

    n = confirmed + unconfirmed
    print("\n" + "=" * 70)
    print(f"配对审计结果: {confirmed}/{n} = {confirmed/n*100:.1f}% 的行 (店名+药品+价格) 三元组能在来源截图的同一张卡上得到验证")
    if fail_rows:
        print("未确认行(需人工看图/可能是批次覆盖或真错配):")
        for r in fail_rows:
            print(f"  [{r['sheet']}] {r['title'][:30]} {r['price']} 店:{r['shop'][:16]} 来源:{r['src']}")


if __name__ == "__main__":
    main()