spides/snapshot_jd.py

@@ -14,6 +14,7 @@ platform_name = "京东"
 class JdMain:
     def __init__(self):
         # self.db_online = MySQLPool39()
+        self.db_online = MySQLPoolOnline()
         self.crawl_count = ""
         self.task_id = ""
         self.task_dict = None
@@ -23,6 +24,69 @@ class JdMain:
         self.cumulative_stored = 0
         self.cumulative_skipped = 0
 
+    def get_status(self, status):
+        if status not in (2, 3, 4):
+            logger.warning(f"未知状态值: {status}, 跳过状态上报")
+            return
+        if status == 2:
+            parmas = {
+                "collect_task_allocate_id": self.task_id, "status": status, "finish_status": 0,
+                "start_time": int(time.time())
+            }
+        if status == 3:
+            parmas = {
+                "collect_task_allocate_id": self.task_id, "status": status, "finish_status": 1,
+                "real_count": self.crawl_count, "end_time": int(time.time()),
+            }
+        if status == 4:
+            parmas = {
+                "collect_task_allocate_id": self.task_id, "status": status, "finish_status": 0,
+                "end_time": int(time.time())}
+
+        # url = "http://scheduletest.dfwy.tech/api/collect_equipment_execute/result_report"
+        url = "http://scheduleapi.findit.ltd/api/collect_equipment_execute/result_report"
+
+        try:
+            res = requests.get(url, params=parmas, timeout=20)
+            res.raise_for_status()
+            logger.info("状态上报: %s", res.text)
+        except Exception as e:
+            logger.warning(f"状态上报失败: {e}")
+
+    def get_task(self):
+        """获取当前设备绑定的京东待执行快照任务。"""
+        sql = """
+            SELECT t.*
+            FROM `retrieve_collect_task_allocate` t
+            INNER JOIN `retrieve_collect_equipment_account` a
+                ON t.`collect_equipment_account_id` = a.`id`
+            WHERE t.`platform` = 2
+              AND t.`status` = 1
+              AND t.`snapshot_collect_status` = 0
+            LIMIT 1
+        """
+        task_list = self.db_online.select_data(sql)
+        print(task_list)
+        if not task_list:
+            return {}
+
+        task_dict = task_list[0]
+        self.task_id = task_dict["id"]
+        print(task_dict)
+        return task_dict
+
+    def heartbeat_task(self):
+        url = "https://scheduleapi.findit.ltd/api/collect_equipment_execute/heartbeat"
+
+        params = {
+            "collect_task_allocate_id": self.task_id,
+        }
+
+        try:
+            res = requests.get(url, params=params, timeout=20)
+            logger.info("心跳任务上报成功")
+        except Exception as e:
+            logger.info("心跳任务上报失败")
 
     def run(self):
 

spides/snapshot_jd.py.bak

@@ -1,123 +0,0 @@
-import json
-import time
-import requests
-from commons.Logger import logger
-from commons.conn_mysql import MySQLPoolOnline, MySQLPool39
-from spiders.jd.jd_auto_crawl_snap2 import JdCrawlerV2
-from commons.scheduler import CrawlerScheduler
-from commons.feishu_webhook import send_text
-import random
-from commons.config import JD_DEVICE_ID
-platform_name = "京东"
-
-
-class JdMain:
-    def __init__(self):
-        # self.db_online = MySQLPool39()
-        self.db_online = MySQLPoolOnline()
-        self.crawl_count = ""
-        self.task_id = ""
-        self.task_dict = None
-        self.driver = None
-        self.cumulative_pages = 0
-        self.cumulative_items = 0
-        self.cumulative_stored = 0
-        self.cumulative_skipped = 0
-
-    def get_status(self, status):
-        if status not in (2, 3, 4):
-            logger.warning(f"未知状态值: {status}, 跳过状态上报")
-            return
-        if status == 2:
-            parmas = {
-                "collect_task_allocate_id": self.task_id, "status": status, "finish_status": 0,
-                "start_time": int(time.time())
-            }
-        if status == 3:
-            parmas = {
-                "collect_task_allocate_id": self.task_id, "status": status, "finish_status": 1,
-                "real_count": self.crawl_count, "end_time": int(time.time()),
-            }
-        if status == 4:
-            parmas = {
-                "collect_task_allocate_id": self.task_id, "status": status, "finish_status": 0,
-                "end_time": int(time.time())}
-
-        # url = "http://scheduletest.dfwy.tech/api/collect_equipment_execute/result_report"
-        url = "http://scheduleapi.findit.ltd/api/collect_equipment_execute/result_report"
-
-        try:
-            res = requests.get(url, params=parmas, timeout=20)
-            res.raise_for_status()
-            logger.info("状态上报: %s", res.text)
-        except Exception as e:
-            logger.warning(f"状态上报失败: {e}")
-
-    def get_task(self):
-        """获取当前设备绑定的京东待执行快照任务。"""
-        sql = """
-            SELECT t.*
-            FROM `retrieve_collect_task_allocate` t
-            INNER JOIN `retrieve_collect_equipment_account` a
-                ON t.`collect_equipment_account_id` = a.`id`
-            WHERE t.`platform` = 2
-              AND t.`status` = 1
-              AND t.`snapshot_collect_status` = 0
-            LIMIT 1
-        """
-        task_list = self.db_online.select_data(sql)
-        print(task_list)
-        if not task_list:
-            return {}
-
-        task_dict = task_list[0]
-        self.task_id = task_dict["id"]
-        print(task_dict)
-        return task_dict
-
-    def heartbeat_task(self):
-        url = "https://scheduleapi.findit.ltd/api/collect_equipment_execute/heartbeat"
-
-        params = {
-            "collect_task_allocate_id": self.task_id,
-        }
-
-        try:
-            res = requests.get(url, params=params, timeout=20)
-            logger.info("心跳任务上报成功")
-        except Exception as e:
-            logger.info("心跳任务上报失败")
-
-    def run(self):
-
-        spider_schedule = CrawlerScheduler(JD_DEVICE_ID, 2)
-        spider_schedule.start()
-        time.sleep(3)
-
-        while 1:
-            if not spider_schedule.end:
-                self.task_dict = spider_schedule.get_task()
-
-                if not self.task_dict:
-                    logger.info(f"{platform_name}暂无任务")
-                    time.sleep(35)
-                    continue
-
-                self.task_id = self.task_dict.get("id", "")
-                self.crawl_count, is_success, self.driver, self.cumulative_pages, self.cumulative_items, self.cumulative_stored, self.cumulative_skipped = JdCrawlerV2(self.task_dict, spider_schedule, self.driver, self.cumulative_pages, self.cumulative_items, self.cumulative_stored, self.cumulative_skipped).run()
-                spider_schedule.stop()
-
-            else:
-                spider_schedule.start()
-                print('休息')
-                time.sleep(35)
-
-            time.sleep(30)
-
-
-if __name__ == '__main__':
-    while True:
-        JdMain().run()
-        interval_time = random.randint(1200, 1800)
-        logger.info(f"程序睡眠{interval_time}秒后继续执行")
-        time.sleep(interval_time)

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):

tbsg/yzm/nine_grid.py

@@ -0,0 +1,238 @@
+"""
+九宫格验证码完整流程
+① jfbym type 10 OCR雪花屏 → ② 截图九宫格 → ③ jfbym 30223 返回坐标 → ④ 点击
+"""
+import os, time, base64, json, random
+import cv2, numpy as np
+import uiautomator2 as u2
+import requests
+from PIL import Image
+
+# ===== 配置 =====
+DEVICE = "O7R4Y9CMPBPBU4VK"
+JFBYM_TOKEN = "1nDVocTE2mJ0yLEYb2sZJ5uUY2VIEoGTkIpW44X7Kgk"
+CAPTCHA_CROP = (42, 424, 1180, 575)           # 雪花屏裁剪
+GRID_CROP = (38, 662, 1185, 1817)              # 九宫格裁剪
+SHOT_COUNT = 20                                # 连拍张数
+AREA_THRESHOLD = 100                           # 连通域面积阈值
+JFBYM_URL = "https://api.jfbym.com/api/YmServer/customApi"
+OUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "image", "success")
+# =================
+
+def solve_snow_captcha():
+    """步骤①: 本地降噪 + jfbym type 10 OCR → 返回 (文字, extra, 设备)"""
+    d = u2.connect(DEVICE)
+
+    # 连拍 + 平均降噪
+    imgs = []
+    for _ in range(SHOT_COUNT):
+        full = d.screenshot(format='pillow')
+        imgs.append(np.array(full.crop(CAPTCHA_CROP).convert('L')))
+        time.sleep(0.05)
+
+    avg = np.mean(imgs, axis=0).astype(np.uint8)
+    t = np.percentile(avg, 15)
+    dark = np.where(avg < t, 0, 255).astype(np.uint8)
+
+    _, thresh = cv2.threshold(dark, 127, 255, cv2.THRESH_BINARY_INV)
+    num_labels, labels, stats, _ = cv2.connectedComponentsWithStats(thresh, connectivity=8)
+    clean = np.full_like(dark, 255)
+    for i in range(1, num_labels):
+        if stats[i, cv2.CC_STAT_AREA] > AREA_THRESHOLD:
+            clean[labels == i] = 0
+
+    os.makedirs(OUT_DIR, exist_ok=True)
+    cleaned_path = os.path.join(OUT_DIR, "cleaned.png")
+    cv2.imwrite(cleaned_path, clean)
+
+    # 发给 jfbym type 10 OCR
+    with open(cleaned_path, 'rb') as f:
+        b64 = base64.b64encode(f.read()).decode()
+
+    resp = requests.post(JFBYM_URL, json={
+        "token": JFBYM_TOKEN, "type": "10", "image": b64
+    }, headers={"Content-Type": "application/json"}, timeout=30).json()
+    print(f"  [jfbym-type10] 返回: {json.dumps(resp, ensure_ascii=False)[:200]}")
+
+    extra = resp.get("data", {}).get("data", {})
+    if isinstance(extra, list):
+        extra = extra[0] if extra else {}
+    if isinstance(extra, dict):
+        text = extra.get("tips", "")
+    elif isinstance(extra, str):
+        text = extra
+    else:
+        text = ""
+    print(f"[1] 降噪+OCR: {text}")
+    return text, extra, d
+
+
+def get_click_pos(d, extra):
+    """步骤②③: 截图九宫格 → jfbym 30223 → 返回坐标"""
+    grid_img = d.screenshot(format='pillow').crop(GRID_CROP)
+    grid_path = os.path.join(OUT_DIR, "grid.png")
+    grid_img.save(grid_path)
+    print(f"[2] 九宫格: {grid_img.size}")
+
+    with open(grid_path, 'rb') as f:
+        grid_b64 = base64.b64encode(f.read()).decode()
+
+    resp = requests.post(JFBYM_URL, json={
+        "token": JFBYM_TOKEN,
+        "type": "30223",
+        "image": grid_b64,
+        "extra": extra
+    }, headers={"Content-Type": "application/json"}, timeout=30).json()
+    print(f"  [jfbym-30223] 返回: {json.dumps(resp, ensure_ascii=False)[:200]}")
+
+    resp_data = resp.get("data", {})
+    if isinstance(resp_data, list):
+        data = resp_data[0] if resp_data else {}
+    else:
+        data = resp_data.get("data", {})
+    click_pos = data.get("click_pos", [])
+    tips = data.get("tips", "")
+    print(f"[3] tips={tips}, click_pos={click_pos}")
+    return click_pos, tips
+
+
+SUBMIT_BTN = (602, 2018)                        # 提交按钮坐标
+REFRESH_WAIT = 1.5                             # 点击后等待刷新秒数(加长)
+MAX_ROUNDS = 15                                 # 最大轮数, 防止死循环
+CLICK_OFFSET = 18                               # 随机偏移范围（±18px）
+GRID_W, GRID_H = GRID_CROP[2] - GRID_CROP[0], GRID_CROP[3] - GRID_CROP[1]  # 九宫格宽高
+STALL_LIMIT = 2                                 # 连续N轮候选完全不变→判定卡滞, 提前提交
+CAPTCHA_KEYWORDS = ("请依次点击", "根据提示", "没有新图片", "提交", "验证失败", "验证码错误")
+
+_ocr_eng = None
+
+
+def _captcha_still_present(d):
+    """截图 + OCR 检测九宫格验证码特征词是否仍在页面上"""
+    from rapidocr_onnxruntime import RapidOCR
+    global _ocr_eng
+    if _ocr_eng is None:
+        _ocr_eng = RapidOCR()
+    shot = d.screenshot(format='opencv')
+    if shot is None:
+        return None
+    r = _ocr_eng(shot)
+    if not r or not r[0]:
+        return False
+    texts = [item[1] for item in r[0]]
+    return any(any(kw in t for kw in CAPTCHA_KEYWORDS) for t in texts)
+
+
+def tap_loop(d, extra):
+    """步骤④: 每次只点一个→等待刷新→重新截图识别→直到无匹配→提交
+    返回 True/False: 提交后延迟一段时间再 OCR 复检验证码特征词是否消失, 以此判定真实成败
+    (而不是把"识别到的提示文字"当作成败信号——jfbym的tips有时是纯字符串, 会被误判为falsy)。"""
+    prev_sig = None
+    stall_count = 0
+    submitted = False
+
+    for round_num in range(1, MAX_ROUNDS + 1):
+        pos, tips = get_click_pos(d, extra)
+        if not pos:
+            print(f"[4] 第{round_num}轮无匹配，点击提交按钮")
+            d.click(SUBMIT_BTN[0] + random.randint(-8, 8), SUBMIT_BTN[1] + random.randint(-5, 5))
+            submitted = True
+            break
+
+        sig = (tips, tuple(sorted(pos)))
+        if sig == prev_sig:
+            stall_count += 1
+        else:
+            stall_count = 0
+        prev_sig = sig
+        if stall_count >= STALL_LIMIT:
+            print(f"[4] 连续{stall_count + 1}轮候选完全未变化，判定检测/点击卡滞，直接提交")
+            d.click(SUBMIT_BTN[0] + random.randint(-8, 8), SUBMIT_BTN[1] + random.randint(-5, 5))
+            submitted = True
+            break
+
+        # 只点第一个，点完重新截图
+        x, y = pos[0]
+        ox = x + random.randint(-CLICK_OFFSET, CLICK_OFFSET)
+        oy = y + random.randint(-CLICK_OFFSET, CLICK_OFFSET)
+        ox = max(5, min(GRID_W - 5, ox))
+        oy = max(5, min(GRID_H - 5, oy))
+        print(f"[4] 第{round_num}轮 共{len(pos)}个匹配，先点({x},{y})→偏移({ox},{oy})")
+        d.click(GRID_CROP[0] + ox, GRID_CROP[1] + oy)
+        time.sleep(REFRESH_WAIT)
+    else:
+        print(f"[4] 超过{MAX_ROUNDS}轮，直接提交")
+        d.click(SUBMIT_BTN[0] + random.randint(-8, 8), SUBMIT_BTN[1] + random.randint(-5, 5))
+        submitted = True
+
+    if not submitted:
+        return False
+
+    # 提交后不能立刻检测(页面还没刷新完成), 等一段随机间隔再 OCR 复检
+    time.sleep(random.uniform(1.8, 2.6))
+    still_present = _captcha_still_present(d)
+    if still_present is None:
+        print("[4] 提交后复检失败(截图/OCR异常)，保守判定为未通过")
+        return False
+    if still_present:
+        print("[4] 提交后复检: 验证码特征词仍在，判定未通过")
+        return False
+    print("[4] 提交后复检: 验证码特征词已消失，判定通过")
+    return True
+
+
+def solve(driver=None):
+    """九宫格验证码求解, 可传入外部driver或自动连接"""
+    if driver is not None:
+        d = driver
+    else:
+        d = u2.connect(DEVICE)
+
+    if not JFBYM_TOKEN:
+        raise ValueError("请先设置 JFBYM_TOKEN")
+    text, extra, _ = solve_snow_captcha_with_driver(d)
+    return tap_loop(d, extra)
+
+def solve_snow_captcha_with_driver(d):
+    """步骤①: 本地降噪 + jfbym type 10 OCR → 返回 (文字, extra, 设备)"""
+    # 连拍 + 平均降噪
+    imgs = []
+    for _ in range(SHOT_COUNT):
+        full = d.screenshot(format='pillow')
+        imgs.append(np.array(full.crop(CAPTCHA_CROP).convert('L')))
+        time.sleep(0.05)
+
+    avg = np.mean(imgs, axis=0).astype(np.uint8)
+    t = np.percentile(avg, 15)
+    dark = np.where(avg < t, 0, 255).astype(np.uint8)
+
+    _, thresh = cv2.threshold(dark, 127, 255, cv2.THRESH_BINARY_INV)
+    num_labels, labels, stats, _ = cv2.connectedComponentsWithStats(thresh, connectivity=8)
+    clean = np.full_like(dark, 255)
+    for i in range(1, num_labels):
+        if stats[i, cv2.CC_STAT_AREA] > AREA_THRESHOLD:
+            clean[labels == i] = 0
+
+    os.makedirs(OUT_DIR, exist_ok=True)
+    cleaned_path = os.path.join(OUT_DIR, "cleaned.png")
+    cv2.imwrite(cleaned_path, clean)
+
+    # 发给 jfbym type 10 OCR
+    with open(cleaned_path, 'rb') as f:
+        b64 = base64.b64encode(f.read()).decode()
+
+    resp = requests.post(JFBYM_URL, json={
+        "token": JFBYM_TOKEN, "type": "10", "image": b64
+    }, headers={"Content-Type": "application/json"}, timeout=30).json()
+    print(f"  [jfbym-type10] 返回: {json.dumps(resp, ensure_ascii=False)[:200]}")
+
+    extra = resp.get("data", {}).get("data", {})
+    if isinstance(extra, list):
+        extra = extra[0] if extra else {}
+    text = extra.get("tips", "") if isinstance(extra, dict) else ""
+    print(f"[1] 降噪+OCR: {text}")
+    return text, extra, d
+
+if __name__ == '__main__':
+    print(f"设备: {DEVICE}\n")
+    print(f"结果: {solve()}")