sg

tbsg/main.py

@@ -76,6 +76,8 @@ def _find_text_in_area(shot_path: str, target: str, max_y: int) -> Optional[dict
 
 def _screenshot(ex: SafeExecutor, name: str) -> str:
     import os
+    # 方案1：按设备 ID 隔离截图，避免多设备并发时写同一个文件
+    name = f"{ex.device_id}_{name}"
     SCREENSHOT_DIR.mkdir(exist_ok=True)
     path = str(SCREENSHOT_DIR / name)
     if os.path.exists(path):
@@ -87,16 +89,23 @@ def _screenshot(ex: SafeExecutor, name: str) -> str:
             shutil.copy2(path, backup)
         except Exception:
             pass
-    ex.driver.screenshot(path)
+    # 方案2：截图后验证完整性（adb 流式传输可能中断，导致 PNG 损坏）
+    for _try in range(3):
+        ex.driver.screenshot(path)
+        if cv2.imread(path) is not None:
+            break
+        time.sleep(0.5)
     return path
 
 
 def _is_search_page(ex: SafeExecutor) -> bool:
     """判断当前是否在搜索页面：只检测屏幕顶部20%区域内是否有「筛选」"""
-    import tempfile
+    import tempfile, os
     w, h = ex.driver.window_size()
-    tmp = str(SCREENSHOT_DIR / "_check_search.png")
+    tmp = str(SCREENSHOT_DIR / f"{ex.device_id}_check_search.png")
     ex.driver.screenshot(tmp)
+    if cv2.imread(tmp) is None:
+        ex.driver.screenshot(tmp)
     texts = OCR.recognize(tmp, rect=[0, 0, w, int(h * 0.2)], detail="text")
     return "筛选" in texts
 
@@ -317,7 +326,7 @@ def _visit_shop(ex: SafeExecutor, shop: list, visited: set) -> dict:
 def _handle_captcha(ex: SafeExecutor, ocr_texts: list) -> bool:
     """处理验证码, 重试5次, 失败等人工, 返回True=已解决"""
     import sys as _sys
-    _sys.path.insert(0, str(Path(__file__).parent / "yzm"))
+    _sys.path.insert(0, r"D:\drug\sg\yzm")
 
     for attempt in range(1, 6):
         print(f"    [验证码] 第{attempt}次尝试...")
@@ -348,9 +357,10 @@ def step4_parse_qr(ex: SafeExecutor, product_title: str, shop_name: str = "") ->
     返回 URL 或空字符串
     """
     # 安全的文件名前缀（用hash避免中文路径cv2兼容问题）
+    # 多设备隔离：加入设备ID，防止并发时两台设备写同一个文件
     import hashlib
     _hash = hashlib.md5(shop_name.encode()).hexdigest()[:8] if shop_name else "unknown"
-    _pfx = lambda name: str(SCREENSHOT_DIR / f"_s4_{_hash}_{name}")
+    _pfx = lambda name: str(SCREENSHOT_DIR / f"_s4_{ex.device_id}_{_hash}_{name}")
 
     time.sleep(6)
 
@@ -519,30 +529,31 @@ def step4_parse_qr(ex: SafeExecutor, product_title: str, shop_name: str = "") ->
             # 检测验证码页面
             captcha_kw = any("拖动滑块" in t or "请按住滑块" in t or "安全验证" in t for t in detail_texts)
             captcha_tpl = str(Path(__file__).parent / "files" / "captcha1.png")
-            tpl_match = False
-            if _os.path.exists(captcha_tpl):
-                si = cv2.imread(detail_check)
-                ti = cv2.imread(captcha_tpl)
-                if si is not None and ti is not None:
-                    gs = cv2.cvtColor(si, cv2.COLOR_BGR2GRAY)
-                    gt = cv2.cvtColor(ti, cv2.COLOR_BGR2GRAY)
-                    h_s, w_s = gs.shape
-                    crop_y1, crop_y2 = int(h_s * 0.25), int(h_s * 0.75)
-                    gs_crop = gs[crop_y1:crop_y2, 0:400]
-                    best_v = 0
-                    for fn, ss, tt in [
-                        ("gray", gs_crop, gt),
-                        ("edge", cv2.Canny(gs_crop,30,100), cv2.Canny(gt,30,100)),
-                        ("hist", cv2.equalizeHist(gs_crop), cv2.equalizeHist(gt)),
-                        ("blur", cv2.GaussianBlur(gs_crop,(3,3),0), cv2.GaussianBlur(gt,(3,3),0)),
-                        ("otsu", cv2.threshold(gs_crop,0,255,cv2.THRESH_BINARY+cv2.THRESH_OTSU)[1],
-                                cv2.threshold(gt,0,255,cv2.THRESH_BINARY+cv2.THRESH_OTSU)[1]),
-                    ]:
-                        if ss.ndim == 2 and tt.ndim == 2 and ss.shape[0] >= tt.shape[0] and ss.shape[1] >= tt.shape[1]:
-                            r = cv2.matchTemplate(ss, tt, cv2.TM_CCOEFF_NORMED)
-                            _, mv, _, _ = cv2.minMaxLoc(r)
-                            best_v = max(best_v, mv)
-                    tpl_match = best_v >= 0.30
+            # tpl_match = False
+            # if _os.path.exists(captcha_tpl):
+            #     si = cv2.imread(detail_check)
+            #     ti = cv2.imread(captcha_tpl)
+            #     if si is not None and ti is not None:
+            #         gs = cv2.cvtColor(si, cv2.COLOR_BGR2GRAY)
+            #         gt = cv2.cvtColor(ti, cv2.COLOR_BGR2GRAY)
+            #         h_s, w_s = gs.shape
+            #         crop_y1, crop_y2 = int(h_s * 0.25), int(h_s * 0.75)
+            #         gs_crop = gs[crop_y1:crop_y2, 0:400]
+            #         best_v = 0
+            #         for fn, ss, tt in [
+            #             ("gray", gs_crop, gt),
+            #             ("edge", cv2.Canny(gs_crop,30,100), cv2.Canny(gt,30,100)),
+            #             ("hist", cv2.equalizeHist(gs_crop), cv2.equalizeHist(gt)),
+            #             ("blur", cv2.GaussianBlur(gs_crop,(3,3),0), cv2.GaussianBlur(gt,(3,3),0)),
+            #             ("otsu", cv2.threshold(gs_crop,0,255,cv2.THRESH_BINARY+cv2.THRESH_OTSU)[1],
+            #                     cv2.threshold(gt,0,255,cv2.THRESH_BINARY+cv2.THRESH_OTSU)[1]),
+            #         ]:
+            #             if ss.ndim == 2 and tt.ndim == 2 and ss.shape[0] >= tt.shape[0] and ss.shape[1] >= tt.shape[1]:
+            #                 r = cv2.matchTemplate(ss, tt, cv2.TM_CCOEFF_NORMED)
+            #                 _, mv, _, _ = cv2.minMaxLoc(r)
+            #                 best_v = max(best_v, mv)
+            #         tpl_match = best_v >= 0.30
+            
             if captcha_kw or tpl_match:
                 print(f"    ⚠ 检测到验证码页面，尝试自动处理...")
                 if _handle_captcha(ex, detail_texts):