project/tbsg/ai_helper_vision_cross.py

"""
AI 视觉助手-交叉验证版（实验） — 与 ai_helper_vision1 平行存在, 不改动稳定版
=============================================================================
流程: 同一屏 OCR 独立识别 2 次
     → 两次都有的卡 = 高置信, 直接收
     → 只在一次出现的卡 = 分歧, 发第 3 次仲裁判真/假
       (真卡标准: 标题在价格上方 + 店铺名在价格下方 + 三者同一张卡)
     → 基础校验 + 几何校验(继承稳定版, 宁丢卡不配错)
实验数据(3张陷阱图×3轮): 完美 5/9→6/9, 平均漏卡 0.78→0.33, 代价=每屏2.2次GLM调用

接入方式(零改动切换): main1.py 里把
    from ai_helper_vision1 import VisionParser
改为
    from ai_helper_vision_cross import VisionParser
其余代码不动; 想换回就改回来。

独立测试: python ai_helper_vision_cross.py  (配置在 __main__ 的 TEST_IMAGE/TEST_KEYWORD)
"""
import json
import os
import re
import sys

from ai_helper_vision1 import VisionParser1

VisionParser = None  # 末尾赋值: VisionParserCross


ARBI_SYSTEM = """你是电商列表页 OCR 复核助手。对同一屏的 OCR 做了两次独立识别，两次结果有分歧。
分歧候选可能是：另一次漏识别的真卡片，也可能是把顶部残卡/相邻卡片误配出来的假卡片。
真卡片必须同时满足：标题在价格上方、店铺名在价格下方、三者属于同一张卡（y坐标相邻）。
判断后只保留真卡，字段值严格取自 OCR 原文。"""


class VisionParserCross(VisionParser1):
    """双次独立识别 + 交叉仲裁 版解析器（几何校验继承稳定版）"""

    # ── 单次识别调用 ────────────────────────────────────
    def _call_cards(self, system_prompt: str, user_text: str) -> tuple:
        """返回 (status, cards): status = ok / page_wrong / failed"""
        resp = self._call_llm(system_prompt, user_text)
        cleaned = (resp or "").strip()
        if cleaned.startswith("```"):
            ls = cleaned.split("\n")
            if ls and ls[0].startswith("```"):
                ls = ls[1:]
            if ls and ls[-1].strip() == "```":
                ls = ls[:-1]
            cleaned = "\n".join(ls).strip()
        try:
            data = json.loads(cleaned)
        except json.JSONDecodeError:
            m = re.search(r'\[[\s\S]*\]', cleaned)
            if m:
                try:
                    data = json.loads(m.group(0))
                except json.JSONDecodeError:
                    data = None
        if not isinstance(data, list):
            return "failed", []
        if len(data) == 1 and isinstance(data[0], dict) and data[0].get("page_wrong"):
            return "page_wrong", []
        return "ok", [d for d in data if isinstance(d, dict)]

    # ── 配对工具 ────────────────────────────────────────
    @staticmethod
    def _pnum(p) -> str:
        m = re.search(r"[\d.]+", str(p or ""))
        return m.group(0).rstrip(".") if m else ""

    @classmethod
    def _shop_eq(cls, a, b) -> bool:
        na, nb = cls._gnorm(a), cls._gnorm(b)
        if len(na) < 3 or len(nb) < 3:
            return False
        if na in nb or nb in na:
            return True
        if len(na) == len(nb):
            return sum(x != y for x, y in zip(na, nb)) <= 1
        return False

    @classmethod
    def _same_card(cls, c1: dict, c2: dict) -> bool:
        """两次识别的两张卡是否同一张: 价格相同 且 (店名容错相同 或 标题重叠≥6字)"""
        if cls._pnum(c1.get("price")) != cls._pnum(c2.get("price")):
            return False
        shop_ok = cls._shop_eq(c1.get("shop"), c2.get("shop"))
        title_ok = cls._glcs(cls._gnorm(c1.get("title")), cls._gnorm(c2.get("title"))) >= 6
        return shop_ok or title_ok

    # ── 基础校验(与稳定版逐字一致) ────────────────────────
    def _basic_filter(self, data: list, screen_w: int, screen_h: int) -> list:
        shops = []
        for i, item in enumerate(data, 1):
            if not isinstance(item, dict):
                continue
            title = str(item.get("title", "") or "")
            price = str(item.get("price", "") or "")
            shop = str(item.get("shop", "") or "")
            sales = str(item.get("sales", "") or "")
            click = item.get("click") or item.get("coords") or []
            if not (shop and title and price):
                print(f"[v1] 卡{i}: [丢弃] 字段不完整: {shop} | {title[:20]} | {price}")
                continue
            if '￥' in title or '¥' in title:
                print(f"[v1] 卡{i}: [丢弃] 标题含价格符号（残卡）")
                continue
            if isinstance(click, list) and len(click) == 2:
                try:
                    cx, cy = int(click[0]), int(click[1])
                except (TypeError, ValueError):
                    continue
                if not (0 <= cx <= screen_w and 0 <= cy <= screen_h):
                    print(f"[v1] 卡{i}: [丢弃] 坐标越界: ({cx},{cy})")
                    continue
            else:
                print(f"[v1] 卡{i}: [丢弃] 无有效点击坐标")
                continue
            shop_y = int(item.get("shop_y", cy)) if str(item.get("shop_y", "")).lstrip('-').isdigit() else cy
            print(f"[v1] 卡{i}: [收录] {shop} | {title[:36]} | {price} | click=({cx},{cy}) | 已售={sales}")
            shops.append([shop, title, price, [cx, cy], shop_y, sales])
        return shops

    # ── 分歧仲裁 ────────────────────────────────────────
    def _arbitrate(self, ocr_text: str, a_only: list, b_only: list) -> list:
        user = (f"{ARBI_SYSTEM}\n\nOCR原文：\n{ocr_text}\n\n"
                f"第一次识别独有的卡片：\n{json.dumps(a_only, ensure_ascii=False)}\n\n"
                f"第二次识别独有的卡片：\n{json.dumps(b_only, ensure_ascii=False)}\n\n"
                f'请输出纯JSON：{{"keep":[{{"title":"...","price":"...","shop":"...","click":[x,y],"shop_y":y,"sales":"..."}}]}}')
        resp = self._call_llm(ARBI_SYSTEM, user)
        try:
            m = re.search(r'\{[\s\S]*\}', resp or "")
            keep = json.loads(m.group(0)).get("keep", []) if m else []
            return [k for k in keep if isinstance(k, dict)]
        except Exception as e:
            print(f"[v1] 仲裁解析失败({e}), 分歧卡全部丢弃(保守)")
            return []

    # ── 决策日志(供 check_cross.py 事后核对仲裁正确率) ────
    @staticmethod
    def _log_cross(record: dict):
        try:
            from pathlib import Path
            log_dir = Path(__file__).parent / "logs"
            log_dir.mkdir(exist_ok=True)
            record["ts"] = __import__("time").strftime("%Y-%m-%d %H:%M:%S")
            with open(log_dir / "cross_validate.jsonl", "a", encoding="utf-8") as f:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")
        except Exception:
            pass

    def parse_shops(self, image_path: str, screen_size: tuple = (720, 1640),
                    keyword: str = "", crop_x: int = 0) -> tuple:
        """记录当前截图路径(供决策日志追溯), 其余与稳定版一致"""
        self._cur_image = str(image_path)
        return super().parse_shops(image_path, screen_size=screen_size, keyword=keyword, crop_x=crop_x)

    # ── 主流程(重写稳定版的GLM分卡部分) ────────────────────
    def _extract_all_cards(self, blocks: list, screen_w: int, screen_h: int, keyword: str = "") -> tuple:
        """双次独立识别 + 交叉仲裁 → 基础校验 → 几何校验。返回 (shops, status)"""
        bs = sorted(blocks, key=lambda b: b["box"][1])
        lines = []
        for b in bs:
            box = b["box"]
            cx = (box[0] + box[2]) // 2
            cy = (box[1] + box[3]) // 2
            lines.append(f"[x={cx:4d}, y={cy:4d}] {b.get('text','')}")
        ocr_text = "\n".join(lines)
        system_prompt = self._cards_prompt()
        user_text = (
            f"屏幕OCR识别结果（共{len(bs)}个文字块）：\n\n{ocr_text}\n\n"
            f"当前搜索的商品是「{keyword}」。请划分商品卡片并逐卡提取字段。"
        )

        def _compact(cards):
            return [{"shop": c.get("shop"), "price": c.get("price"),
                     "title": str(c.get("title"))[:40], "click": c.get("click")} for c in cards]

        rec = {"image": getattr(self, "_cur_image", ""), "keyword": keyword,
               "blocks": len(bs), "n1": 0, "n2": 0, "agreement": 0,
               "run1": [], "run2": [], "a_only": [], "b_only": [], "kept": [],
               "final_cards": [], "final": 0, "status": ""}

        # 第1次独立识别
        status_a, cards_a = self._call_cards(system_prompt, user_text)
        rec["n1"] = len(cards_a)
        rec["run1"] = _compact(cards_a)
        if status_a != "ok":
            print(f"[v1] 第1次识别: {status_a}")
            rec["status"] = status_a
            self._log_cross(rec)
            return [], status_a

        # 第2次独立识别
        status_b, cards_b = self._call_cards(system_prompt, user_text)
        rec["n2"] = len(cards_b)
        rec["run2"] = _compact(cards_b)
        if status_b == "page_wrong":
            print("[v1] 第2次识别判定: page_wrong")
            rec["status"] = "page_wrong"
            self._log_cross(rec)
            return [], "page_wrong"
        if status_b != "ok":
            # 第2次失败 → 无交叉, 退回单次结果
            print("[v1] 第2次识别失败, 退回单次结果")
            rec["status"] = "ok(单次降级)"
            shops = self._basic_filter(cards_a, screen_w, screen_h)
            shops = self._validate_cards(shops, bs, screen_w, screen_h)
            rec["final"] = len(shops)
            self._log_cross(rec)
            return shops, "ok"

        # 交叉: 两次都有的 = 高置信
        # 某一侧空返回(GLM偶发) → 直接采用另一侧, 跳过仲裁(省一次调用)
        if not cards_a and cards_b:
            print(f"[v1] 第1次识别空返回, 直接采用第2次的 {len(cards_b)} 张(免仲裁)")
            shops = self._basic_filter(cards_b, screen_w, screen_h)
            shops = self._validate_cards(shops, bs, screen_w, screen_h)
            rec["final"] = len(shops)
            rec["final_cards"] = [{"shop": s[0], "title": str(s[1])[:40], "price": s[2],
                                   "click": s[3]} for s in shops]
            rec["status"] = "ok(单侧空,采用另一侧)"
            self._log_cross(rec)
            return shops, "ok"
        if cards_a and not cards_b:
            print(f"[v1] 第2次识别空返回, 直接采用第1次的 {len(cards_a)} 张(免仲裁)")
            shops = self._basic_filter(cards_a, screen_w, screen_h)
            shops = self._validate_cards(shops, bs, screen_w, screen_h)
            rec["final"] = len(shops)
            rec["final_cards"] = [{"shop": s[0], "title": str(s[1])[:40], "price": s[2],
                                   "click": s[3]} for s in shops]
            rec["status"] = "ok(单侧空,采用另一侧)"
            self._log_cross(rec)
            return shops, "ok"

        b_unused = list(cards_b)
        agreement, a_only = [], []
        for ca in cards_a:
            hit = None
            for cb in b_unused:
                if self._same_card(ca, cb):
                    hit = cb
                    break
            if hit is not None:
                b_unused.remove(hit)
                agreement.append(ca)
            else:
                a_only.append(ca)
        b_only = b_unused
        rec["a_only"] = _compact(a_only)
        rec["b_only"] = _compact(b_only)
        print(f"[v1] 交叉: 两次一致 {len(agreement)} 张, 分歧 {len(a_only)}+{len(b_only)} 张")

        # 分歧仲裁(仅分歧时第3次调用; 仲裁失败则保守丢弃分歧卡)
        merged = list(agreement)
        if a_only or b_only:
            kept = self._arbitrate(ocr_text, a_only, b_only)
            rec["kept"] = _compact(kept)
            print(f"[v1] 仲裁保留分歧卡 {len(kept)} 张")
            merged += kept

        shops = self._basic_filter(merged, screen_w, screen_h)
        shops = self._validate_cards(shops, bs, screen_w, screen_h)
        rec["agreement"] = len(agreement)
        rec["final"] = len(shops)
        rec["final_cards"] = [{"shop": s[0], "title": str(s[1])[:40], "price": s[2],
                               "click": s[3]} for s in shops]
        rec["status"] = "ok"
        self._log_cross(rec)
        return shops, "ok"


VisionParser = VisionParserCross


# ── 独立测试 ───────────────────────────────────────────
if __name__ == "__main__":
    TEST_IMAGE = r""      # 留空 = 自动找最新 step3_b0.png
    TEST_KEYWORD = "甲巯咪唑片"
    TEST_CROP_X = 287

    import glob
    _proj = os.path.dirname(os.path.abspath(__file__))
    _shots = []
    if os.path.isdir(os.path.join(_proj, "screenshots")):
        for p in glob.glob(os.path.join(_proj, "screenshots", "**", "step3", "step3_b0.png"), recursive=True):
            _shots.append(p)
    _shots.sort(key=os.path.getmtime, reverse=True)
    if TEST_IMAGE.strip():
        _shots = [TEST_IMAGE.strip()]
    if not _shots:
        print("没找到截图"); sys.exit(1)

    from PIL import Image
    for img_path in _shots[:3]:
        print("=" * 60)
        print(f"图片: {img_path}")
        try:
            w, h = Image.open(img_path).size
        except Exception:
            w, h = 720, 1640
        try:
            shops, status = VisionParserCross().parse_shops(img_path, screen_size=(w, h),
                                                            keyword=TEST_KEYWORD, crop_x=TEST_CROP_X)
            print(f"  → status={status}, {len(shops)} 卡:")
            for s in shops:
                print(f"     {s[0]} | {s[1][:36]} | {s[2]} | click={s[3]}")
        except Exception as e:
            print(f"  异常: {e}")
        print()