project/tbsg/audit_ocr.py

# -*- coding: utf-8 -*-
"""
OCR 准确率核对工具 — 用截图当标准答案，核对 Excel 采集数据
=============================================================================
用法:
    python audit_ocr.py                        # 默认核对 采集数据/ 最新一天的 xlsx
    python audit_ocr.py 采集数据/20260909.xlsx  # 指定文件
    python audit_ocr.py xxx.xlsx --max 10      # 最多核对10张截图（PP-OCR较慢，约30~60秒/张）

核对口径:
  A. 店名字准率: Excel里 列表页OCR店名 vs 详情页AI店名（独立来源互为答案）
  B. 现场覆盖率: 来源截图列 追溯到原图 → 本地PP-OCR重新识别 → 图上枚举出的完整卡片
     是否都进了 Excel（漏采检测）；Excel行记录的价格是否真的出现在原图上（追溯检测）
说明:
  B 口径的图上枚举用本地PP-OCR+几何规则（价格块+下方店铺块+上方标题块），与GLM无关，
  所以能发现 GLM 漏卡/错配；字级别的标题准确率仍以人工抽查为准。
"""
import json
import random
import re
import sys
from pathlib import Path

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

MAX_GAP = 216          # 价格→店铺窗口（与采集端校验一致，按1220x2712标定）
TITLE_WIN = 330        # 标题窗口（价格上方）
PROMO_RE = re.compile(r"起送|满\d|配送|发货|日达|红包|已售|月售|预估|减\d|处方|OTC|同款|低价|原研|新客|超值")


def norm(s):
    return re.sub(r"\s+", "", str(s or "")).lower()


def cy(b):
    return (b["box"][1] + b["box"][3]) // 2


def cx(b):
    return (b["box"][0] + b["box"][2]) // 2


def price_num(p):
    m = re.search(r"[\d.]+", str(p or ""))
    return m.group(0).rstrip(".") if m else ""


def edit_dist(a, b):
    if abs(len(a) - len(b)) > 2:
        return 99
    prev = list(range(len(b) + 1))
    for i, ca in enumerate(a, 1):
        cur = [i]
        for j, cb in enumerate(b, 1):
            cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
        prev = cur
    return prev[-1]


def shop_match(a, b):
    na, nb = norm(a), norm(b)
    if len(na) < 3 or len(nb) < 3:
        return False
    if na in nb or nb in na:
        return True
    return edit_dist(na, nb) <= 1


def load_excel(path):
    from openpyxl import load_workbook
    wb = load_workbook(path, read_only=True)
    rows = []
    for ws in wb.worksheets:
        for r in list(ws.iter_rows(values_only=True))[1:]:
            t, p, s1, s2, snap, link, src = (list(r) + [None] * 7)[:7]
            rows.append({"sheet": ws.title, "title": str(t or ""), "price": str(p or ""),
                         "shop": str(s1 or ""), "shop_ai": str(s2 or ""), "src": str(src or "")})
    return rows


def shop_accuracy(rows):
    """口径A: 列表OCR店名 vs 详情页AI店名"""
    comparable = [r for r in rows if norm(r["shop"]) and norm(r["shop_ai"])]
    exact = sum(1 for r in comparable if norm(r["shop"]) == norm(r["shop_ai"]))
    fuzzy = [r for r in comparable
             if norm(r["shop"]) != norm(r["shop_ai"]) and shop_match(r["shop"], r["shop_ai"])]
    diff = [r for r in comparable
            if norm(r["shop"]) != norm(r["shop_ai"]) and not shop_match(r["shop"], r["shop_ai"])]
    print("=" * 78)
    print(f"口径A 店名字准率（可比 {len(comparable)}/{len(rows)} 行）")
    print(f"  完全一致   {exact:>4}  {exact/len(comparable)*100 if comparable else 0:.1f}%")
    print(f"  同店近似   {len(fuzzy):>4}  （错1~2字/前后缀差）")
    print(f"  不一致     {len(diff):>4}  ← 逐条人工判断是'读错字'还是'配错卡'")
    for r in diff:
        print(f"    [{r['sheet']}] {r['title'][:30]} {r['price']}")
        print(f"       列表OCR: {r['shop'][:24]}  ↔  详情AI: {r['shop_ai'][:24]}  来源:{r['src']}")
    n = len(comparable)
    if n:
        print(f"  → 店名识别成功率 ≈ {(exact + len(fuzzy)) / n * 100:.1f}%")
    return diff


def split_right_col(blocks):
    """定位右列分界x（同采集端：￥块左边缘中位数-15，图片区在左、文字区在右）"""
    price_x = sorted(b["box"][0] for b in blocks if "¥" in b["text"] or "￥" in b["text"])
    if len(price_x) < 3:
        return 0
    return max(price_x[len(price_x) // 2] - 15, 0)


def enumerate_cards(blocks):
    """几何规则枚举图上的完整卡片: 价格块 + 下方店铺块 + 上方标题块（与GLM无关的独立读图）
    先裁掉左列商品图（包装文字会污染店铺/标题），价格块必须以¥开头（排除'起送¥0'类促销行）"""
    split = split_right_col(blocks)
    blocks = [b for b in blocks if cx(b) >= split] if split else blocks
    cards = []
    used_shop = []
    prices = [b for b in blocks if norm(b["text"]).startswith(("¥", "￥"))
              and re.search(r"¥[\d.]+|￥[\d.]+", b["text"])]
    prices.sort(key=cy)
    for pb in prices:
        py = cy(pb)
        pn = price_num(pb["text"])
        if not pn:
            continue
        # 店铺: 价格下方最近的有效块
        shop = None
        for b in sorted(blocks, key=cy):
            y = cy(b)
            if not (py + 5 < y < py + MAX_GAP):
                continue
            t = norm(b["text"])
            if len(t) < 4 or "¥" in b["text"] or "￥" in b["text"] or PROMO_RE.search(t):
                continue
            if any(abs(y - u) < 25 for u in used_shop):
                continue
            shop = b
            break
        if shop is None:
            continue
        used_shop.append(cy(shop))
        # 标题: 价格上方最长有效块
        title, tlen = "", 0
        for b in blocks:
            y = cy(b)
            if not (py - TITLE_WIN < y < py - 5):
                continue
            t = norm(b["text"])
            if len(t) < 6 or "¥" in b["text"] or "￥" in b["text"] or PROMO_RE.search(t):
                continue
            if len(t) > tlen:
                title, tlen = b["text"], len(t)
        cards.append({"price": pn, "shop": re.split(r"[(（]", re.sub(r"\s+", "", shop["text"]))[0],
                      "title": re.sub(r"\s+", "", title), "py": py})
    return cards


def field_audit(rows, max_images):
    """口径B: 来源截图追溯 + 图上卡片漏采检测"""
    src_rows = [r for r in rows if r["src"]]
    if not src_rows:
        print("\n口径B 跳过：该Excel没有「来源截图」列数据（列是今天新加的，明天采集后生效）")
        return
    try:
        from ai_helper_vision1 import VisionParser1
    except Exception as e:
        print(f"\n口径B 跳过：PP-OCR加载失败({e})")
        return

    # 来源 → 截图文件（取 base 文件；若被覆盖可用同名时间戳副本里最新的一张）
    shots = {}
    for r in src_rows:
        dev, batch = r["src"].split("/step3_b", 1)
        base = ROOT / "screenshots" / dev / "step3" / f"step3_b{batch}.png"
        if base.exists():
            shots.setdefault(str(base), (dev, batch))
        else:  # 找时间戳副本
            cands = sorted((ROOT / "screenshots" / dev / "step3").glob(f"step3_b{batch}_*.png"),
                           key=lambda p: p.stat().st_mtime)
            if cands:
                shots.setdefault(str(cands[-1]), (dev, batch))

    parser = VisionParser1()
    keys = list(shots.keys())
    if max_images and len(keys) > max_images:
        keys = random.sample(keys, max_images)   # 跨设备随机抽样,避免样本偏斜
    total_cards = matched = 0
    missing_report, trace_fail = [], []
    audited_srcs = {shots[k] for k in keys}
    print("\n" + "=" * 78)
    print(f"口径B 现场核对（{len(keys)}/{len(shots)} 张截图，PP-OCR本地识别，每张约30~60秒）")
    for k in keys:
        dev, batch = shots[k]
        try:
            blocks = parser._ppocr_blocks(k, 0)   # 整图识别，由几何规则自行分列
        except Exception as e:
            print(f"  [跳过] {dev}/b{batch}: OCR失败 {e}")
            continue
        cards = enumerate_cards(blocks)
        # 图上卡片 → Excel 全量匹配（价格+店名）
        hit = 0
        for c in cards:
            rows_p = [r for r in rows if price_num(r["price"]) == c["price"]]
            if any(shop_match(c["shop"], r["shop"]) for r in rows_p):
                hit += 1
            else:
                missing_report.append((dev, batch, c))
        total_cards += len(cards)
        matched += hit
        print(f"  {dev}/b{batch}: 图上{len(cards)}卡, 入Excel {hit}")
        # 追溯: 来源=这张图的Excel行,其价格必须出现在图上
        for r in src_rows:
            if (r["src"].split("/step3_b")[0], r["src"].split("/step3_b")[1]) == (dev, batch):
                pn = price_num(r["price"])
                if pn and not any(pn in norm(b["text"]) for b in blocks if "¥" in b["text"] or "￥" in b["text"]):
                    trace_fail.append(r)
    print("-" * 78)
    if total_cards:
        print(f"图上卡片共 {total_cards} 张, 已入Excel {matched} 张 → 现场覆盖率 {matched/total_cards*100:.1f}%")
    for dev, batch, c in missing_report:
        print(f"  [图上有但Excel没有] {dev}/b{batch} ¥{c['price']} {c['shop'][:14]} {c['title'][:26]}")
    if trace_fail:
        print(f"追溯异常 {len(trace_fail)} 行（Excel记的价格在来源截图上找不到）:")
        for r in trace_fail[:10]:
            print(f"    [{r['sheet']}] {r['title'][:26]} {r['price']} 来源:{r['src']}")
    if not trace_fail:
        print("追溯检测: 全部Excel行的价格都能在其来源截图上找到 ✓")


def main():
    args = [a for a in sys.argv[1:] if not a.startswith("--")]
    max_images = None
    if "--max" in sys.argv:
        try:
            max_images = int(sys.argv[sys.argv.index("--max") + 1])
        except (IndexError, ValueError):
            pass
    if args:
        xlsx = ROOT / args[0]
    else:
        cands = sorted((ROOT / "采集数据").glob("*.xlsx"), key=lambda p: p.stat().st_mtime)
        if not cands:
            print("采集数据/ 下没有 xlsx"); sys.exit(1)
        xlsx = cands[-1]
    if not xlsx.exists():
        print(f"文件不存在: {xlsx}"); sys.exit(1)
    print(f"核对文件: {xlsx}\n")
    rows = load_excel(xlsx)
    print(f"共 {len(rows)} 行, {len(set(r['sheet'] for r in rows))} 个药品\n")
    shop_accuracy(rows)
    field_audit(rows, max_images)
    print("\n提示: 口径B依赖「来源截图」列（今天新加），旧数据该列为空会自动跳过。")


if __name__ == "__main__":
    main()