""" 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 "") shop = re.split(r'[（(]', shop)[0].strip() # 只要括号前的店名(去「(快递发…)」后缀) 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()