tbsg

tbsg/ai_helper.py

@@ -0,0 +1,336 @@
+"""
+AI 助手 — 调用 DeepSeek 分析 OCR 文本结果
+只负责：接收文本 → 调 AI → 返回结构化数据
+不负责 OCR 识别
+"""
+import json
+import os
+import re
+import time
+from pathlib import Path
+from typing import Optional
+
+import requests
+
+
+# ── 配置 ────────────────────────────────────────────────
+DEEPSEEK_API_URL = "https://ai-api.dfwy.tech/v1/messages"
+DEEPSEEK_MODEL = "deepseek-v4-flash-nothinking"
+
+# 项目根目录（用于读取 config.json）
+PROJECT_ROOT = Path(__file__).parent
+
+
+class AIParser:
+    """调用 DeepSeek 分析 OCR 文本，返回结构化店铺列表"""
+
+    # 默认 API Key（写死，也可通过环境变量或 config.json 覆盖）
+    _DEFAULT_KEY = ""
+
+    def __init__(self, api_key: Optional[str] = None):
+        """
+        Args:
+            api_key: DeepSeek API Key，不传则从环境变量或 config.json 读取
+        """
+        self.api_key = api_key or self._load_api_key()
+        if not self.api_key:
+            raise RuntimeError(
+                "未设置 DeepSeek API Key。请:\n"
+                "1. 设环境变量 DEEPSEEK_API_KEY\n"
+                "2. 或在 config.json 中添加 deepseek_api_key 字段\n"
+                "3. 或在初始化时传入 api_key"
+            )
+
+    @staticmethod
+    def _load_api_key() -> str:
+        """从环境变量、config.json 或默认值读取 API Key"""
+        key = os.environ.get("DEEPSEEK_API_KEY", "")
+        if key:
+            return key
+        cfg_path = PROJECT_ROOT / "config.json"
+        if cfg_path.exists():
+            try:
+                with open(cfg_path) as f:
+                    cfg = json.load(f)
+                key = cfg.get("deepseek_api_key", "")
+                if key:
+                    return key
+            except Exception:
+                pass
+        return AIParser._DEFAULT_KEY
+
+    # ── 核心：分析 OCR 结果 → 店铺列表 ──────────────────
+
+    def parse_shops(
+        self,
+        ocr_results: list,
+        screen_size: tuple = (1220, 2712),
+    ) -> list:
+        """
+        将 OCR 识别结果传给 DeepSeek，返回结构化店铺列表
+
+        Args:
+            ocr_results: OCR.recognize(detail="all") 返回值
+                [{"text": str, "confidence": float, "bbox": [[x1,y1],...]}, ...]
+            screen_size: (宽, 高)，默认 1220x2712
+
+        Returns:
+            [[店铺名, 商品标题, 价格, [点击x, 点击y], 配送距离y], ...]
+            点击坐标是店铺卡片中心点
+            配送距离y = 配送时间/距离文字的y坐标，用于滑动到其上方
+        """
+        # 1. 格式化 OCR 数据 — 按 y 坐标排序（从上到下）
+        sorted_results = sorted(ocr_results, key=lambda r: r["bbox"][0][1])
+
+        ocr_lines = []
+        for r in sorted_results:
+            box = r["bbox"]
+            cy = (box[0][1] + box[2][1]) // 2
+            cx = (box[0][0] + box[2][0]) // 2
+            ocr_lines.append(
+                f"[x={cx:4d}, y={cy:4d}, conf={r['confidence']:.2f}] {r['text']}"
+            )
+        ocr_text = "\n".join(ocr_lines)
+
+        screen_w, screen_h = screen_size
+
+        # 2. system prompt
+        system_prompt = """你是OCR文本结构化助手。从OCR文字列表中找出每个"店铺卡片"的4个字段，必须严格使用OCR原文。
+
+【OCR数据格式】
+每行: [x=坐标x, y=坐标y, conf=置信度] OCR识别出的文字
+y越大越靠下，文字按从上到下排列。
+
+【判断店铺卡片方法】
+- 同卡片的文字y坐标连续相近，卡片间有明显y间隔(>80px)
+- 卡片从上到下：商品标题(多行) → 价格(￥/¥开头) → 店铺名(以店/超市/旗舰店等结尾)
+
+【字段规则 — 严格使用OCR原文，不得修改】
+1. 店铺名 = 卡片最下方像店名的文字。
+   如果文字含"（"，只取"（"之前的部分（括号内是分店名，不重要）。
+   店铺名通常以"店/超市/旗舰店"结尾。
+2. 商品标题 = 卡片上方描述商品的文字。
+   多行时要完整拼接在一起，不要截断。
+3. 价格 = 卡片中间带￥或¥符号的文字。
+   只取**数值+货币符号**部分（如"￥12.8"、"¥67"），不要含"预估价"等额外文字。
+   如果找不到价格，留空字符串""。
+4. 点击坐标[x,y] = OCR数据中该卡片区域中间位置的实际坐标，用整数。
+
+【约束】
+- 所有字段值必须严格等于或取自OCR原文，不得修改、拼凑、推断
+- 店铺名如果找不到，留空字符串""
+- 找不到价格的卡片也要列出，价格留空
+- 有多少个卡片就输出多少个，不要遗漏
+- 输出纯JSON数组，不要任何多余文字
+
+返回格式:
+[[店铺名(去括号), 商品标题(完整), 价格(纯数字), [点击x, 点击y], 店铺名y], ...]
+店铺名y = 店铺名文字在OCR数据中的y坐标，必须从OCR数据中找"""
+
+        # 3. user prompt
+        user_prompt = (
+            f"屏幕OCR识别结果（屏幕尺寸 {screen_w}x{screen_h}）：\n\n"
+            f"{ocr_text}\n\n请按格式返回店铺列表。"
+        )
+
+        # 4. 调 AI
+        print(f"[ai] 分析 {len(ocr_results)} 个文本块...")
+        response = self._call(system_prompt, user_prompt)
+
+        # 5. 解析返回
+        return self._parse_response(response, screen_w, screen_h)
+
+    # ── 页面类型检测 ──────────────────────────────────
+
+    def check_page(self, ocr_results: list) -> dict:
+        """
+        分析进入商品页后的页面类型
+
+        Returns:
+            {"type": "normal"|"risk"|"qrcode"|"unknown",
+             "shop": "店铺名(如有)",
+             "detail": "额外描述"}
+        """
+        # 1. 格式化
+        sorted_r = sorted(ocr_results, key=lambda r: r["bbox"][0][1])
+        lines = []
+        for r in sorted_r:
+            box = r["bbox"]
+            cy = (box[0][1] + box[2][1]) // 2
+            cx = (box[0][0] + box[2][0]) // 2
+            lines.append(f"[x={cx:4d}, y={cy:4d}] {r['text']}")
+        ocr_text = "\n".join(lines)
+
+        system_prompt = """你是手机屏幕分析助手。收到OCR文字列表，判断当前屏幕是什么类型的页面。
+
+【页面类型】
+1. normal = 正常商品/店铺详情页（可以看到商品信息、价格、评价、店铺名等）
+2. risk = 风控/验证码/拦截页（包含"验证"、"拖动滑块"、"安全验证"、"禁止搜索"、"操作频繁"、"风险"、"请输入验证码"等）
+3. qrcode = 二维码/扫码页（包含"二维码"、"扫码"、"请用微信"等）
+4. unknown = 其他无法识别的页面
+
+**优先检测 risk 类型**：只要有类似验证码、风控拦截的文字，就返回risk
+
+【输出格式】
+只返回一个JSON对象，不要任何多余文字：
+{"type": "normal|risk|qrcode|unknown", "shop": "店铺名或空字符串", "detail": "简要说明"}
+"""
+        user_prompt = f"屏幕OCR文字：\n{ocr_text}\n\n请判断页面类型。"
+
+        print(f"[ai] 检测页面类型...")
+        response = self._call(system_prompt, user_prompt)
+
+        import json
+        cleaned = response.strip()
+        if cleaned.startswith("```"):
+            lines = cleaned.split("\n")
+            if lines[0].startswith("```"):
+                lines = lines[1:]
+            if lines and lines[-1].strip() == "```":
+                lines = lines[:-1]
+            cleaned = "\n".join(lines).strip()
+
+        try:
+            data = json.loads(cleaned)
+            return {
+                "type": data.get("type", "unknown"),
+                "shop": data.get("shop", ""),
+                "detail": data.get("detail", ""),
+            }
+        except json.JSONDecodeError:
+            if not response.strip():
+                return {"type": "normal", "shop": "", "detail": "API调用失败，默认继续"}
+            return {"type": "unknown", "shop": "", "detail": response[:100]}
+
+    # ── API 调用 ────────────────────────────────────────
+
+    def _call(self, system_prompt: str, user_content: str, timeout: int = 120, retries: int = 3) -> str:
+        """调用 API（Anthropic 格式），失败自动重试"""
+        headers = {
+            "x-api-key": self.api_key,
+            "anthropic-version": "2023-06-01",
+            "Content-Type": "application/json",
+        }
+        payload = {
+            "model": DEEPSEEK_MODEL,
+            "max_tokens": 8192,
+            "system": system_prompt,
+            "messages": [
+                {"role": "user", "content": user_content},
+            ],
+        }
+
+        last_err = None
+        for attempt in range(retries):
+            try:
+                resp = requests.post(DEEPSEEK_API_URL, json=payload, headers=headers, timeout=timeout)
+                resp.raise_for_status()
+                data = resp.json()
+                # 从 content 中提取所有 text 类型块
+                texts = [c["text"] for c in data.get("content", []) if c.get("type") == "text"]
+                return "".join(texts)
+            except Exception as e:
+                last_err = e
+                if attempt < retries - 1:
+                    print(f"    [ai] API调用失败(第{attempt+1}次)，重试... ({e})")
+                    time.sleep(2)
+        print(f"    [ai] API调用{retries}次均失败，跳过: {last_err}")
+        return ""
+
+    # ── 解析 ────────────────────────────────────────────
+
+    @staticmethod
+    def _parse_response(response: str, screen_w: int, screen_h: int) -> list:
+        """解析 AI 返回的 JSON"""
+        cleaned = response.strip()
+
+        # 去掉 markdown 代码块包裹
+        if cleaned.startswith("```"):
+            lines = cleaned.split("\n")
+            if lines[0].startswith("```"):
+                lines = lines[1:]
+            if lines and lines[-1].strip() == "```":
+                lines = lines[:-1]
+            cleaned = "\n".join(lines).strip()
+
+        # 解析 JSON
+        try:
+            data = json.loads(cleaned)
+        except json.JSONDecodeError:
+            print(f"[ai] WARNING: JSON 解析失败，尝试提取数组片段")
+            match = re.search(r'\[[\s\S]*\]', cleaned)
+            if match:
+                try:
+                    data = json.loads(match.group(0))
+                except json.JSONDecodeError:
+                    print(f"[ai] ERROR: JSON 解析失败")
+                    return []
+            else:
+                print(f"[ai] ERROR: 未找到 JSON 数组")
+                return []
+
+        if not isinstance(data, list):
+            print(f"[ai] ERROR: 返回不是数组")
+            return []
+
+        validated = []
+        for item in data:
+            # 兼容 dict 格式（模型偶发不按 list 模板输出时自动转换）
+            if isinstance(item, dict):
+                shop = str(item.get("name") or item.get("shop") or item.get("shop_name") or "")
+                title = str(item.get("title") or item.get("description") or item.get("product_name") or "")
+                price = str(item.get("price") or item.get("min_price") or "")
+                coords = item.get("coords") or item.get("click_pos") or item.get("click")
+                if not isinstance(coords, list) and item.get("x") is not None and item.get("y") is not None:
+                    coords = [item.get("x"), item.get("y")]
+                shop_name_y = item.get("shop_name_y") or item.get("y")
+                item = [shop, title, price, coords, shop_name_y]
+            if not isinstance(item, list) or len(item) < 4:
+                continue
+            shop = str(item[0]) if item[0] else ""
+            title = str(item[1]) if item[1] else ""
+            price = str(item[2]) if item[2] else ""
+            coords = item[3]
+            if isinstance(coords, list) and len(coords) == 2:
+                cx, cy = int(coords[0]), int(coords[1])
+                if 0 <= cx <= screen_w and 0 <= cy <= screen_h:
+                    # 第5个字段=店铺名文字的y坐标
+                    shop_name_y = int(item[4]) if len(item) >= 5 and isinstance(item[4], (int, float)) else min(cy + 200, screen_h)
+                    validated.append([shop, title, price, [cx, cy], shop_name_y])
+
+        print(f"[ai] 解析到 {len(validated)} 个店铺")
+        return validated
+
+
+# ── 独立测试 ────────────────────────────────────────────
+
+if __name__ == "__main__":
+    import sys
+    from steps.ocr import OCR
+
+    # 读取已有截图
+    test_img = str(PROJECT_ROOT / "screenshots" / "step2_result.png")
+    if not os.path.exists(test_img):
+        print(f"测试图片不存在: {test_img}")
+        print("请先跑一次 main.py 生成截图")
+        sys.exit(1)
+
+    # OCR 识别
+    ocr = OCR()
+    raw = ocr.recognize(test_img, detail="all")
+    print(f"OCR 识别到 {len(raw)} 个文本块\n")
+
+    # AI 分析
+    parser = AIParser()
+    shops = parser.parse_shops(raw)
+
+    print("\n" + "=" * 60)
+    print(f"  店铺列表 ({len(shops)} 个)")
+    print("=" * 60)
+    for s in shops:
+        print(f"  店铺: {s[0]}")
+        print(f"  商品: {s[1]}")
+        print(f"  价格: {s[2]}")
+        print(f"  点击: ({s[3][0]}, {s[3][1]})")
+        print(f"  配送y: {s[4]}")
+        print()

tbsg/detect_captcha_edge.py

@@ -0,0 +1,95 @@
+import cv2
+import numpy as np
+from PIL import Image
+import os
+import sys
+
+
+def detect_captcha_left_edge(image_path_or_array, output_path=None):
+    """
+    识别滑块验证码中橙色缺口覆盖层的最左边X坐标，并生成绿线标记图。
+    支持传入图片路径(str)或numpy.ndarray(BGR/RGB)。
+    """
+    if isinstance(image_path_or_array, str):
+        img = Image.open(image_path_or_array)
+        img_np = np.array(img)
+    else:
+        img_np = image_path_or_array.copy()
+        # OpenCV 截图通常是 BGR，转成 RGB
+        if img_np.shape[2] == 3:
+            img_np = cv2.cvtColor(img_np, cv2.COLOR_BGR2RGB)
+
+    h, w = img_np.shape[:2]
+
+    # 转换为HSV颜色空间
+    hsv = cv2.cvtColor(img_np, cv2.COLOR_RGB2HSV)
+
+    # 检测 peach/orange 覆盖层
+    # 色调: 10-35 (橙色/桃色), 饱和度: 40-255, 亮度: 140-255
+    lower = np.array([10, 40, 140])
+    upper = np.array([35, 255, 255])
+    mask = cv2.inRange(hsv, lower, upper)
+
+    # 形态学操作去噪
+    kernel = np.ones((5, 5), np.uint8)
+    mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel)
+    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel)
+
+    # 查找轮廓
+    contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+
+    if not contours:
+        raise ValueError("未检测到橙色区域")
+
+    # 筛选候选区域：
+    # 1. 轮廓中心位于验证码图片区域内（y + bh/2 在 0.4h ~ 0.65h 之间）
+    # 2. 面积足够大
+    # 3. y 范围足够长（bh > 0.2h），排除滑块按钮等短区域
+    candidates = []
+    for cnt in contours:
+        x, y, bw, bh = cv2.boundingRect(cnt)
+        area = bw * bh
+        if area < 5000:
+            continue
+        center_y = y + bh / 2
+        if center_y < h * 0.4 or center_y > h * 0.65:
+            continue
+        if bh < h * 0.2:
+            continue
+        candidates.append((cnt, x, y, bw, bh, area, bh))
+
+    if not candidates:
+        raise ValueError("未检测到拼图覆盖层")
+
+    # 选择 y 范围最长的候选
+    best = max(candidates, key=lambda c: c[6])
+    _, x, y, bw, bh, area, bh_max = best
+
+    # 左边缘 x 坐标
+    left_x = x
+
+    # 在原图上画绿线
+    canvas = img_np.copy()
+    cv2.line(canvas, (left_x, 0), (left_x, h), (0, 255, 0), 2)
+
+    # 保存绿线标记图
+    if output_path is None:
+        base, ext = os.path.splitext(image_path_or_array if isinstance(image_path_or_array, str) else "")
+        output_path = f"{base}_green.png"
+
+    cv2.imwrite(output_path, cv2.cvtColor(canvas, cv2.COLOR_RGB2BGR))
+
+    return left_x, output_path
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("用法: python detect_captcha_edge.py <图片路径> [输出路径]")
+        sys.exit(1)
+
+    image_path = sys.argv[1]
+    output_path = sys.argv[2] if len(sys.argv) > 2 else None
+
+    left_x, out = detect_captcha_left_edge(image_path, output_path)
+    print(f"拼图最左边X坐标: {left_x}")
+    print(f"绿线标记图: {out}")

tbsg/detect_slider_button.py

@@ -0,0 +1,112 @@
+import cv2
+import numpy as np
+from PIL import Image
+import os
+import sys
+
+
+def detect_slider_button(image_path, output_path=None):
+    """
+    识别滑块验证码底部拖动滑块按钮的位置，返回其四边坐标，并用红框标记。
+    改进版：只识别滑块轨道内的橙色小方块按钮，避免把轨道边框识别进去。
+
+    参数:
+        image_path: 输入图片路径
+        output_path: 输出图片路径，默认在原图目录下生成 <原图名>_slider_red.png
+
+    返回:
+        coords: dict, 包含 top_left, top_right, bottom_left, bottom_right
+        output_path: 红框标记图路径
+    """
+    img = Image.open(image_path)
+    img_np = np.array(img)
+    h, w = img_np.shape[:2]
+
+    # 只检测图片下半部分
+    start_y = int(h * 0.55)
+    lower_half = img_np[start_y:, :]
+
+    # 转换为HSV颜色空间
+    hsv = cv2.cvtColor(lower_half, cv2.COLOR_RGB2HSV)
+
+    # 检测橙色滑块按钮（严格范围，只保留高饱和高亮度的按钮内部）
+    lower = np.array([8, 230, 230])
+    upper = np.array([22, 255, 255])
+    mask = cv2.inRange(hsv, lower, upper)
+
+    # 形态学操作去噪
+    kernel = np.ones((5, 5), np.uint8)
+    mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel)
+    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel)
+
+    # 查找轮廓
+    contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+
+    if not contours:
+        raise ValueError("未检测到滑块按钮")
+
+    # 筛选轮廓：近似正方形、面积适中、位于轨道区域内
+    candidates = []
+    for cnt in contours:
+        x, y, bw, bh = cv2.boundingRect(cnt)
+        area = bw * bh
+        if area < 300 or bw < 30 or bh < 30:
+            continue
+        aspect_ratio = min(bw, bh) / max(bw, bh) if max(bw, bh) > 0 else 0
+        if aspect_ratio < 0.5:
+            continue
+        candidates.append((cnt, area, aspect_ratio, x, y, bw, bh))
+
+    if not candidates:
+        raise ValueError("未找到合适的滑块按钮")
+
+    # 选择最像滑块按钮的：面积适中、宽高比接近1
+    best = max(candidates, key=lambda c: c[1] * c[2])
+    _, _, _, x, y, bw, bh = best
+
+    # 转换回全图坐标
+    x = x
+    y = y + start_y
+
+    # 四边坐标
+    top_left = (x, y)
+    top_right = (x + bw, y)
+    bottom_left = (x, y + bh)
+    bottom_right = (x + bw, y + bh)
+
+    coords = {
+        "top_left": top_left,
+        "top_right": top_right,
+        "bottom_left": bottom_left,
+        "bottom_right": bottom_right,
+    }
+
+    # 在原图上画红框
+    canvas = img_np.copy()
+    cv2.rectangle(canvas, top_left, bottom_right, (255, 0, 0), 2)
+
+    # 保存红框标记图
+    if output_path is None:
+        base, ext = os.path.splitext(image_path)
+        output_path = f"{base}_slider_red.png"
+
+    cv2.imwrite(output_path, cv2.cvtColor(canvas, cv2.COLOR_RGB2BGR))
+
+    return coords, output_path
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("用法: python detect_slider_button.py <图片路径> [输出路径]")
+        sys.exit(1)
+
+    image_path = sys.argv[1]
+    output_path = sys.argv[2] if len(sys.argv) > 2 else None
+
+    coords, out = detect_slider_button(image_path, output_path)
+    print(f"滑块按钮四边坐标:")
+    print(f"  左上: {coords['top_left']}")
+    print(f"  右上: {coords['top_right']}")
+    print(f"  左下: {coords['bottom_left']}")
+    print(f"  右下: {coords['bottom_right']}")
+    print(f"红框标记图: {out}")

tbsg/main.py

@@ -0,0 +1,920 @@
+"""
+饿了么闪购 — 主入口  V1
+步骤驱动，按步执行
+"""
+import sys
+import time
+import re
+import cv2
+import numpy as np
+from pathlib import Path
+from typing import Optional
+
+sys.path.insert(0, str(Path(__file__).parent / "steps"))
+from ocr import OCR
+from executor import SafeExecutor
+from ai_helper import AIParser
+
+# ── 配置 ────────────────────────────────────────────────
+APP_PACKAGE = "me.ele"
+SCREENSHOT_DIR = Path(__file__).parent / "screenshots"
+OCR = OCR()
+
+
+def _find_device(device_id: str = "") -> str:
+    import subprocess
+    r = subprocess.run(["adb", "devices"], capture_output=True, text=True, timeout=5)
+    devices = []
+    for line in r.stdout.strip().split("\n")[1:]:
+        if line.strip() and "device" in line and "offline" not in line:
+            s = line.split("\t")[0].strip()
+            if s:
+                devices.append(s)
+
+    if not devices:
+        raise RuntimeError("未找到设备")
+
+    # 如果指定了设备ID，精确匹配
+    if device_id:
+        for d in devices:
+            if d == device_id:
+                return d
+        raise RuntimeError(f"未找到指定设备: {device_id}，可用设备: {devices}")
+
+    # 只有一台直接返回
+    if len(devices) == 1:
+        return devices[0]
+
+    # 多台设备：列出并让用户选择
+    print(f"\n发现 {len(devices)} 台设备:")
+    for i, d in enumerate(devices):
+        print(f"  [{i}] {d}")
+    while True:
+        try:
+            choice = input(f"请选择设备 [0-{len(devices)-1}]，回车默认第一台: ").strip()
+            if choice == "":
+                return devices[0]
+            idx = int(choice)
+            if 0 <= idx < len(devices):
+                return devices[idx]
+        except ValueError:
+            pass
+        print(f"输入无效，请输入 0-{len(devices)-1}")
+
+
+def _find_text_in_area(shot_path: str, target: str, max_y: int) -> Optional[dict]:
+    results = OCR.recognize(shot_path, detail="all")
+    for r in results:
+        if target in r["text"]:
+            y = r["bbox"][0][1]
+            if y < max_y:
+                cx = r["bbox"][0][0] + (r["bbox"][2][0] - r["bbox"][0][0]) // 2
+                cy = y + (r["bbox"][2][1] - y) // 2
+                return {"x": cx, "y": cy, "text": r["text"], "conf": r["confidence"]}
+    return None
+
+
+def _screenshot(ex: SafeExecutor, name: str) -> str:
+    import os
+    SCREENSHOT_DIR.mkdir(exist_ok=True)
+    path = str(SCREENSHOT_DIR / name)
+    if os.path.exists(path):
+        # 保留历史截图副本：固定文件名被 test/调试脚本引用，不能被后续运行覆盖丢失
+        import shutil
+        stem, ext = os.path.splitext(name)
+        backup = str(SCREENSHOT_DIR / f"{stem}_{int(time.time() * 1000)}{ext}")
+        try:
+            shutil.copy2(path, backup)
+        except Exception:
+            pass
+    ex.driver.screenshot(path)
+    return path
+
+
+def _is_search_page(ex: SafeExecutor) -> bool:
+    """判断当前是否在搜索页面：只检测屏幕顶部20%区域内是否有「筛选」"""
+    import tempfile
+    w, h = ex.driver.window_size()
+    tmp = str(SCREENSHOT_DIR / "_check_search.png")
+    ex.driver.screenshot(tmp)
+    texts = OCR.recognize(tmp, rect=[0, 0, w, int(h * 0.2)], detail="text")
+    return "筛选" in texts
+
+
+# ── 步骤 1：打开 App ────────────────────────────────────
+
+def step1_open_app(ex: SafeExecutor) -> bool:
+    print("=" * 40)
+    print("  步骤 1：打开饿了么闪购")
+    print("=" * 40)
+    w, h = ex.driver.window_size()
+    print(f"[step1] 屏幕尺寸: {w}x{h}")
+
+    print(f"[step1] 关闭 {APP_PACKAGE}...")
+    ex.driver.app_stop(APP_PACKAGE)
+    time.sleep(2)
+
+    print(f"[step1] 启动 {APP_PACKAGE}...")
+    ex.driver.app_start(APP_PACKAGE)
+    time.sleep(5)
+
+    shot = _screenshot(ex, "step1_home.png")
+    texts = OCR.recognize(shot, rect=[0, int(h * 0.88), w, h], detail="text")
+    print(f"[step1] 底部识别: {texts}")
+
+    for t in texts:
+        if "我的" in t:
+            print("[step1] OK - 成功进入 App")
+            return True
+    print("[step1] FAIL - 未检测到「我的」")
+    return False
+
+
+# ── 步骤 2：搜索商品 ────────────────────────────────────
+
+def step2_search(ex: SafeExecutor, keyword: str) -> bool:
+    print("\n" + "=" * 40)
+    print(f"  步骤 2：搜索「{keyword}」")
+    print("=" * 40)
+    w, h = ex.driver.window_size()
+    top_th = int(h * 0.3)
+
+    # ── 阶段0：点击「看病买药」 ──
+    shot0 = _screenshot(ex, "step2_phase0.png")
+    btn_med = _find_text_in_area(shot0, "看病买药", h)
+    if not btn_med:
+        print("[step2] FAIL - 未找到「看病买药」")
+        return False
+    print(f"[step2] 找到「看病买药」: ({btn_med['x']}, {btn_med['y']})")
+    ex.tap(btn_med["x"], btn_med["y"])
+    time.sleep(3)
+
+    # ── 阶段1：点击首页搜索栏 ──
+    shot = _screenshot(ex, "step2_phase1.png")
+    btn = _find_text_in_area(shot, "搜索", top_th)
+    if not btn:
+        print("[step2] FAIL - 未找到「搜索」")
+        return False
+    print(f"[step2] 找到「搜索」: ({btn['x']}, {btn['y']})")
+
+    cx = btn["x"] - 120  # 搜索左边约120px
+    cy = btn["y"]
+    ex.tap(cx, cy)
+    time.sleep(3)
+
+    shot2 = _screenshot(ex, "step2_phase2.png")
+    btn2 = _find_text_in_area(shot2, "搜索", top_th)
+    if not btn2:
+        print("[step2] FAIL - 进入搜索页后找不到「搜索」")
+        return False
+    moved = abs(btn2["x"] - btn["x"]) > 50 or abs(btn2["y"] - btn["y"]) > 50
+    if not moved:
+        print("[step2] FAIL - 搜索位置未改变")
+        return False
+    print(f"[step2] 搜索页搜索: ({btn2['x']}, {btn2['y']})")
+
+    cx2 = btn2["x"] - 180  # 搜索页输入框在搜索左边约180px
+    cy2 = btn2["y"]
+    ex.tap(cx2, cy2)
+    time.sleep(2)
+    print(f"[step2] 聚焦输入框，等待2s")
+
+    print(f"[step2] 输入关键词: {keyword}")
+    ex.driver.set_input_ime(True)
+    time.sleep(0.3)
+    ex.driver.send_keys(keyword)
+    time.sleep(1)
+
+    ex.tap(btn2["x"], btn2["y"])
+    time.sleep(3)
+
+    shot3 = _screenshot(ex, "step2_result.png")
+    raw3 = OCR.recognize(shot3, detail="all")
+    all_texts = [r["text"] for r in raw3]
+    has_filter = "筛选" in all_texts
+    has_express = "快递" in all_texts
+    kw_found = any(keyword in t for t in all_texts)
+
+    print(f"[step2] 有筛选: {has_filter}, 有快递: {has_express}, 关键词存在: {kw_found}")
+
+    # 如果有「快递」则点击它
+    if has_express:
+        for r in raw3:
+            if "快递" in r["text"]:
+                bx = r["bbox"]
+                cx = (bx[0][0] + bx[2][0]) // 2
+                cy = (bx[0][1] + bx[2][1]) // 2
+                print(f"[step2] 点击「快递」: ({cx}, {cy})")
+                ex.tap(cx, cy)
+                time.sleep(3)
+                break
+
+    if has_filter or has_express:
+        print("[step2] OK - 搜索成功")
+        return True
+
+    print("[step2] FAIL - 搜索未成功")
+    return False
+
+
+def _adb_swipe_up(ex: SafeExecutor, distance: int):
+    """ADB 手指从下往上滑，内容下滑"""
+    import subprocess
+    w, h = ex.driver.window_size()
+    swipe_x = w // 2
+    seg = 3
+    seg_px = distance // seg
+    for i in range(seg):
+        s = int(h * 0.8) - i * 80
+        e = s - seg_px
+        if e < 50:
+            e = 50
+        subprocess.run(
+            ["adb", "-s", ex.device_id, "shell", "input", "swipe",
+             str(swipe_x), str(s), str(swipe_x), str(e), "400"],
+            capture_output=True, timeout=10
+        )
+        time.sleep(0.35)
+    time.sleep(1.4)
+
+
+def _get_named_shops(ex: SafeExecutor, shot_name: str) -> list:
+    """截图 + OCR + AI → 返回有店铺名的列表"""
+    shot = _screenshot(ex, shot_name)
+    raw = OCR.recognize(shot, detail="all")
+    w, h = ex.driver.window_size()
+    parser = AIParser()
+    shops = parser.parse_shops(raw, screen_size=(w, h))
+    # 只保留有效店铺名+价格：店铺名必须含中文或字母（排除纯数字/标点/空格）
+    import re as _re
+    valid = []
+    for s in shops:
+        name = (s[0] or "").strip()
+        price = (s[2] or "").strip()
+        if name and _re.search(r'[一-鿿＀-￯a-zA-Z]', name) and price:
+            valid.append(s)
+    return valid
+
+
+def _shop_key(shop: list) -> str:
+    """用店铺名+价格去重（去括号内分店名、去尾部点号）"""
+    import re
+    name = shop[0]
+    price = shop[2] if len(shop) > 2 else ""
+    name = name.replace("(", "（").replace(")", "）")
+    name = re.sub(r'（.*', '', name)
+    name = re.sub(r'[.．…]+$', '', name)
+    return f"{name.strip()}|{price.strip()}"
+
+
+def _visit_shop(ex: SafeExecutor, shop: list, visited: set) -> dict:
+    """点击进入店铺 → step4 → 返回完整数据 dict"""
+    key = _shop_key(shop)
+    if key in visited:
+        return None
+
+    visited.add(key)
+    shop_name = shop[0]
+    product_title = shop[1]
+    price = shop[2]
+    click_x, click_y = shop[3]
+    print(f"  → 进入 [{shop_name}] 商品: {product_title[:30]}  价格: {price}")
+
+    ex.tap(click_x, click_y)
+
+    try:
+        qr_url = step4_parse_qr(ex, product_title, shop_name)
+    except Exception as e:
+        print(f"    ⚠ step4异常: {e}，跳过此店铺")
+        qr_url = ""
+
+    if qr_url == "__TERMINATE__":
+        print(f"    ⚠ 遇到终止信号，停止遍历")
+        return {"__terminate__": True}
+
+    if qr_url:
+        print(f"    ✅ QR: {qr_url[:80]}")
+        print(f"    📦 采集完成: {shop_name} | {product_title[:30]} | {price} | {qr_url[:60]}")
+    else:
+        print(f"    ⚠ 未获取到二维码链接")
+        print(f"    📦 采集完成(无链接): {shop_name} | {product_title[:30]} | {price}")
+
+    # 返回搜索页：最多退3次，每次检测顶部区域是否有「筛选」
+    for _ in range(3):
+        ex.driver.press("back")
+        time.sleep(1.4)
+        if _is_search_page(ex):
+            break
+
+    return {
+        "shop": shop_name,
+        "title": product_title,
+        "price": price,
+        "link": qr_url or "",
+    }
+
+
+def _handle_captcha(ex: SafeExecutor, ocr_texts: list) -> bool:
+    """处理验证码, 重试5次, 失败等人工, 返回True=已解决"""
+    import sys as _sys
+    _sys.path.insert(0, str(Path(__file__).parent / "yzm"))
+
+    for attempt in range(1, 6):
+        print(f"    [验证码] 第{attempt}次尝试...")
+        nine_kw = any("提交" in t or "没有新图片" in t for t in ocr_texts)
+        if nine_kw:
+            from nine_grid import solve as solve_nine
+            ok = solve_nine(ex.driver)
+        else:
+            from tmp_captcha_test2 import solve_slider
+            ok = solve_slider(ex.driver)
+        if ok:
+            print(f"    ✅ 验证码已解决")
+            return True
+        print(f"    ❌ 第{attempt}次失败")
+        time.sleep(1)
+
+    print(f"    ⚠ 5次自动处理失败, 请人工处理...")
+    input("    处理完成后按回车继续...")
+    return True
+
+
+def step4_parse_qr(ex: SafeExecutor, product_title: str, shop_name: str = "") -> str:
+    """
+    1. 等待加载 → OCR → AI找商品标题坐标
+    2. 点击商品标题 → 进入商品详情
+    3. 找右上角"分享" → 点击 → 二维码弹窗
+    4. 截图 → pyzbar 解析二维码
+    返回 URL 或空字符串
+    """
+    # 安全的文件名前缀（用hash避免中文路径cv2兼容问题）
+    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}")
+
+    time.sleep(6)
+
+    # ── 检测页面类型：验证码/风控/正常（unknown/qrcode 重试3次）──
+    for page_retry in range(3):
+        shot_check = _pfx("page_check.png")
+        ex.driver.screenshot(shot_check)
+        check_raw = OCR.recognize(shot_check, detail="all")
+
+        # 方法A: 模板匹配检测验证码
+        import os as _os
+        captcha_tpl = str(Path(__file__).parent / "files" / "captcha1.png")
+        if _os.path.exists(captcha_tpl):
+            si = cv2.imread(shot_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)
+                crop_x1, crop_x2 = 0, 400
+                gs_crop = gs[crop_y1:crop_y2, crop_x1:crop_x2]
+                scores = []
+                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)
+                        scores.append((mv, fn))
+                if scores:
+                    best_v = max(s[0] for s in scores)
+                    best_m = max(scores, key=lambda s: s[0])[1]
+                    print(f"    验证码模板匹配: {best_m}={best_v:.3f}")
+                    captcha_kw = any("拖动滑块" in r["text"] or "请按住滑块" in r["text"] or "安全验证" in r["text"] for r in check_raw)
+                    nine_kw = any("提交" in r["text"] or "没有新图片" in r["text"] for r in check_raw)
+                    if best_v >= 0.30 and (captcha_kw or nine_kw):
+                        print(f"    ⚠ 检测到验证码，尝试自动处理...")
+                        if _handle_captcha(ex, [r["text"] for r in check_raw]):
+                            continue
+                        return "__TERMINATE__"
+                    elif best_v >= 0.30 and not captcha_kw:
+                        print(f"    ⚠ 模板匹配命中但OCR无验证码关键词，忽略")
+        page_type = AIParser().check_page(check_raw)
+        ptype = page_type.get("type", "unknown")
+
+        if ptype == "risk":
+            print(f"    ⚠ AI检测到验证码，尝试自动处理...")
+            if _handle_captcha(ex, [r["text"] for r in check_raw]):
+                continue
+            return "__TERMINATE__"
+
+        if ptype == "normal":
+            break  # 正常，跳出重试循环
+
+        # qrcode 或 unknown → 可能未加载完成
+        if page_retry < 2:
+            print(f"    检测到{ptype}页面，可能未加载完成，第{page_retry+1}次重试...")
+            time.sleep(2)
+    else:
+        # 3次重试后仍异常
+        import shutil
+        err_dir = SCREENSHOT_DIR / "unrecognized"
+        err_dir.mkdir(exist_ok=True)
+        shutil.copy(shot_check, str(err_dir / f"{ptype}_{int(time.time())}.png"))
+        print(f"    ⚠ 3次检测均为{ptype}页面，终止程序")
+        return "__TERMINATE__"
+    # normal → 继续
+
+    # ── 店铺页判断：OCR同时检测到「刚刚搜过」和「评价」说明在店铺页 ──
+    in_shop = False
+    for _ in range(10):
+        shop_check = _pfx("shop_check.png")
+        ex.driver.screenshot(shop_check)
+        shop_raw = OCR.recognize(shop_check, detail="text")
+        has_ganggang = any("刚刚搜过" in t for t in shop_raw)
+        has_pingjia = any("评价" in t for t in shop_raw)
+        if has_ganggang and has_pingjia:
+            in_shop = True
+            print(f"    已确认在店铺页")
+            break
+        time.sleep(1)
+    if not in_shop:
+        print(f"    ⚠ 未检测到店铺页，继续尝试...")
+
+    # ── 第1步：截图 + AI找商品标题坐标 ──
+    shot = _pfx("shop.png")
+    ex.driver.screenshot(shot)
+    raw = OCR.recognize(shot, detail="all")
+
+    sorted_r = sorted(raw, key=lambda r: r["bbox"][0][1])
+    lines = []
+    for r in sorted_r:
+        cx = (r["bbox"][0][0] + r["bbox"][2][0]) // 2
+        cy = (r["bbox"][0][1] + r["bbox"][2][1]) // 2
+        lines.append(f"[x={cx:4d}, y={cy:4d}] {r['text']}")
+    ocr_text = "\n".join(lines)
+
+    system_prompt = """你收到店铺页的OCR文字。商品标题文字坐标已知(从OCR中有x,y)。
+请找到和以下商品标题匹配的文字块，返回其点击坐标。
+
+【重要规则】
+- 坐标必须从OCR数据中选取，不得编造或估算
+- 如果找不到完全匹配的，找最相似的
+- 如果完全找不到，返回null
+
+只返回JSON：
+{"title_xy": [x, y] 或 null, "shop": "店铺名"}"""
+
+    parser = AIParser()
+    resp = parser._call(system_prompt, f"商品标题: {product_title}\n\nOCR文字：\n{ocr_text}\n\n请返回商品标题坐标。")
+
+    import json
+    cleaned = resp.strip()
+    if cleaned.startswith("```"):
+        cl = cleaned.split("\n")
+        if cl[0].startswith("```"): cl = cl[1:]
+        if cl and cl[-1].strip() == "```": cl = cl[:-1]
+        cleaned = "\n".join(cl).strip()
+    try:
+        data = json.loads(cleaned)
+        title_xy = data.get("title_xy")
+    except json.JSONDecodeError:
+        title_xy = None
+
+    if not title_xy or not isinstance(title_xy, list) or len(title_xy) != 2:
+        print(f"    ⚠ AI未返回有效坐标: {title_xy}")
+        return ""
+
+    tx, ty = title_xy
+    if tx is None or ty is None:
+        print(f"    ⚠ AI返回空坐标")
+        return ""
+
+    tx, ty = int(tx), int(ty)
+    w, h = ex.driver.window_size()
+    if not (0 <= tx <= w and 0 <= ty <= h):
+        print(f"    ⚠ 坐标越界: ({tx},{ty}) 超出屏幕 {w}x{h}")
+        return ""
+
+    # ── 第2步：点击商品标题 → 进入商品详情（最多重试3次）──
+    entered_detail = False
+    for attempt in range(3):
+        print(f"    点击商品标题: ({tx},{ty}) (第{attempt+1}次)")
+        ex.tap(tx, ty)
+
+        # 检测是否进入商品详情页
+        for _ in range(5):
+            time.sleep(2)
+            detail_check = _pfx("detail_check.png")
+            ex.driver.screenshot(detail_check)
+            detail_raw = OCR.recognize(detail_check, detail="all")
+            detail_texts = [r["text"] for r in detail_raw]
+
+            # 检测商品详情页关键词
+            if any("加入购物车" in t or "立即购买" in t or "选规格" in t or "商品详情页" in t for t in detail_texts):
+                print(f"    已进入商品详情页")
+                entered_detail = True
+                break
+
+            # 检测验证码页面
+            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
+            if captcha_kw or tpl_match:
+                print(f"    ⚠ 检测到验证码页面，尝试自动处理...")
+                if _handle_captcha(ex, detail_texts):
+                    continue
+                return "__TERMINATE__"
+
+            # 不在详情页，检测是否还在店铺页
+            has_ganggang = any("刚刚搜过" in t for t in detail_texts)
+            has_pingjia = any("评价" in t for t in detail_texts)
+            if has_ganggang and has_pingjia:
+                print(f"    仍在店铺页，重试...")
+                break  # 跳出内层循环
+
+        if entered_detail:
+            break
+
+        # 第1次失败后，重新OCR+AI获取坐标（可能是页面滚动导致坐标偏移）
+        if attempt < 2:
+            print(f"    重新OCR获取坐标...")
+            re_shot = _pfx("shop.png")
+            ex.driver.screenshot(re_shot)
+            re_raw = OCR.recognize(re_shot, detail="all")
+            re_sorted = sorted(re_raw, key=lambda r: r["bbox"][0][1])
+            re_lines = []
+            for r in re_sorted:
+                r_cx = (r["bbox"][0][0] + r["bbox"][2][0]) // 2
+                r_cy = (r["bbox"][0][1] + r["bbox"][2][1]) // 2
+                re_lines.append(f"[x={r_cx:4d}, y={r_cy:4d}] {r['text']}")
+            re_ocr_text = "\n".join(re_lines)
+            re_resp = parser._call(system_prompt, f"商品标题: {product_title}\n\nOCR文字：\n{re_ocr_text}\n\n请返回商品标题坐标。")
+            re_cleaned = re_resp.strip()
+            if re_cleaned.startswith("```"):
+                rl = re_cleaned.split("\n")
+                if rl[0].startswith("```"): rl = rl[1:]
+                if rl and rl[-1].strip() == "```": rl = rl[:-1]
+                re_cleaned = "\n".join(rl).strip()
+            try:
+                re_data = json.loads(re_cleaned)
+                re_xy = re_data.get("title_xy")
+                if re_xy and isinstance(re_xy, list) and len(re_xy) == 2 and re_xy[0] is not None:
+                    tx, ty = int(re_xy[0]), int(re_xy[1])
+                    ww, hh = ex.driver.window_size()
+                    if not (0 <= tx <= ww and 0 <= ty <= hh):
+                        print(f"    ⚠ 新坐标越界: ({tx},{ty})，保持原坐标")
+                    else:
+                        print(f"    新坐标: ({tx},{ty})")
+            except Exception:
+                pass
+    else:
+        pass  # 3次重试结束
+
+    if not entered_detail:
+        print(f"    ⚠ 3次点击未进入商品详情页，跳过")
+        return ""
+
+    # ── 第3步：ORB特征匹配找分享图标 ──
+    share_shot = _pfx("find_share.png")
+    ex.driver.screenshot(share_shot)
+    screen = cv2.imread(share_shot)
+    template_path = str(Path(__file__).parent / "files" / "share.png")
+    template = cv2.imread(template_path)
+    sx, sy = None, None
+
+    if screen is not None and template is not None:
+        h_s, w_s = screen.shape[:2]
+
+        # 右上角区域（分享图标永远在右上）
+        roi_x1, roi_y1 = w_s * 2 // 3, 0
+        roi = screen[roi_y1:h_s // 4, roi_x1:w_s]
+
+        # 方法A: SIFT 特征匹配（限制右上角区域，减少干扰）
+        sx, sy = None, None
+        sift = cv2.SIFT_create(nfeatures=1500)
+        kp1, des1 = sift.detectAndCompute(template, None)
+        kp2, des2 = sift.detectAndCompute(roi, None)
+
+        if des1 is not None and des2 is not None and len(kp1) >= 2 and len(kp2) >= 2:
+            bf = cv2.BFMatcher()
+            matches = bf.knnMatch(des1, des2, k=2)
+            good = []
+            for m, n in matches:
+                if m.distance < 0.75 * n.distance:
+                    good.append(m)
+            print(f"    分享SIFT(右上区域): 模板{len(kp1)}特征 ROI{len(kp2)}特征 优质{len(good)}")
+            if len(good) >= 4:
+                src_pts = np.float32([kp1[m.queryIdx].pt for m in good]).reshape(-1, 1, 2)
+                dst_pts = np.float32([kp2[m.trainIdx].pt for m in good]).reshape(-1, 1, 2)
+                matrix, _ = cv2.findHomography(src_pts, dst_pts, cv2.RANSAC, 5.0)
+                if matrix is not None:
+                    h_t, w_t = template.shape[:2]
+                    corners = np.float32([[0, 0], [w_t, 0], [w_t, h_t], [0, h_t]]).reshape(-1, 1, 2)
+                    transformed = cv2.perspectiveTransform(corners, matrix)
+                    sx = roi_x1 + int(np.mean(transformed[:, 0, 0]))
+                    sy = int(np.mean(transformed[:, 0, 1]))
+                    print(f"    分享SIFT匹配: ({sx},{sy})")
+
+        # 方法B: 多尺度模板匹配（右上角区域）
+        if sx is None:
+            best_val, best_loc, best_sw, best_sh = 0, None, 0, 0
+            for scale in [0.7, 0.8, 0.9, 1.0, 1.1, 1.2, 1.3]:
+                scaled = cv2.resize(template, None, fx=scale, fy=scale)
+                sw, sh = scaled.shape[1], scaled.shape[0]
+                if sh > roi.shape[0] or sw > roi.shape[1]:
+                    continue
+                res = cv2.matchTemplate(roi, scaled, cv2.TM_CCOEFF_NORMED)
+                _, mv, _, ml = cv2.minMaxLoc(res)
+                if mv > best_val:
+                    best_val, best_loc, best_sw, best_sh = mv, ml, sw, sh
+                t_edge = cv2.Canny(scaled, 30, 100)
+                r_edge = cv2.Canny(roi, 30, 100)
+                if t_edge.shape[0] <= r_edge.shape[0] and t_edge.shape[1] <= r_edge.shape[1]:
+                    res2 = cv2.matchTemplate(r_edge, t_edge, cv2.TM_CCOEFF_NORMED)
+                    _, mv2, _, ml2 = cv2.minMaxLoc(res2)
+                    if mv2 > best_val:
+                        best_val, best_loc, best_sw, best_sh = mv2, ml2, sw, sh
+            print(f"    分享模板匹配(右上): 最佳={best_val:.3f}")
+            if best_val >= 0.26 and best_loc is not None:
+                sx = roi_x1 + best_loc[0] + best_sw // 2
+                sy = best_loc[1] + best_sh // 2
+
+        if sx is not None and sy is not None:
+            print(f"    分享图标: ({sx},{sy})")
+            ex.tap(sx, sy)
+        else:
+            print(f"    ⚠ 未找到分享图标")
+            return ""
+
+    # 等弹窗出现，同时记录"分享到"y坐标用于QR裁剪
+    share_y = None
+    waimai_y = None
+    for _ in range(8):
+        time.sleep(1)
+        ck = _pfx("share_popup.png")
+        ex.driver.screenshot(ck)
+        detail = OCR.recognize(ck, detail="all")
+        texts = [r["text"] for r in detail]
+        if any("分享到" in t for t in texts):
+            print(f"    分享弹窗出现")
+            # 记录"分享到"和"外卖"的y坐标
+            for r in detail:
+                cy = (r["bbox"][0][1] + r["bbox"][2][1]) // 2
+                if "分享到" in r["text"] and share_y is None:
+                    share_y = cy
+                if "外卖" in r["text"] and waimai_y is None:
+                    waimai_y = cy
+            break
+    if share_y is None:
+        share_y = int(ex.driver.window_size()[1] * 0.74)  # fallback
+
+    # ── 第4步：截图 → 多方法解析二维码（多次重试） ──
+    def _decode_qr(img, share_y):
+        """基于OCR定位的share_y裁剪QR区域解析"""
+        if img is None: return ""
+        h, w = img.shape[:2]
+        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+        detector = cv2.QRCodeDetector()
+
+        # 裁剪区域：y从"外卖"(或估计值)到"分享到", x从60%到92%
+        y_top = waimai_y if waimai_y else max(0, share_y - 280)
+        y_bot = share_y
+        x_l, x_r = int(w * 0.60), int(w * 0.92)
+        crop_save = img[y_top:y_bot, x_l:x_r]
+        cv2.imwrite(_pfx("qr_crop.png"), crop_save)
+
+        def _try_decode(roi_gray, zooms=(1,)):
+            """在灰度图上尝试多种方式解码"""
+            if roi_gray is None or roi_gray.size == 0 or roi_gray.shape[0] == 0 or roi_gray.shape[1] == 0:
+                return ""
+            for z in zooms:
+                if z > 1:
+                    big = cv2.resize(roi_gray, None, fx=z, fy=z, interpolation=cv2.INTER_NEAREST)
+                else:
+                    big = roi_gray
+                data, _, _ = detector.detectAndDecode(big)
+                if data: return data
+            # OTSU + zoom
+            for z in zooms:
+                _, th = cv2.threshold(roi_gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
+                big = cv2.resize(th, None, fx=z, fy=z, interpolation=cv2.INTER_NEAREST) if z > 1 else th
+                data, _, _ = detector.detectAndDecode(big)
+                if data: return data
+            return ""
+
+        # 方法A: 全图detect定位QR → 裁200x200
+        ok, points = detector.detect(gray)
+        if ok and points is not None and len(points) > 0:
+            pts = points[0].astype(int)
+            cx = int(np.mean(pts[:, 0]))
+            cy = int(np.mean(pts[:, 1]))
+            x1, y1 = max(0, cx - 100), max(0, cy - 100)
+            x2, y2 = min(w, cx + 100), min(h, cy + 100)
+            if x2 > x1 and y2 > y1:
+                data = _try_decode(gray[y1:y2, x1:x2], (1, 2, 3))
+                if data: return data
+        # 方法B: 200x200滑动窗口扫描（基于OCR定位区域）
+        scan_area = gray[y_top:y_bot, x_l:x_r]
+        sh, sw = scan_area.shape
+        step = min(80, max(40, sh // 3, sw // 3))
+        for y in range(0, max(1, sh - 200), step):
+            for x in range(0, max(1, sw - 200), step):
+                patch = scan_area[y:y+200, x:x+200]
+                data = _try_decode(patch, (1, 2))
+                if data: return data
+
+        # 方法C: 固定区域 fallback（基于OCR定位）
+        crop = gray[y_top:y_bot, x_l:x_r]
+        data = _try_decode(crop, (1, 2, 3, 4))
+        if data: return data
+
+        # 方法D: 全图兜底（裁剪失败时直接在全图上尝试）
+        data = _try_decode(gray, (1, 2, 3))
+        if data: return data
+
+        # 方法E: 全图 OTSU + 放大
+        _, full_th = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
+        for z in (1, 2, 3):
+            big = cv2.resize(full_th, None, fx=z, fy=z, interpolation=cv2.INTER_CUBIC) if z > 1 else full_th
+            data, _, _ = detector.detectAndDecode(big)
+            if data: return data
+
+        return ""
+
+    for retry in range(6):  # 最多等 5+2*5=15秒
+        time.sleep(5 if retry == 0 else 2)
+        qr_shot = _pfx("qr.png")
+        ex.driver.screenshot(qr_shot)
+        data = _decode_qr(cv2.imread(qr_shot), share_y)
+        if data:
+            print(f"    QR链接: {data[:100]}")
+            return data
+
+    return ""
+
+
+# ── 步骤 3：滑动 + 逐个点击店铺 ──────────────────────
+
+def step3_swipe_and_enter(ex: SafeExecutor, keyword: str) -> list:
+    """
+    截图 → AI分析 → 逐个点击全部可见店铺 → 下滑加载更多 → 继续点击 → 直到全部遍历
+    """
+    print("\n" + "=" * 40)
+    print("  步骤 3：遍历店铺")
+    print("=" * 40)
+
+    visited = set()
+    w, h = ex.driver.window_size()
+    batch_no = 0
+    empty_streak = 0  # 连续没有新店铺的批次数
+    all_results = []
+
+    while True:
+        named = _get_named_shops(ex, f"step3_b{batch_no}.png")
+
+        new_ones = [s for s in named if _shop_key(s) not in visited]
+
+        print(f"[step3] 批次{batch_no}: 共{len(named)}个, 新{len(new_ones)}个")
+
+        if not new_ones:
+            empty_streak += 1
+            print(f"[step3] 无新店铺 (连续{empty_streak}/3)")
+            if empty_streak >= 3:
+                print(f"[step3] 连续3批无新店铺，结束")
+                break
+            # 滑动后再试
+            print(f"[step3] 滑动查看下一批")
+            if len(named) >= 2:
+                target_y = named[-2][4]
+                swipe_dist = target_y - int(h * 0.15)
+                if swipe_dist > 0:
+                    _adb_swipe_up(ex, swipe_dist)
+                else:
+                    _adb_swipe_up(ex, int(h * 0.15))
+            else:
+                _adb_swipe_up(ex, int(h * 0.3))
+            time.sleep(2)
+            batch_no += 1
+            continue
+
+        empty_streak = 0  # 有新店铺，重置计数
+
+        for shop in new_ones:
+            result = _visit_shop(ex, shop, visited)
+            if result and result.get("__terminate__"):
+                print("[step3] 收到终止信号，停止遍历")
+                all_results = [r for r in all_results if not r.get("__terminate__")]
+                break
+            if result:
+                all_results.append(result)
+        else:
+            # for 正常结束 → 滑动到倒数第二个卡片的配送距离位置
+            print(f"[step3] 已访问 {len(visited)} 个，滑动查看下一批")
+            if len(named) >= 2:
+                target_y = named[-2][4]  # 倒数第二个卡片的配送距离y坐标
+                swipe_dist = target_y - int(h * 0.15)
+                if swipe_dist > 0:
+                    _adb_swipe_up(ex, swipe_dist)
+                else:
+                    _adb_swipe_up(ex, int(h * 0.15))
+            else:
+                _adb_swipe_up(ex, int(h * 0.3))
+            time.sleep(2)
+            batch_no += 1
+            continue
+        # break 出来 → 结束
+        break
+
+    # ── 输出最终结果表 ──
+    print("\n" + "=" * 70)
+    print(f"  最终结果 ({len(all_results)} 个店铺)")
+    print("=" * 70)
+    for i, r in enumerate(all_results, 1):
+        link_short = r["link"][:55] + "..." if len(r["link"]) > 55 else r["link"]
+        print(f"  [{i}] {r['shop']}")
+        print(f"      商品: {r['title'][:40]}")
+        print(f"      价格: {r['price']}")
+        print(f"      链接: {link_short}")
+        print()
+
+    return all_results
+
+
+# ── 步骤 4：（保留，当前为空 ──────────────────────────
+
+def step4_empty(ex: SafeExecutor):
+    """占位，供后续扩展"""
+    pass
+
+
+# ── 主入口 ──────────────────────────────────────────────
+
+if __name__ == "__main__":
+    args = sys.argv[1:]
+    device_id = "T4VK4LM7AAUOV8AY"
+    # 解析 --device 参数
+    filtered = []
+    i = 0
+    while i < len(args):
+        if args[i] == "--device" and i + 1 < len(args):
+            device_id = args[i + 1]
+            i += 2
+        else:
+            filtered.append(args[i])
+            i += 1
+
+    cmd = filtered[0] if filtered else "all"
+    keyword = filtered[1] if len(filtered) > 1 else "矿泉水"
+
+    print("设备连接中...")
+    device_id = _find_device(device_id)
+    print(f"设备: {device_id}")
+    ex = SafeExecutor(device_id)
+
+    if cmd in ("all", "step1"):
+        ok = step1_open_app(ex)
+        if not ok:
+            sys.exit(1)
+
+    if cmd in ("all", "step2"):
+        ok = step2_search(ex, keyword)
+        if not ok:
+            sys.exit(1)
+
+    if cmd in ("all", "step3"):
+        visited = step3_swipe_and_enter(ex, keyword)
+        print(f"\n最终访问: {visited}")
+        sys.exit(0)
+
+    if cmd in ("all", "step4"):
+        # step4 需要先跑完 step3 获取所有商品标题，单独跑时需要手动传标题
+        title = keyword
+        link = step4_parse_qr(ex, title)
+        print(f"\n链接: {link}")
+        sys.exit(0)
+
+    sys.exit(0)

tbsg/steps/executor.py

@@ -0,0 +1,307 @@
+import uiautomator2 as u2
+import time
+import subprocess
+import random
+from typing import Optional, Tuple, Union
+
+
+def find_template_position(small_img_path, big_img_path, threshold=0.8):
+    """
+    在小图中查找大图的位置，返回中心坐标
+    
+    Args:
+        small_img_path: 小块碎片路径（要查找的图片）
+        big_img_path: 整张图片路径（被查找的图片）
+        threshold: 匹配阈值 0-1，默认0.8
+    
+    Returns:
+        (x, y): 中心坐标，如果没找到返回 None
+    """
+    # 1. 读取图片
+    small = cv2.imread(small_img_path)
+    big = cv2.imread(big_img_path)
+    
+    if small is None or big is None:
+        print("❌ 图片读取失败")
+        return None
+    
+    # 2. 模板匹配
+    result = cv2.matchTemplate(big, small, cv2.TM_CCOEFF_NORMED)
+    min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result)
+    
+    print(f"匹配度: {max_val:.3f}")
+    
+    # 3. 判断是否匹配成功
+    if max_val < threshold:
+        print(f"❌ 未找到匹配 (阈值: {threshold})")
+        return None
+    
+    # 4. 计算中心坐标
+    h, w = small.shape[:2]
+    center_x = max_loc[0] + w // 2
+    center_y = max_loc[1] + h // 2
+    
+    
+    return (center_x, center_y)
+
+
+
+
+class SafeExecutor:
+    """
+    安全执行器：为 uiautomator2 提供重试、降级、随机延迟等能力
+    所有方法失败时直接抛出异常，不返回结果对象
+    """
+    
+    def __init__(self,device_id, default_retry: int = 2, human_delay: bool = True):
+        """
+        参数:
+            driver: uiautomator2 连接对象
+            default_retry: 默认重试次数
+            human_delay: 是否添加随机延迟（模拟人类操作）
+        """
+        self.code = 5
+        self.device_id = device_id
+        self.driver = u2.connect_usb(self.device_id)
+        self.default_retry = default_retry
+        self.human_delay = human_delay
+        w, h = self.driver.window_size()
+        self.h = h
+    
+    # ========== 私有工具方法 ==========
+
+
+
+    
+    def _random_sleep(self, min_sec: float = 0.05, max_sec: float = 0.2):
+        """随机睡眠，模拟人类操作间隔"""
+        if self.human_delay:
+            time.sleep(random.uniform(min_sec, max_sec))
+    
+    def _wait_before_retry(self, attempt: int):
+        """重试前等待（指数退避）"""
+        wait_time = min(0.5 * (2 ** attempt), 3)
+        time.sleep(wait_time)
+    
+    # ========== 基础操作 ==========
+    
+    def safe_click(self,xpath):
+        try:
+            if self.No_slider():
+                if type(xpath)!=str:
+                    elem = xpath 
+                else:
+                    elem = self.driver.xpath(xpath)
+  
+                    if not elem.exists:
+                        return False
+                elem.click()
+                time.sleep(2)
+
+                if self.No_slider():
+                    return True
+                else:
+                    return False
+                
+        except Exception as exc:
+            print(f"点击失败，已跳过: {xpath}, error: {exc}")
+            return False
+
+    def tap(self, x: int, y: int, retry: Optional[int] = None):
+        """
+        点击坐标，带重试
+        失败抛出异常
+        """
+        
+        try:
+            self.driver.click(x, y)
+            self._random_sleep(0.1, 0.3)
+            return  # 成功，直接返回
+        
+        except Exception as e:
+            raise Exception(f"点击失败 {x},{y},{e}")
+
+    def swipe(self,distance,times=2):
+        """
+        向上滑动指定距离，分多次完成
+        
+        Args:
+            distance: 总滑动距离（如 1200）
+            times: 分成几次（默认2次）
+        """
+        w, h = self.driver.window_size()
+        x = w // 2 + random.randint(-30, 30)  # X轴随机偏移
+        start_y = int(h * 0.7) + random.randint(-20, 20)  # 起始位置随机偏移
+        per_distance = distance // times
+        
+        current_y = start_y
+        
+        # print(f"📏 向上滑动 {distance}px，分 {times} 次")
+        
+        for i in range(times):
+            # 每次滑动距离随机波动
+            if i == times - 1:
+                end_y = current_y - (distance - (per_distance * (times - 1)))
+            else:
+                # 每段距离 ±20px 波动
+                seg_distance = per_distance + random.randint(-15, 15)
+                end_y = current_y - seg_distance
+            
+            end_y = max(50, end_y)
+            
+            # 滑动时长随机
+            duration = random.uniform(0.06, 0.12)
+            
+            # print(f"  第{i+1}次: {current_y} → {end_y}")
+            
+            self.driver.drag(x, current_y, x, end_y, duration=duration)
+            
+            if i < times - 1:
+                # 暂停时间随机
+                pause_time = random.uniform(0.2, 0.5)
+                #print(f"    ⏸️ 暂停 {pause_time:.2f}s")
+                time.sleep(pause_time)
+            
+            current_y = end_y
+        
+    
+    
+
+
+    def input_text(self,xpaths,text):
+        """
+        输入文本，带重试
+        失败抛出异常
+        """
+        
+        try:
+            search_input = self.driver.xpath(xpaths)
+            if search_input.exists:
+                search_input.set_text('') 
+                time.sleep(1)
+            # 模拟打字间隔
+                search_input.set_text(text)
+
+                return True
+            return False
+        except Exception as e:
+            raise Exception(f"输入失败 '{text}'，: {e}")
+        
+
+
+
+    def is_connected(self) -> bool:
+        """检查与设备的连接是否正常"""
+        try:
+            # 方式1：检查 atx-agent 是否响应
+            info = self.driver.info  # 会请求 atx-agent
+            return True
+        except Exception:
+            self._recover_service()
+
+            return False
+    
+    def _recover_service(self):
+        #重新连接
+
+            # 停止 atx-agent
+            subprocess.run(["adb", "-s", self.device_id, "shell",
+                            "/data/local/tmp/atx-agent", "server", "-d", "--stop"],
+                           capture_output=True, timeout=5)
+            time.sleep(1)
+            # 启动 atx-agent
+            subprocess.run(["adb", "-s", self.device_id, "shell",
+                            "/data/local/tmp/atx-agent", "server", "-d"],
+                           capture_output=True, timeout=5)
+            
+            time.sleep(2)
+            # 重新连接 uiautomator2
+            self.driver = u2.connect_usb(self.device_id)
+
+
+    def No_slider(self) -> bool:
+        """检测当前是否有滑块（默认实现，可覆盖）"""
+        # 这里用你实际的滑块检测逻辑
+        # 示例：检测特定图片或UI元素
+        for i in range(4):
+            if '检测成功' in '检测成功':
+                
+                self._solve_slider()
+            else:
+
+                break                
+            return True
+        
+        return False
+    
+    def _solve_slider(self):
+        """解决滑块验证（默认实现，可覆盖"""
+
+        num = 0
+        while(self.driver.xpath('//*[@text="检测到您的账号存在风险，为保障账号安全，请进行安全验证"]').exists or self.driver.xpath('//*[@text="身份安全验证"]').exists):
+            self.driver.press('back')
+            time.sleep(0.3)
+            num+=1
+            if(num>=6):
+                break
+        
+        time.sleep(random.uniform(2, 3))
+        if self.driver.xpath('//*[@text="检测到您的账号存在风险，为保障账号安全，请进行安全验证"]').exists or self.driver.xpath('//*[@text="身份安全验证"]').exists:
+            self._solve_slider()
+
+            
+        if(self.driver.xpath('//*[@text="验证码"]').exists):
+            print('其他验证码,无xpath')
+            self.driver.screenshot(f'test{code}.jpg')
+            code+=1
+            # self.driver.screenshot('test.jpg')
+            #time.sleep(1)
+            # x= find_template_position('t.jpg','test.jpg')
+            # self.driver.click(x[0],x[1])
+            
+        # 这里写你的滑块处理逻辑
+        # 例如：找到滑块位置，模拟滑动轨迹
+        pass
+
+
+    def safe_get_all(self,xpath):
+        try:
+            return self.driver.xpath(xpath).all()
+        except Exception as exc:
+            print(f"获取元素列表失败，已跳过: {xpath}, error: {exc}")
+            return []
+        
+    def safe_exists(self,xpath):
+        try:
+            return self.driver.xpath(xpath).exists
+        except Exception as exc:
+            print(f"xpath不存在或检查失败，已跳过: {xpath}, error: {exc}")
+            return False
+    def ele(self,xpath):
+        try:
+            return self.driver.xpath(xpath)
+        except Exception as exc:
+            print(f"xpath不存在或检查失败，已跳过: {xpath}, error: {exc}")
+            return False
+
+
+    def safe_get_text(self,xpath):
+        try:
+            elem = self.driver.xpath(xpath)
+            if not elem.exists:
+                return ""
+            node = elem.get()
+            return (node.text or "").strip() if node else ""
+        except Exception as exc:
+            print(f"获取文本失败，已跳过: {xpath}, error: {exc}")
+            return ""
+        
+
+
+    def safe_press_back(self):
+        try:
+            self.driver.press('back')
+            return True
+        except Exception as exc:
+            print(f"safe_press_back 返回失败: {exc}")
+            return False

tbsg/steps/ocr.py

@@ -0,0 +1,110 @@
+"""
+OCR 识别类 — 纯识别，不操作手机
+基于 RapidOCR (rapidocr_onnxruntime)，无需 GPU
+"""
+import re
+from pathlib import Path
+from typing import Optional, Union
+
+import cv2
+import numpy as np
+from rapidocr_onnxruntime import RapidOCR
+
+
+class OCR:
+    """RapidOCR 封装，只负责图片文字识别，不参与设备操作"""
+
+    def __init__(self):
+        import os as _os
+        _os.environ.setdefault("OMP_NUM_THREADS", "1")
+        self._engine = RapidOCR(text_det_thresh=0.3, text_det_box_thresh=0.3)
+
+    # ── 核心识别 ────────────────────────────────────────────
+
+    def recognize(
+        self,
+        image: Union[str, Path, np.ndarray],
+        rect: Optional[list] = None,
+        confidence: float = 0.1,
+        pattern: Optional[str] = None,
+        detail: str = "all",
+    ) -> list:
+        """
+        对图片执行 OCR 识别
+
+        Args:
+            image: 图片路径(str/Path) 或 numpy 数组
+            rect:  裁剪区域 [left, top, right, bottom]，不传则全图
+            confidence: 置信度阈值，默认 0.1
+            pattern:    正则过滤，只返回匹配的文字
+            detail:     'all'=返回完整结果, 'text'=只返回文字列表, 'first'=只返回第一个匹配文字
+
+        Returns:
+            detail='all'时:
+                [{'text': str, 'confidence': float,
+                  'bbox': [[x1,y1],[x2,y1],[x2,y2],[x1,y2]]}, ...]
+            detail='text'时:
+                ['文字1', '文字2', ...]
+            detail='first'时:
+                str 或 None
+        """
+        # 1. 读取图片
+        if isinstance(image, (str, Path)):
+            img = cv2.imread(str(image))
+            if img is None:
+                raise FileNotFoundError(f"无法读取图片: {image}")
+            img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
+        elif isinstance(image, np.ndarray):
+            img = image
+        else:
+            raise TypeError(f"不支持的图片类型: {type(image)}")
+
+        # 2. 裁剪区域（在压缩前，用原始坐标）
+        crop_offset_l, crop_offset_t = 0, 0
+        if rect:
+            l, t, r, b = rect
+            img = img[t:b, l:r]
+            crop_offset_l, crop_offset_t = l, t
+
+        # 3. 压缩大图，避免ONNX OOM
+        orig_h, orig_w = img.shape[:2]
+        if orig_h < 1 or orig_w < 1:
+            return [] if detail != "first" else ""
+        scale = 1.0
+        max_size = 960
+        if max(orig_h, orig_w) > max_size:
+            scale = max_size / max(orig_h, orig_w)
+            img = cv2.resize(img, (max(1, int(orig_w * scale)), max(1, int(orig_h * scale))))
+
+        # 4. 执行 OCR
+        raw, _ = self._engine(img)
+
+        # 5. 结构化 & 过滤，把坐标映射回原始图
+        results = []
+        for bbox, text, conf in (raw or []):
+            if conf < confidence:
+                continue
+            if pattern and not re.search(pattern, text):
+                continue
+            results.append({
+                "text": text,
+                "confidence": round(float(conf), 4),
+                "bbox": [[int(p[0] / scale) + crop_offset_l, int(p[1] / scale) + crop_offset_t] for p in bbox],
+            })
+
+        # 5. 按需返回格式
+        if detail == "text":
+            return [r["text"] for r in results]
+        elif detail == "first":
+            return results[0]["text"] if results else None
+        return results  # detail='all'
+
+    # ── 快捷方法 ────────────────────────────────────────────
+
+    def recognize_file(self, path: str, **kwargs) -> list:
+        """从文件路径识别"""
+        return self.recognize(path, **kwargs)
+
+    def recognize_array(self, arr: np.ndarray, **kwargs) -> list:
+        """从 numpy 数组识别"""
+        return self.recognize(arr, **kwargs)

tbsg/yzm/tmp_captcha_test2.py

@@ -0,0 +1,349 @@
+"""test.py 轨迹 + TouchPipe压感 + 可靠执行"""
+import sys, os, json, time, random, math, threading
+import uiautomator2 as u2
+from uiautomator2.core import AdbHTTPConnection
+import cv2, numpy as np, base64, requests
+from rapidocr_onnxruntime import RapidOCR
+
+os.environ["PYTHONIOENCODING"] = "utf-8"
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from detect_slider_button import detect_slider_button
+from detect_captcha_edge import detect_captcha_left_edge
+
+
+ocr_eng = RapidOCR()
+TOKEN = "1nDVocTE2mJ0yLEYb2sZJ5uUY2VIEoGTkIpW44X7Kgk"
+JFBYM_URL = "http://api.jfbym.com/api/YmServer/customApi"
+IMG_ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "image", "test2_image")
+OUT = os.path.join(IMG_ROOT, "d")
+ALIGN_DIR = os.path.join(IMG_ROOT, "align")
+SUCCESS_DIR = os.path.join(IMG_ROOT, "success")
+FAILURE_DIR = os.path.join(IMG_ROOT, "failure")
+os.makedirs(OUT, exist_ok=True)
+os.makedirs(ALIGN_DIR, exist_ok=True)
+os.makedirs(SUCCESS_DIR, exist_ok=True)
+os.makedirs(FAILURE_DIR, exist_ok=True)
+
+
+OFFSET_COMPENSATE = random.uniform(-7,-5)         # MICRO_MODE="fixed" 偏移量(正=右,负=左)
+MICRO_MODE = "fixed"            # "fixed"=固定偏移 / "visual"=红绿线对齐 / "none"=不用
+
+
+def _clamp(v, lo, hi):
+    return max(lo, min(v, hi))
+
+
+def _pressure_curve(i, n, phase):
+    """压力曲线：开头高→巡航低→末尾高。phase: 'right' 或 'return'。"""
+    t = i / max(1, n - 1)
+    if t < 0.1:        return random.randint(60, 90)   # 开头按重
+    elif t > 0.90:     return random.randint(55, 85)   # 末尾对准按重
+    else:              return random.randint(15, 35)   # 巡航轻按
+
+
+class TouchPipe:
+    def __init__(self, dev):
+        self._dev = dev; self._conn = None; self._sock = None
+        self._stop = threading.Event(); self._drainer = None
+        self._lock = threading.Lock(); self._fallback = False
+
+    def open(self):
+        try:
+            self._conn = AdbHTTPConnection(self._dev.adb_device, port=9008)
+            self._conn.timeout = 15; self._conn.connect()
+            self._sock = self._conn.sock; self._sock.settimeout(0.5)
+            self._drainer = threading.Thread(target=self._drain, daemon=True)
+            self._drainer.start()
+        except Exception: self._fallback = True
+        return self
+
+    def _drain(self):
+        while not self._stop.is_set():
+            try:
+                if not self._sock.recv(65536): break
+            except Exception: continue
+
+    def move(self, x, y, pressure=0):
+        if self._fallback:
+            try: self._dev.touch.move(int(x), int(y))
+            except Exception: pass
+            return
+        try:
+            with self._lock:
+                self._sock.sendall(self._req(x, y, pressure))
+        except Exception: self._fallback = True; self.move(x, y, pressure)
+
+    def _req(self, x, y, pressure):
+        body = json.dumps({"jsonrpc":"2.0","id":1,"method":"injectInputEvent",
+                           "params":[2,int(x),int(y),int(pressure)]}).encode()
+        return (f"POST /jsonrpc/0 HTTP/1.1\r\nHost: localhost\r\n"
+                f"User-Agent: u2\r\nAccept-Encoding: \r\n"
+                f"Content-Type: application/json\r\nContent-Length: {len(body)}\r\n"
+                f"Connection: keep-alive\r\n\r\n").encode() + body
+
+    def close(self):
+        self._stop.set()
+        if self._drainer: self._drainer.join(timeout=1)
+        try:
+            if self._sock: self._sock.close()
+        except Exception: pass
+
+
+def _build_right_track(start_x, start_y, end_x):
+    """test.py 原样 — 钟形鼓包 wobble, EMA 0.55"""
+    dist = abs(end_x - start_x)
+    steps = int(_clamp(dist / random.uniform(6.0, 8.0), 35, 50)) + 20
+    end_yd = random.randint(-30, 30)
+    wobbles = [(random.uniform(0.15, 0.85), random.uniform(-25, 25))
+               for _ in range(random.randint(2, 4))]
+    points = [(int(start_x), int(start_y))]
+    smooth_y = float(start_y)
+    for i in range(1, steps + 1):
+        t = i / steps
+        x = start_x + dist * (1.0 - (1.0 - t) ** 2.0)
+        y = start_y + end_yd * (t ** 0.6)
+        if 0.15 < t < 0.9:
+            for wp, wv in wobbles:
+                dw = abs(t - wp)
+                if dw < 0.12: y += wv * (1 - dw / 0.12)
+        smooth_y = smooth_y * 0.55 + y * 0.45
+        points.append((int(round(x)), int(round(smooth_y))))
+    return points
+
+
+def _build_human_return_track(start_x, start_y, target_x):
+    """test.py 原样 — wobble, EMA 0.55, blend t>0.5 不打勾"""
+    distance = start_x - target_x
+    if distance <= 0:
+        return [(int(target_x), int(start_y))]
+    # 点数按折回距离自适应，对标真机但控制上限避免等太久
+    if distance > 600:
+        steps = random.randint(150, 200)       # 长折, Y必往下漂
+        end_dy = random.randint(80, 200); cat = '长折'
+    elif distance > 450:
+        steps = random.randint(120, 170)       # 中折
+        end_dy = random.randint(-20, 40); cat = '中折'
+    else:
+        steps = int(_clamp(distance / random.uniform(3.0, 5.0), 35, 55))
+        end_dy = random.randint(-20, 40); cat = '短折'
+    print(f'  折回: {cat} distance={distance:.0f}px {steps}点 Y漂={end_dy:+d}')
+
+    # 两段: 巡航60%点数走88%距离(大步), 减速40%点数走12%距离(密集蠕动)
+    cruise_n = int(steps * 0.60)
+    decel_n = steps - cruise_n
+    target_y = start_y + end_dy
+    r_wobbles = [(random.uniform(0.15, 0.5), random.uniform(-20, 20))
+                  for _ in range(random.randint(2, 4))]
+    track, smooth_y = [], float(start_y)
+
+    cruise_dist = distance * 0.88
+    decel_dist = distance * 0.12
+
+    # ── 前90%大步巡航 ──
+    for i in range(1, cruise_n + 1):
+        t = i / cruise_n
+        x = start_x - cruise_dist * t
+        y = start_y + (target_y - start_y) * t
+        if 0.1 < t < 0.5:
+            for wp, wv in r_wobbles:
+                dw = abs(t - wp)
+                if dw < 0.12: y += wv * (1 - dw / 0.12)
+        smooth_y = smooth_y * 0.55 + y * 0.45
+        track.append((int(round(x)), int(round(smooth_y))))
+
+    # ── 后10%小步减速对准 ──
+    dsx, dsy = track[-1]
+    for i in range(1, decel_n + 1):
+        tt = i / decel_n
+        ease = 1 - (1 - tt) ** 2
+        x = dsx - decel_dist * ease
+        y = dsy + (target_y - dsy) * tt
+        track.append((int(round(x)), int(round(y))))
+
+    if track: track[-1] = (int(target_x), int(target_y))
+    return track
+
+
+def save(name, img):
+    cv2.imwrite(os.path.join(OUT, name), img)
+
+
+def save_track_image(pts, filepath):
+    if len(pts) < 2: return
+    xs = [p[0] for p in pts]; ys = [p[1] for p in pts]
+    m = 50
+    w = max(xs)-min(xs)+m*2; h = max(ys)-min(ys)+m*2
+    w, h = max(w,200), max(h,100)
+    c = np.ones((h, w, 3), dtype=np.uint8)*255
+    for i in range(1, len(pts)):
+        r = i/len(pts); g = 200 if r<0.5 else int(200*(1-r)*2); b = int(200*r*2) if r<0.5 else 200
+        cv2.line(c, (pts[i-1][0]-min(xs)+m, pts[i-1][1]-min(ys)+m),
+                 (pts[i][0]-min(xs)+m, pts[i][1]-min(ys)+m), (0,g,b), 1)
+    cv2.circle(c, (pts[0][0]-min(xs)+m, pts[0][1]-min(ys)+m), 4, (0,200,0), -1)
+    cv2.circle(c, (pts[-1][0]-min(xs)+m, pts[-1][1]-min(ys)+m), 4, (0,0,200), -1)
+    cv2.imwrite(filepath, c)
+
+
+def solve_slider(driver, sx=None):
+    global d, W, H
+    d = driver; W, H = d.window_size(); d.screen_on()
+
+    # 自动判断分辨率: 宽>1000=高分屏(sx=163), 否则=低分屏(sx=87)
+    if sx is None:
+        sx = 163 if W > 1000 else 87
+    print(f"  屏幕: {W}x{H} → sx={sx} ({'高分' if W>1000 else '低分'})")
+
+    screen = d.screenshot(format="opencv")
+    slider_y = y_top = slider_bottom = None
+    try:
+        cv2.imwrite(os.path.join(OUT, "_tmp_slider.png"), screen)
+        info = detect_slider_button(os.path.join(OUT, "_tmp_slider.png"))
+        if info and len(info) >= 3: sx = info[0] + int(info[2]/2)
+    except: pass
+    ocr_r = ocr_eng(screen)
+    if ocr_r and ocr_r[0]:
+        for item in ocr_r[0]:
+            t = item[1]; cy = int((item[0][0][1]+item[0][2][1])/2)
+            if "请按照说明拖动滑块" in t: slider_y = slider_y or cy; slider_bottom = int(item[0][2][1])
+            if "松开" in t: y_top = int(item[0][0][1])
+    if slider_y is None: return False
+    sy = slider_y
+
+    # 右滑 — TouchPipe + 压感
+    d.touch.down(sx, sy); time.sleep(0.08)
+    pipe = TouchPipe(d).open()
+    t0 = time.perf_counter(); track_pts = []
+    right_track = _build_right_track(sx, sy, W-30)
+    for i, (x, y) in enumerate(right_track):
+        p = _pressure_curve(i, len(right_track), 'right')
+        pipe.move(x, y, p)
+        track_pts.append({"x": x, "y": y, "pressure": p,
+                          "rel_ms": (time.perf_counter()-t0)*1000})
+        time.sleep(random.uniform(0.002, 0.004))
+
+    # 等设备消化完右滑点再截图 — 每点预留 28ms 处理时间
+    settle_ms = max(500, len(right_track) * 18)
+    time.sleep(settle_ms / 1000.0)
+
+    # 展开后的幕布截图（后面叠加轨迹用）
+    unfolded = d.screenshot(format="opencv")
+
+    # JFBYM
+    crop = d.screenshot(format="opencv")
+    if y_top and slider_bottom: crop = crop[y_top:slider_bottom, :]
+    _, buf = cv2.imencode(".png", crop)
+    gap = None
+    for a in range(3):
+        try:
+            r = requests.post(JFBYM_URL, json={"token":TOKEN,"type":"20226","image":base64.b64encode(buf).decode()}, timeout=35).json()
+            if r.get("data") and r["data"].get("data"): gap = int(r["data"]["data"])
+            elif r.get("data") and isinstance(r["data"],(int,float)): gap = int(r["data"])
+            if gap is not None: break
+            time.sleep(2)
+        except: time.sleep(2)
+    if gap is None: pipe.close(); d.touch.up(W-20, sy); return False
+
+    # 折回
+    target_x = gap + OFFSET_COMPENSATE
+    return_track = _build_human_return_track(W-30, right_track[-1][1], target_x)
+    for i, (x, y) in enumerate(return_track):
+        p = _pressure_curve(i, len(return_track), 'return')
+        pipe.move(x, y, p)
+        track_pts.append({"x": x, "y": y, "pressure": p,
+                          "rel_ms": (time.perf_counter()-t0)*1000})
+        # 前60%巡航快发, 后40%减速慢发
+        frac = i / max(1, len(return_track)-1)
+        if frac > 0.60:
+            time.sleep(random.uniform(0.004, 0.008))
+        else:
+            time.sleep(random.uniform(0.000, 0.002))
+    # 等设备消化完折回点
+    # settle 按距离缩放: 每px约2~4ms, 上下限保护
+    ret_distance = (W-30) - target_x
+    settle_ms2 = int(ret_distance * random.uniform(2.0, 4.0))
+    settle_ms2 = max(500, min(3500, settle_ms2))
+    if ret_distance > 600: scat = '长等'
+    elif ret_distance > 450: scat = '中等'
+    else: scat = '短等'
+    print(f'  settle={settle_ms2}ms({settle_ms2/1000:.1f}s)')
+    time.sleep(settle_ms2 / 1000.0)
+
+    # 微调: 三种模式
+    cur_x, cur_y = return_track[-1]
+    aligned_img = None
+    if MICRO_MODE == "fixed":
+        target_cur_x = int(_clamp(cur_x + OFFSET_COMPENSATE, 10, W-10))
+        steps = abs(target_cur_x - cur_x)
+        sign = 1 if target_cur_x > cur_x else -1
+        for s in range(steps):
+            cur_x += sign
+            pipe.move(cur_x, cur_y, 60)
+            time.sleep(random.uniform(0.005, 0.010))
+        print(f"  固定偏移: {OFFSET_COMPENSATE:+.0f}px 分{steps}步 -> x={cur_x}")
+    elif MICRO_MODE == "visual":
+        fallback_x = cur_x; prev_pl = None
+        for attempt in range(3):
+            time.sleep(0.3)
+            check = d.screenshot(format="opencv")
+            cv2.imwrite(os.path.join(ALIGN_DIR, f"align_{attempt}.png"), check)
+            try: pl, _ = detect_captcha_left_edge(check)
+            except Exception: break
+            diff = pl - gap
+            marked = check.copy()
+            cv2.line(marked, (pl, 0), (pl, marked.shape[0]), (0, 255, 0), 3)
+            cv2.line(marked, (gap, 0), (gap, marked.shape[0]), (0, 0, 255), 3)
+            cv2.imwrite(os.path.join(ALIGN_DIR, f"align_{attempt}_marked.png"), marked)
+            if abs(diff) <= 5: break
+            if prev_pl is not None and pl == prev_pl:
+                cur_x = int(_clamp(fallback_x,10,W-10)); pipe.move(cur_x,cur_y,60); break
+            prev_pl = pl
+            cur_x = int(_clamp(cur_x-diff,10,W-10)); pipe.move(cur_x,cur_y,50); time.sleep(0.05)
+    # 滑块对准缺口的截图，做轨迹背景
+    aligned_img = d.screenshot(format="opencv")
+    pipe.close(); d.touch.up(int(cur_x),int(cur_y)); time.sleep(2)
+
+    # 验证
+    final = d.screenshot(format="opencv"); check_r = ocr_eng(final)
+    passed = True
+    if check_r and check_r[0]:
+        if any("拖动滑块" in it[1] or "请按住滑块" in it[1] or "安全验证" in it[1] for it in check_r[0]): passed = False
+    if not passed:
+        time.sleep(2); final = d.screenshot(format="opencv"); check_r = ocr_eng(final)
+        passed = True
+        if check_r and check_r[0]:
+            if any("拖动滑块" in it[1] or "请按住滑块" in it[1] or "安全验证" in it[1] for it in check_r[0]): passed = False
+
+    all_track = right_track + return_track
+    rd = SUCCESS_DIR if passed else FAILURE_DIR
+    now = time.localtime()
+    dev_id = getattr(d, 'serial', getattr(d, '_serial', 'unknown'))
+    prefix = f"{dev_id}_{now.tm_year:04d}{now.tm_mon:02d}{now.tm_mday:02d}_{now.tm_hour:02d}{now.tm_min:02d}{now.tm_sec:02d}"
+    save_track_image(all_track, os.path.join(rd, f"{prefix}_track.png"))
+    # 轨迹叠加在滑块对齐缺口时的截图上
+    if aligned_img is not None:
+        b_img = aligned_img.copy()
+        for i in range(1, len(all_track)):
+            cv2.line(b_img, all_track[i-1], all_track[i], (0, 200, 200), 2)
+        cv2.circle(b_img, all_track[0], 6, (0, 255, 0), -1)
+        cv2.circle(b_img, all_track[-1], 6, (0, 0, 255), -1)
+        cv2.imwrite(os.path.join(rd, f"{prefix}_b.png"), b_img)
+    tj = {"device": dev_id, "screen": {"w": W, "h": H},
+          "time": time.strftime("%Y-%m-%d %H:%M:%S", now),
+          "total_points": len(track_pts),
+          "duration_ms": track_pts[-1]["rel_ms"] if track_pts else 0,
+          "start": {"x": track_pts[0]["x"], "y": track_pts[0]["y"]},
+          "end": {"x": track_pts[-1]["x"], "y": track_pts[-1]["y"]},
+          "dx": track_pts[-1]["x"]-track_pts[0]["x"],
+          "dy": track_pts[-1]["y"]-track_pts[0]["y"],
+          "passed": passed, "gap": gap,
+          "points": track_pts}
+    with open(os.path.join(rd, f"{prefix}_track.json"), 'w', encoding='utf-8') as fp:
+        json.dump(tj, fp, ensure_ascii=False)
+    return passed
+
+
+if __name__ == "__main__":
+    DEVICE = "NJZX8DZXT47HQGWO"
+    d = u2.connect(DEVICE)
+    print(f"设备: {DEVICE}")
+    solve_slider(d, sx=None)  # None=自动判断分辨率