project/mt_V2/main.py

from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
import requests
import base64
import cv2
import uiautomator2 as u2
import time
import subprocess
import re
import random
import datetime
import sys
import json
import unicodedata
from aip import AipOcr
import threading
from collections import deque
import numpy as np
import secrets
import os
import oss2
import urllib.parse
from config import Config
from logger import setup_logger
import logging

from PIL import Image
import http.client
import traceback
from pathlib import Path
from db import get_mysql
from scheduler import CrawlerScheduler
from decimal import Decimal

from area import AreaService

try:
    from yzm import yzm as solve_captcha
except Exception:
    solve_captcha = None

setup_logger("mt_spider")  # 初始化日志


def get_access_token():
    AppKey = "tRK2RhyItCSh6BzyT4CNVXQa"
    AppSrcret = "TDgKiPo94i2mOM1sDqOuDnlcK1bG66jh"
    token_url = 'https://aip.baidubce.com/oauth/2.0/token'
    url = f"{token_url}?grant_type=client_credentials&client_id={AppKey}&client_secret={AppSrcret}"

    payload = ""
    headers = {
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }

    response = requests.request("POST", url, headers=headers, data=payload)

    try:
        return response.json()['access_token']
    except:
        return None

LOOP_INTERVAL_SECONDS = 600  # 每轮任务之间的等待间隔（秒），默认10分钟
DEVICE_ID = "T4VK4LM7AAUOV8AY"  # 指定要连接的设备
PLATFORM_MT = 4
# True: 跑 device_list 里手动配置的任务；False: 从外部调度器获取任务
MANUAL_MODE = False
SEARCH_TASK_MODE = "name_with_each_spec"
MAX_RUN_DEVICE_RETRIES = 3  # run_device 外层重试上限，超过后回告并停止
OPEN_PRODUCT_LIST_PAGE_RETRY = 3
FAILURE_NOTICE_THRESHOLD = 3
REQUEST_ERROR_STOP_THRESHOLD = 5
PRODUCT_LINK_STOP_THRESHOLD = 5
# 是否记录"因为何种问题重新开始"的日志
ENABLE_RESTART_REASON_LOG = True
# 重新开始原因日志文件
RESTART_REASON_LOG_FILE = "./restart_reason_logs/a_mt_restart_reason.log"
# 是否启用"验证码连续重启失败"告警（飞书）
ENABLE_CAPTCHA_RESTART_ALERT = False
# 店铺补齐调试日志开关
ENABLE_SHOP_DEBUG = False
# 验证码连续重启失败告警阈值
CAPTCHA_RESTART_ALERT_THRESHOLD = 3
# 验证码连续重启计数持久化文件
CAPTCHA_RESTART_COUNTER_FILE = "./restart_reason_logs/captcha_restart_counter.json"
# 1小时内验证码导致重启超过此值则直接停止任务并回告
CAPTCHA_STOP_THRESHOLD = 5
CAPTCHA_STOP_WINDOW_SECONDS = 3600

failure_notice_counters = {}
failure_notice_lock = threading.Lock()
captcha_restart_lock = threading.Lock()
captcha_restart_counts_cache = None
oss_bucket_cache_lock = threading.Lock()
oss_bucket_cache = {}


def decode_qr(image_path):
    for i in range(3):
        img = cv2.imread(image_path)

        # 放大2~4倍（关键）
        if (i == 0):
            img2 = img[1200:, :300]

            img3 = img[1500:2000,50:500]
            img = cv2.resize(img, None, fx=3, fy=3, interpolation=cv2.INTER_CUBIC)
            img2 = cv2.resize(img2, None, fx=3, fy=3, interpolation=cv2.INTER_CUBIC)

        else:
            img3 = img[1500:2000,50:500]
            img2 = img[1000:, :200]
            img = cv2.resize(img, None, fx=(3 - i), fy=(3 - i), interpolation=cv2.INTER_CUBIC)
            img2 = cv2.resize(img2, None, fx=(3 - i), fy=(3 - i), interpolation=cv2.INTER_CUBIC)

        detector = cv2.wechat_qrcode_WeChatQRCode()

        data, points = detector.detectAndDecode(img2)
        if data != ():
            return data

        data, points = detector.detectAndDecode(img3)
        if data != ():
            return data

        data, points = detector.detectAndDecode(img)

        if data != ():
            return data

        if (data == ()):
            data = ''
    return data


class CollectionStopError(RuntimeError):
    """需要立即停止当前采集流程的致命异常。"""


class AccountBlockedError(CollectionStopError):
    """疑似账号被封禁，需立即停止当前采集流程。"""


class ProductLinkUnavailableError(CollectionStopError):
    """连续获取不到商品链接，需立即停止当前采集流程。"""


class WindControlStuckError(CollectionStopError):
    """检测到风控卡死：连续重复商品 + "加载更多"按钮持续存在，疑似被风控限制。"""


def is_high_resolution_device(d):
    """检测是否为高分辨率设备 (1220x2712)。
    通过屏幕高度判断：>2000 为高分辨率，否则为低分辨率 (720x1640)。
    """
    try:
        h = d.info.get('displayHeight', 0)
        return h > 2000
    except Exception:
        return False


def parse_optional_int(value, default=None):
    if value in (None, ""):
        return default
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def parse_spec_list(value):
    if value is None:
        return []
    if isinstance(value, (list, tuple)):
        return [str(item).strip() for item in value if str(item).strip()]
    text = str(value).strip()
    if not text:
        return []
    parts = re.split(r"[，,/\s]+", text)
    return [part.strip() for part in parts if part.strip()]


def normalize_match_text(value):
    text = "" if value is None else str(value)
    # 统一全角/半角，移除各种空白和零宽字符，避免"看起来一样但匹配失败"
    text = unicodedata.normalize("NFKC", text)
    text = re.sub(r"[\s\u00A0\u200B-\u200D\uFEFF]+", "", text)
    return text


def build_search_variants(search_key, spec_list, mode=SEARCH_TASK_MODE):
    base_search_key = str(search_key or "").strip()
    cleaned_specs = [str(spec).strip() for spec in (spec_list or []) if str(spec).strip()]

    if mode == "name_with_each_spec" and cleaned_specs:
        variants = []
        seen = set()
        for spec in cleaned_specs:
            query = f"{base_search_key}{spec}".strip() if base_search_key else spec
            if query and query not in seen:
                variants.append({
                    "search_key": query,
                    "spec_list": [spec],
                })
                seen.add(query)
        if variants:
            return variants

    return [{
        "search_key": base_search_key,
        "spec_list": cleaned_specs,
    }]


def _build_failure_counter_key(source, device_id, task_id=None, search_key=None):
    return f"{source}:{device_id}:{task_id or ''}:{search_key or ''}"


def _is_transient_open_page_error(err_msg):
    text = str(err_msg or "")
    keywords = (
        "open_product_list_page",
        "点击首页搜索入口失败",
        "进入看病买药页失败",
        "vf_search_carousel_text",
        "看病买药",
    )
    return any(k in text for k in keywords)


def should_send_failure_notice(counter_key, err_msg, threshold=FAILURE_NOTICE_THRESHOLD):
    is_transient = _is_transient_open_page_error(err_msg)
    with failure_notice_lock:
        if is_transient:
            count = failure_notice_counters.get(counter_key, 0) + 1
            failure_notice_counters[counter_key] = count
            return count >= threshold, count, True
        failure_notice_counters[counter_key] = 0
        return True, 1, False


def reset_failure_notice_counter(counter_key):
    with failure_notice_lock:
        failure_notice_counters[counter_key] = 0


def _is_captcha_related_error(err_msg, traceback_text=None):
    text = f"{err_msg or ''}\n{traceback_text or ''}"
    lower_text = text.lower()
    keywords_cn = ("验证码", "滑块", "拼图", "安全验证", "人机验证", "请点击")
    keywords_en = ("captcha", "slider", "puzzle", "verify", "verification", "yoda")
    return any(k in text for k in keywords_cn) or any(k in lower_text for k in keywords_en)


def _load_captcha_restart_counts():
    global captcha_restart_counts_cache
    if captcha_restart_counts_cache is not None:
        return captcha_restart_counts_cache

    data = {}
    try:
        if os.path.exists(CAPTCHA_RESTART_COUNTER_FILE):
            with open(CAPTCHA_RESTART_COUNTER_FILE, "r", encoding="utf-8") as f:
                raw = json.load(f)
                if isinstance(raw, dict):
                    for k, v in raw.items():
                        try:
                            key = str(k)
                            if isinstance(v, list):
                                # 新格式：时间戳列表，只保留字符串
                                data[key] = [str(ts) for ts in v if isinstance(ts, str)]
                            elif isinstance(v, (int, float)):
                                # 兼容旧格式：单个整数 → 转为空列表（旧数据清零）
                                data[key] = []
                            else:
                                data[key] = []
                        except Exception:
                            continue
    except Exception as e:
        logging.exception(f"读取验证码重启计数失败: {e}")
    captcha_restart_counts_cache = data
    return captcha_restart_counts_cache


def _save_captcha_restart_counts(data):
    try:
        log_dir = os.path.dirname(CAPTCHA_RESTART_COUNTER_FILE)
        if log_dir:
            os.makedirs(log_dir, exist_ok=True)
        with open(CAPTCHA_RESTART_COUNTER_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    except Exception as e:
        logging.exception(f"写入验证码重启计数失败: {e}")


def _prune_expired_timestamps(timestamps, window_seconds):
    """剔除超过时间窗口的旧时间戳"""
    now = time.time()
    cutoff = now - window_seconds
    return [ts for ts in timestamps if _timestamp_to_epoch(ts) > cutoff]


def _timestamp_to_epoch(ts):
    """将 ISO 时间字符串转为 epoch 秒，解析失败返回 0"""
    try:
        return time.mktime(time.strptime(str(ts), "%Y-%m-%dT%H:%M:%S"))
    except Exception:
        return 0


def increase_captcha_restart_count(counter_key):
    """记录一次验证码重启，返回当前时间窗口内的累计次数"""
    with captcha_restart_lock:
        data = _load_captcha_restart_counts()
        timestamps = data.get(counter_key, [])
        if not isinstance(timestamps, list):
            timestamps = []
        # 追加当前时间
        now_str = time.strftime("%Y-%m-%dT%H:%M:%S")
        timestamps.append(now_str)
        # 剔除超过 1 小时的旧记录
        timestamps = _prune_expired_timestamps(timestamps, CAPTCHA_STOP_WINDOW_SECONDS)
        data[counter_key] = timestamps
        _save_captcha_restart_counts(data)
        return len(timestamps)


def get_captcha_restart_count_in_window(counter_key, window_seconds=None):
    """查询时间窗口内的验证码重启次数（不追加新记录）"""
    if window_seconds is None:
        window_seconds = CAPTCHA_STOP_WINDOW_SECONDS
    with captcha_restart_lock:
        data = _load_captcha_restart_counts()
        timestamps = data.get(counter_key, [])
        if not isinstance(timestamps, list):
            return 0
        timestamps = _prune_expired_timestamps(timestamps, window_seconds)
        return len(timestamps)


def reset_captcha_restart_count(counter_key):
    with captcha_restart_lock:
        data = _load_captcha_restart_counts()
        timestamps = data.get(counter_key)
        if timestamps:
            data[counter_key] = []
            _save_captcha_restart_counts(data)


def should_send_captcha_restart_alert(restart_count):
    if not ENABLE_CAPTCHA_RESTART_ALERT:
        return False
    if CAPTCHA_RESTART_ALERT_THRESHOLD <= 0:
        return False
    if restart_count < CAPTCHA_RESTART_ALERT_THRESHOLD:
        return False
    return restart_count % CAPTCHA_RESTART_ALERT_THRESHOLD == 0


def record_restart_reason(
        reason,
        device_id=None,
        task_id=None,
        step=None,
        action=None,
        fail_count=None,
        retry_limit=None,
        cycle_no=None,
        search_key=None,
        source=None,
        exc=None,
        traceback_text=None,
):
    if not ENABLE_RESTART_REASON_LOG:
        return
    payload = {
        "time": time.strftime("%Y-%m-%d %H:%M:%S"),
        "source": source,
        "reason": str(reason),
        "device_id": device_id,
        "task_id": task_id,
        "step": step,
        "action": action,
        "fail_count": fail_count,
        "retry_limit": retry_limit,
        "cycle_no": cycle_no,
        "search_key": search_key,
        "error": str(exc) if exc is not None else None,
        "traceback": traceback_text,
    }
    try:
        log_dir = os.path.dirname(RESTART_REASON_LOG_FILE)
        if log_dir:
            os.makedirs(log_dir, exist_ok=True)
        with open(RESTART_REASON_LOG_FILE, "a", encoding="utf-8") as f:
            f.write(json.dumps(payload, ensure_ascii=False) + "\n")
    except Exception as log_err:
        logging.exception(f"写入重启原因日志失败: {log_err}")


class SpiderMonitor(threading.Thread):
    """全局弹窗监控线程（增强版）"""

    def __init__(self, spider_instance):
        super().__init__(daemon=True)
        self.spider = spider_instance
        self.running = True
        self.pausing = threading.Event()  # 主线程同步事件
        self.last_verification_time = 0
        self.verification_count = 0
        # 验证码重试次数
        self.MAX_VERIFICATION_RETRY = 6
        self.recent_clicks = deque(maxlen=10)  # 防重复点击
        self.logger = logging.getLogger("SpiderMonitor")
        self.last_verification_probe_log_ts = 0
        self.last_reconnect_ts = 0
        # 验证码出现频率统计（1小时滑动窗口）
        self.captcha_appearance_timestamps = []  # 验证码出现时间戳列表
        self.MAX_CAPTCHA_APPEARANCES = 5  # 1小时内最多出现次数
        self.CAPTCHA_APPEARANCE_WINDOW = 3600  # 时间窗口（秒）
        self.captcha_appearance_limit_reached = False  # 是否已达到验证码出现上限

        # 可配置化弹窗规则
        self.popup_rules = {
            "simple": [
                ('//*[@text="确定"]', "点击确定"),
                ('//*[@text="允许"]', "点击允许"),
                ('//*[@text="关闭"]', "点击关闭"),
                ('//*[@resource-id="com.sankuai.meituan:id/close"]', "关闭按钮"),
                ('//*[@resource-id="com.sankuai.meituan:id/address_center_location_close"]', "关闭按钮"),
                ('//*[@resource-id="com.sankuai.meituan:id/location_close"]', "关闭按钮"),
                ('//*[@resource-id="com.sankuai.meituan:id/btn_close"]', "关闭按钮"),
            ],
            "verification": [
                '//*[contains(@text, "依次点击")]',
                '//*[contains(@text, "拖动滑块刚")]',  # 这个需要拖动滑块至最右边，然后再截图
                '//*[contains(@text, "请输入图片中的内容")]',
                '//*[contains(@text, "用最短线连接")]',
                '//*[contains(@text, "请按语序依次点击")]',
                '//*[contains(@text, "请向右滑动滑块")]',
                '//*[contains(@text, "请拖动下方滑块完成拼图")]',
                '//*[contains(@text, "请点击") and contains(@text, "图")]',
                '//*[contains(@text, "验证码")]',
                '//*[contains(@text, "安全验证")]',
                '//*[contains(@text, "完成拼图")]',
                '//*[contains(@text, "滑块")]',
                '//*[contains(@resource-id, "captcha")]',
                '//*[contains(@resource-id, "yoda")]',
                '//*[contains(@resource-id, "puzzle")]',
                '//*[contains(@resource-id, "slider")]',
                '//*[contains(@resource-id, "verify")]',
            ]
        }

    def _device_tag(self):
        device_id = getattr(self.spider, "device_id", None) or "unknown"
        device_name = getattr(self.spider, "device_name", None) or "unknown"
        equipment_id = getattr(self.spider, "equipment_id", None)
        if equipment_id not in (None, ""):
            return f"[device_id={device_id} device_name={device_name} equipment_id={equipment_id}]"
        return f"[device_id={device_id} device_name={device_name}]"

    def run(self):
        while self.running:
            try:
                handled = self.check_and_handle_popup()
                time.sleep(2 if handled else 1)
            except http.client.RemoteDisconnected as e:
                self.logger.exception("%s 监控线程连接断开: %s", self._device_tag(), e)
                now = time.time()
                # 监控线程受控重连：最小间隔内仅触发一次，避免抖动重连
                if now - self.last_reconnect_ts >= 8:
                    self.last_reconnect_ts = now
                    try:
                        if hasattr(self.spider, "reconnect_device"):
                            if self.spider.reconnect_device():
                                self.logger.info("%s 监控线程触发重连成功", self._device_tag())
                    except Exception:
                        self.logger.exception("%s 监控线程触发重连失败", self._device_tag())
                time.sleep(2)
            except Exception as e:
                self.logger.exception("%s 监控线程异常: %s", self._device_tag(), e)
                time.sleep(1)

    def _is_recent_click(self, xpath):
        """防止重复点击同一个弹窗"""
        key = f"{xpath}_{int(time.time())}"
        if key in self.recent_clicks:
            return True
        self.recent_clicks.append(key)
        return False

    def _try_auto_solve_verification(self, d):
        if solve_captcha is None:
            return False
        try:
            device_id = getattr(self.spider, "device_id", None)
            try:
                solve_captcha(d, device_id=device_id)
            except TypeError:
                solve_captcha(d)
            return True
        except Exception as e:
            self.logger.exception("%s auto captcha solve failed: %s", self._device_tag(), e)
            return False

    def _get_xpath_bounds(self, d, xpath):
        try:
            node = d.xpath(xpath)
            if not node.exists:
                return None
            info = node.info or {}
            bounds = info.get("visibleBounds") or info.get("bounds") or {}
            if not bounds:
                return None
            left = int(bounds.get("left", 0))
            right = int(bounds.get("right", 0))
            top = int(bounds.get("top", 0))
            bottom = int(bounds.get("bottom", 0))
            width = right - left
            height = bottom - top
            if width <= 0 or height <= 0:
                return None
            return {
                "left": left,
                "right": right,
                "top": top,
                "bottom": bottom,
                "width": width,
                "height": height,
            }
        except Exception:
            return None

    def _xpath_exists_meaningfully(self, d, xpath, min_width=24, min_height=12):
        bounds = self._get_xpath_bounds(d, xpath)
        if not bounds:
            return False
        return bounds["width"] >= min_width and bounds["height"] >= min_height

    def _get_active_verification_xpaths(self, d):
        active = []
        active_containers = []
        container_xpaths = [
            '//*[contains(@resource-id, "captcha")]',
            '//*[contains(@resource-id, "yoda")]',
            '//*[@resource-id="puzzleSliderBox"]',
            '//*[@resource-id="puzzleImageMain"]',
            '//*[@resource-id="yodaBoxWrapper"]',
            '//*[@resource-id="yodaBox"]',
            '//*[contains(@resource-id, "verify")]',
            '//*[contains(@resource-id, "slider")]',
            '//*[contains(@resource-id, "puzzle")]',
        ]
        for xpath in container_xpaths:
            try:
                if self._xpath_exists_meaningfully(d, xpath, min_width=80, min_height=40):
                    active_containers.append(xpath)
            except Exception:
                continue

        for xpath in self.popup_rules["verification"]:
            try:
                if self._xpath_exists_meaningfully(d, xpath):
                    active.append(xpath)
            except Exception:
                continue

        if active:
            return list(dict.fromkeys(active))

        if active_containers:
            now = time.time()
            if now - self.last_verification_probe_log_ts >= 8:
                self.logger.info(
                    "%s captcha container hit without explicit text rules: %s",
                    self._device_tag(),
                    active_containers[:3]
                )
                self.last_verification_probe_log_ts = now
            return list(dict.fromkeys(active_containers))

        fallback_xpaths = [
            '//*[contains(@text, "验证")]',
            '//*[contains(@text, "滑块")]',
            '//*[contains(@text, "拼图")]',
            '//*[contains(@text, "请点击")]',
        ]
        for xpath in fallback_xpaths:
            try:
                if self._xpath_exists_meaningfully(d, xpath):
                    active.append(xpath)
            except Exception:
                continue
        return list(dict.fromkeys(active))

    def _confirm_active_verification_xpaths(self, d, rounds=3, interval=0.35, min_hits=2):
        hit_counter = {}
        for idx in range(rounds):
            for xpath in self._get_active_verification_xpaths(d):
                hit_counter[xpath] = hit_counter.get(xpath, 0) + 1
            if idx < rounds - 1:
                time.sleep(interval)
        confirmed = [xpath for xpath, hits in hit_counter.items() if hits >= min_hits]
        return list(dict.fromkeys(confirmed))

    def _wait_verification_cleared(
            self,
            d,
            timeout=120,
            stable_rounds=3,
            interval=1.2,
            solve_retry_interval=6
    ):
        """
        必须连续 stable_rounds 次都检测不到验证码，才认为真正处理完成。
        """
        start = time.time()
        stable_count = 0
        last_active = []
        last_retry_solve_ts = 0
        while self.running and time.time() - start < timeout:
            active = self._get_active_verification_xpaths(d)
            if active:
                last_active = active
                stable_count = 0
                now = time.time()
                # 卡住时持续重试自动验证码处理，避免"只处理一次后一直挂起"
                if now - last_retry_solve_ts >= solve_retry_interval:
                    if self._try_auto_solve_verification(d):
                        self.logger.info("%s captcha auto-solver retried", self._device_tag())
                    last_retry_solve_ts = now
            else:
                stable_count += 1
                if stable_count >= stable_rounds:
                    return True, []
            time.sleep(interval)
        remaining = self._confirm_active_verification_xpaths(d, rounds=4, interval=0.4, min_hits=2)
        if not remaining:
            return True, []
        return False, remaining or last_active

    def _get_active_simple_popups(self, d):
        rules = self.popup_rules["simple"]
        if not rules:
            return []

        active = []
        for xpath, desc in rules:
            try:
                exists = bool(d.xpath(xpath).exists)
            except Exception:
                exists = False
            if exists:
                active.append((xpath, desc))
        return active

    def _handle_special_verify_pages(self, d):
        """定向处理验证码后的异常页，避免误点通用按钮。"""
        rules = [
            (
                '//*[@text="您的网络好像不太给力，请稍后再试"]',
                '//*[@text="重新加载"]',
                "网络不给力页-点击重新加载",
            ),
            (
                '//*[contains(@text, "verify.meituan.com/v2/app/general_page")]',
                '//*[@text="关闭页面"]',
                "general_page页-点击关闭页面",
            ),
        ]
        for detect_xpath, click_xpath, desc in rules:
            try:
                if d.xpath(detect_xpath).exists and d.xpath(click_xpath).exists:
                    self.logger.info("%s 检测到%s", self._device_tag(), desc)
                    d.xpath(click_xpath).click()
                    return True
            except Exception:
                continue
        return False

    def check_and_handle_popup(self):
        d = self.spider.d

        # 0. 处理验证码后的异常页（定向按钮）
        if self._handle_special_verify_pages(d):
            return True

        # 1. 处理简单弹窗
        for xpath, desc in self._get_active_simple_popups(d):
            if self._is_recent_click(xpath):
                continue
            try:
                self.logger.info("%s 检测到弹窗: %s", self._device_tag(), desc)
                d.xpath(xpath).click()
                return True
            except Exception:
                continue

        # 2. 处理验证码弹窗
        active_verification = self._confirm_active_verification_xpaths(d, rounds=3, interval=0.35, min_hits=2)
        if active_verification:
            if not self.pausing.is_set():
                # 记录验证码出现时间戳，统计1小时窗口内出现次数
                now = time.time()
                self.captcha_appearance_timestamps.append(now)
                cutoff = now - self.CAPTCHA_APPEARANCE_WINDOW
                self.captcha_appearance_timestamps = [
                    ts for ts in self.captcha_appearance_timestamps if ts > cutoff
                ]
                captcha_appearance_count = len(self.captcha_appearance_timestamps)

                self.last_verification_time = now
                self.verification_count += 1
                self.logger.warning(
                    "%s 验证码弹窗触发(%s)，1小时内第%s次，等待处理完成后再继续...",
                    self._device_tag(),
                    active_verification[:3],
                    captcha_appearance_count,
                )

                self.pausing.set()  # 通知主线程暂停
                if self._try_auto_solve_verification(d):
                    self.logger.info("%s captcha auto-solver triggered", self._device_tag())

            cleared, remaining = self._wait_verification_cleared(d)
            if cleared:
                self.logger.info("%s 验证码已处理，准备恢复任务", self._device_tag())
                self.pausing.clear()  # 放行主线程
            else:
                # 超时未清除：计数累加，主线程继续阻塞，等待下一轮处理或达到上限
                self.verification_count += 1
                self.logger.warning(
                    "%s 验证码处理超时(第%s次)，继续阻塞主线程: %s",
                    self._device_tag(),
                    self.verification_count,
                    remaining[:3]
                )

            # 1小时内验证码出现次数上限检查
            captcha_appearance_count = len(self.captcha_appearance_timestamps)
            if captcha_appearance_count >= self.MAX_CAPTCHA_APPEARANCES:
                self.logger.error(
                    "%s 1小时内验证码出现%s次，达到上限%s，终止任务",
                    self._device_tag(),
                    captcha_appearance_count,
                    self.MAX_CAPTCHA_APPEARANCES,
                )
                self.captcha_appearance_limit_reached = True
                self.pausing.clear()
                self.running = False
                return True

            # 重试上限检查（放在外面，超时累加计数后也能触发）
            if self.verification_count > self.MAX_VERIFICATION_RETRY:
                self.logger.error("%s 验证码重试超限(%s次)，终止任务", self._device_tag(), self.verification_count)
                self.pausing.clear()
                self.running = False
            return True

        # 3. 处理广告弹窗（点击右上角）
        if d.xpath('//*[contains(@text, "广告")]').exists:
            w, h = d.info['displayWidth'], d.info['displayHeight']
            d.click(w - 50, 50)
            self.logger.info("%s 关闭广告弹窗", self._device_tag())
            return True

        return False

    def stop(self):
        self.running = False


class MTScreenshot:
    def __init__(self, d, oss_config, search_key, title_key, scroll_times=2, compress_quality=7, resize_ratio=0.8,device_id=None,
                 monitor=None):
        self.device_id = device_id
        # 接收外部已连接好的u2设备实例
        self.d = d
        self.is_high_res = is_high_resolution_device(d)
        self.search_key = search_key  # 添加这行
        self.title_key = title_key
        # 优先复用外部监控，避免频繁创建监控线程
        self.monitor = monitor
        self.loggerMT = logging.getLogger()

        # 日志初始化
        self.logger = self._init_logger()

        # OSS配置与初始化（核心配置，无冗余）
        self.oss_config = oss_config
        self.oss_bucket = self._init_oss_bucket()

        # 截图核心参数
        self.scroll_times = scroll_times
        self.compress_quality = compress_quality
        self.resize_ratio = resize_ratio

    def _init_logger(self):
        # 极简日志配置，仅保留必要输出
        logger = logging.getLogger("mt_screenshot")
        logger.setLevel(logging.INFO)
        logger.handlers.clear()
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
        logger.addHandler(handler)
        return logger

    def _init_oss_bucket(self):
        # 仅做OSS配置校验和Bucket连接，无额外功能
        access_key_id = self.oss_config.get("access_key_id")
        access_key_secret = self.oss_config.get("access_key_secret")
        endpoint = self.oss_config.get("endpoint")
        bucket_name = self.oss_config.get("bucket_name")
        if not all([access_key_id, access_key_secret, endpoint, bucket_name]):
            self.logger.warning("OSS配置不完整，无法上传")
            return None

        # 进程内复用同一套配置的Bucket，避免重复连接
        cache_key = (access_key_id, access_key_secret, endpoint, bucket_name)
        with oss_bucket_cache_lock:
            cached_bucket = oss_bucket_cache.get(cache_key)
            if cached_bucket is not None:
                self.logger.info("复用已缓存的OSS Bucket连接")
                return cached_bucket

        try:
            auth = oss2.Auth(access_key_id, access_key_secret)
            bucket = oss2.Bucket(auth, endpoint, bucket_name)
            bucket.get_bucket_info()  # 验证连接
            with oss_bucket_cache_lock:
                oss_bucket_cache[cache_key] = bucket
            self.logger.info("OSS Bucket连接成功")
            return bucket
        except Exception as e:
            self.logger.error(f"OSS Bucket连接失败: {e}")
            return None

    def _upload_to_oss(self, local_path):
        # 极简上传逻辑，仅返回OSS URL或None
        if not self.oss_bucket or not os.path.exists(local_path):
            return None

        file_name = os.path.basename(local_path)
        safe_name = re.sub(r'[^\w\.\-]', '_', file_name)
        oss_key = f"{self.oss_config.get('oss_prefix', 'scrape_data/')}{safe_name}"

        try:
            oss2.resumable_upload(self.oss_bucket, oss_key, local_path)
            # 生成并返回完整OSS URL
            oss_file_url = f"https://{self.oss_config['bucket_name']}.{self.oss_config['endpoint']}/{urllib.parse.quote(oss_key, safe='/')}"
            return oss_file_url
        except Exception as e:
            self.logger.error(f"OSS上传失败: {e}")
            return None

    def safe_exec(self, func, *args, **kwargs):
        """
        万能安全壳：执行 func 前检查验证码，
        若监控线程已置位 pausing，则一直阻塞直到放行。
        """
        while self.monitor is not None and self.monitor.pausing.is_set():
            time.sleep(1)
        # 执行真正逻辑

        return func(*args, **kwargs)

    def _get_title(self):
        # try:

        def _inner():

            print(f'获取商品title时的搜索关键字:{self.title_key}')

            # 初始化
            drugs_name = ''
            specifications = ''
            title = ''

            # 循环的获取title为了有时间来处理人机验证
            for m in range(1, 6000):
                if self.d.xpath(f'//*[contains(@text, "{self.title_key}")]').exists:
                    title = self.safe_exec(
                        lambda: self.d.xpath(f'//*[contains(@text, "{self.title_key}")]').text
                    )
                    self.loggerMT.info(f"第{m}次获取title成功")
                    print(f"第{m}次获取title成功")
                    break
                else:
                    time.sleep(1)
                    # return drugs_name, specifications
            title = title[1:] if title.startswith('0') else title

            print(f'获取到药品标题:{title}')

            match = re.match(r'(\[[^\]]+\])(.*?)\s*((?:\d+\S*|\(.+))$', title)

            if match:
                drugs_name = title
                specifications = match.group(3).strip()
                print("药品名:", drugs_name)
                print("规格:", specifications)
                # print('完整药名:', drugs_name + specifications)
                return drugs_name  # , specifications
            else:
                drugs_name = title
                specifications = ''
                return drugs_name

        # 用 safe_exec 包装内部逻辑，确保验证码阻塞
        return self.safe_exec(_inner)

    def _merge_screenshots(self, screens):
        # 仅拼接截图，无额外功能
        if len(screens) == 1:
            return screens[0].convert('RGB')

        rgb_screens = [s.convert('RGB') for s in screens]
        total_width = rgb_screens[0].width
        total_height = sum(s.height for s in rgb_screens)

        merged_img = Image.new('RGB', (total_width, total_height))
        y_offset = 0
        for img in rgb_screens:
            merged_img.paste(img, (0, y_offset))
            y_offset += img.height
        return merged_img

    def get_oss_url(self, title=None):
        """核心方法：截图+临时本地保存+上传OSS+上传成功删本地文件+返回OSS URL，可直接赋值给oss_file"""
        local_file_path = None
        try:
            # 1. 优先使用外部已采集标题，避免重复读取页面标题
            title = str(title or "").strip()
            if not title:
                title = self._get_title()
                self.logger.info(f"获取标题: {title[:20]}..." if title else "未获取到标题")
            else:
                self.logger.info(f"使用外部标题: {title[:20]}...")

            # 2. 生成本地文件路径
            timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + '_mt_' + self.device_id
            safe_title = re.sub(r'[\\/*?:"<>|]', '_', title)
            local_dir = "../scrape_data"
            os.makedirs(local_dir, exist_ok=True)
            local_file_path = os.path.join(local_dir, f"{timestamp}_{safe_title}.jpg")

            # 3. 滚动截图
            screen_list = [self.d.screenshot()]
            w, h = self.d.window_size()

            for i in range(self.scroll_times):
                # 可能滑动距离太短，截不到店名。原本是0.8
                # self.d.swipe(w // 2, h * 0.9, w // 2, h * 0.1, duration=random.uniform(0.6, 1.2))
                if self.is_high_res:
                    self.d.swipe(w // 2, h * 0.75, w // 2, h * 0.25, duration=random.uniform(0.8, 1.5))
                else:
                    self.d.swipe(w // 2, h * 0.85, w // 2, h * 0.15, duration=random.uniform(0.8, 1.5))
                time.sleep(random.uniform(2.0, 4.0))
                screen_list.append(self.d.screenshot())

                if self.d(textContains='商家服务').exists:
                    # 看情况是否需要补滑
                    break

            # 4. 拼接+压缩+保存
            merged_img = self._merge_screenshots(screen_list)
            if 0.1 < self.resize_ratio < 1.0:
                new_size = (int(merged_img.width * self.resize_ratio), int(merged_img.height * self.resize_ratio))
                resample_mode = Image.Resampling.LANCZOS if hasattr(Image, 'Resampling') else Image.LANCZOS
                merged_img = merged_img.resize(new_size, resample_mode)

            # 临时保存到本地
            merged_img.save(local_file_path, format='JPEG', quality=self.compress_quality)
            merged_img.close()  # 释放长图句柄

            # 5. 上传OSS
            oss_url = self._upload_to_oss(local_file_path)

            # 6. 核心：OSS上传成功后，删除本地临时文件
            if oss_url is not None:
                try:
                    self.logger.info(f"✅ OSS上传成功，已删除本地临时文件: {local_file_path}")
                except Exception as e:
                    self.logger.warning(f"⚠️ OSS上传成功，但删除本地文件失败: {e}")
            return oss_url


        except Exception as e:
            self.logger.error(f"截图/上传失败: {e}")
            return None


class MT:
    # 封号检测：类变量，跨实例共享，不因重建实例而重置
    open_page_fail_count = 0
    open_page_fail_threshold = 6

    def __init__(
            self,
            key,
            title_key,
            spec_list,
            brand,
            sort=None,
            collect_range=None,
            page_range=None,
            workflow_retry_limit=None,
            workflow_error_action=None,
            platform=None,
            task_id=None,
            enterprise_id=None,
            sampling_cycle=None,
            sampling_start_time=None,
            sampling_end_time=None,
            count=None,
            collect_equipment_id=None,
            device_name=None,
            collect_equipment_account_id=None,
            collect_region_id=None,
            collect_round=None,
            scheduler=None,
    ):
        self.scheduler = scheduler
        self.package_name = Config.PACKAGE_NAME
        self.access_token = get_access_token()
        self.APP_ID = '116857964'
        self.API_KEY = '1gAzACJOAr7BeILKqkqPOETh'
        self.SECRET_KEY = 'ZNArANb9GwJYgLKg4EfYhukKBfPdl1n3'
        self.client = AipOcr(self.APP_ID, self.API_KEY, self.SECRET_KEY)

        self.area_service = AreaService("city.json", "addr_prefix.json")
        self.table_name = "retrieve_scrape_data"
        self.shop_table_name = "retrieve_scrape_shop_info"
        self.loggerMT = logging.getLogger()
        self.logger = self.loggerMT
        self.task_id = task_id
        self.enterprise_id = enterprise_id
        self.platform = platform
        self.collect_equipment_id = collect_equipment_id
        self.device_name = device_name
        self.page = 0
        self.search_key = str(key or "").strip()
        self.title_key = str(title_key or "").strip()
        self.spec_list = [str(spec).strip() for spec in (spec_list or []) if str(spec).strip()]
        self.brand = str(brand or "").strip()
        self.sort = sort
        self.collect_equipment_account_id = collect_equipment_account_id
        self.collect_region_id = collect_region_id
        self.collect_round = collect_round
        self.sampling_cycle = sampling_cycle
        self.sampling_start_time = sampling_start_time
        self.sampling_end_time = sampling_end_time
        # self.count = parse_optional_int(count, None)
        self.count = 200
        if self.count is not None and self.count <= 0:
            self.count = None
        self.loggerMT.info(
            "[MT初始化] task_id=%s device_id=%s collect_equipment_id=%s platform=%s enterprise_id=%s "
            "sampling_cycle=%s sampling_start_time=%s sampling_end_time=%s target_count=%s",
            self.task_id,
            getattr(self, "device_id", None),
            self.collect_equipment_id,
            self.platform,
            self.enterprise_id,
            self.sampling_cycle,
            self.sampling_start_time,
            self.sampling_end_time,
            self.count,
        )

        self.collect_range = self.normalize_collect_range(collect_range)
        self.page_range = self.normalize_page_range(page_range)
        self.sort_key = 0
        self.unrelated_data = 0
        self.shop_data_num = 0
        self.collected_data_count = 0
        self.collected_count_lock = threading.Lock()
        self.target_count_reached_event = threading.Event()
        self.max_unrelated_data = 15
        self.request_error_count = 0
        self.request_error_threshold = REQUEST_ERROR_STOP_THRESHOLD
        self.product_link_missing_count = 0
        self.product_link_missing_threshold = PRODUCT_LINK_STOP_THRESHOLD
        # 封号检测：open_product_list_page 连续失败计数（类变量，跨实例共享，不因重建实例而重置）
        # 风控卡死检测："加载更多"按钮检测
        self.load_more_check_rounds = 4
        self.load_more_check_interval = 30
        self.app_closed = False
        self.collection_cursor = {"page_no": 1, "item_index": 0}
        self.workflow_retry_limit = workflow_retry_limit or {
            "start_app": 3,
            "open_product_list_page": 3,
            "collect_single_product": 3,
        }
        self.workflow_error_action = workflow_error_action or {
            "start_app": "start_app",
            "open_product_list_page": "start_app",
            "collect_single_product": "back_to_list_page",
        }
        self.is_high_res = False  # 连接设备后更新
        self.finish_reported = False
        self.post_process_executor = ThreadPoolExecutor(max_workers=4)
        self.post_process_futures = []
        self.post_process_lock = threading.Lock()
        self.max_pending_post_tasks = 80

    def get_collected_data_count(self):
        with self.collected_count_lock:
            return self.collected_data_count

    def has_reached_target_count(self):
        if self.count is None:
            return False
        if self.target_count_reached_event.is_set():
            return True
        return self.get_collected_data_count() >= self.count

    def mark_collected_data_saved(self):
        with self.collected_count_lock:
            self.collected_data_count += 1
            current_count = self.collected_data_count
        if self.count is not None and current_count >= self.count:
            self.target_count_reached_event.set()
        return current_count

    def finish_task_normally(self, end_page, reason):
        if not self.finish_reported and self.task_id:
            if self.scheduler:
                self.scheduler.post_report({
                    "task_id": self.task_id,
                    "platform": str(self.platform),
                    "username": self.scheduler.username,
                    "current_page": end_page,
                    "crawled_count": self.get_collected_data_count(),
                    "is_finished": 1,
                })
            self.finish_reported = True
        self.wr_re("删", self.device_id)
        print(reason)
        self.close()
        return True

    def finish_task_abnormally(self, end_page, reason, finish_status=0):
        if not self.finish_reported and self.task_id:
            self.scheduler.post_report({
                "task_id": self.task_id,
                "platform": str(self.platform),
                "username": self.scheduler.username,
                "current_page": end_page,
                "crawled_count": self.get_collected_data_count(),
                "is_finished": 0,
                "need_reassign": 1,
                "exception_type": 5,
                "remark": reason,
            })
            self.finish_reported = True
        print(reason)
        self.close()
        return False

    def _progress_file_path(self, device_id=None):
        target_device = device_id or getattr(self, "device_id", None)
        if not target_device:
            return None
        return f'./ycwj/{target_device}_{self.title_key}.txt'

    @staticmethod
    def normalize_collect_range(collect_range):
        if not collect_range:
            return None
        start = None
        end = None

        if isinstance(collect_range, dict):
            start = collect_range.get("start")
            end = collect_range.get("end")
        elif isinstance(collect_range, (list, tuple)) and len(collect_range) >= 2:
            start, end = collect_range[0], collect_range[1]
        elif isinstance(collect_range, str):
            matched = re.match(r"^\s*(\d+(?:\.\d+)?)\s*[-,~]\s*(\d+(?:\.\d+)?)\s*$", collect_range)
            if matched:
                start, end = matched.group(1), matched.group(2)

        try:
            start = float(start)
            end = float(end)
        except (TypeError, ValueError):
            return None

        if start < 0 or end < 0:
            return None
        if start > end:
            start, end = end, start
        return {"start": start, "end": end}

    @staticmethod
    def normalize_page_range(page_range):
        if not page_range:
            return None

        start = None
        end = None

        if isinstance(page_range, dict):
            start = page_range.get("start")
            end = page_range.get("end")
        elif isinstance(page_range, (list, tuple)) and len(page_range) >= 2:
            start, end = page_range[0], page_range[1]
        elif isinstance(page_range, str):
            matched = re.match(r"^\s*[\[\(]?\s*(\d+)\s*[,，\-~]\s*(\d+)\s*[\]\)]?\s*$", page_range)
            if matched:
                start, end = matched.group(1), matched.group(2)

        try:
            start = int(float(start))
            end = int(float(end))
        except (TypeError, ValueError):
            return None

        if start <= 0 or end <= 0:
            return None
        if start > end:
            start, end = end, start
        return {"start": start, "end": end}

    def stop_app(self):
        if getattr(self, "d", None) is None or self.app_closed:
            return
        try:
            self.d.app_stop(self.package_name)
            self.app_closed = True
        except Exception as e:
            self.loggerMT.warning(f"关闭应用失败: {e}")
        time.sleep(1)

    def start_app(self):
        self.d.app_start(self.package_name)
        self.app_closed = False
        time.sleep(1)

    def restart_app(self):
        self.stop_app()
        self.start_app()

    def close(self):
        self.stop_app()

    def li_or_lo(self, key="升序"):
        """
        排序操作：升序或降序
        :param key: "升序" 或 "降序"
        """
        if key == "升序":
            # 增加重试机制，最多尝试3次
            max_retries = 3
            for attempt in range(max_retries):
                try:
                    # 1. 点击“综合”标签
                    # 优先使用text定位，若失败可尝试其他属性
                    comprehensive = self.d.xpath('//*[@text="综合"]')
                    if comprehensive.exists:
                        comprehensive.click(timeout=3)  # 等待元素可点击
                        time.sleep(0.5)

                    # 2. 点击“总价低到高”选项
                    low_to_high = self.d.xpath('//*[@text="总价低到高"]')
                    if low_to_high.exists:
                        low_to_high.click(timeout=3)
                        time.sleep(0.7)
                        self.sort_key += 1
                        self.logger.info("排序已切换为升序（总价低到高）")
                        return  # 成功执行后退出

                except Exception as e:
                    self.logger.warning(f"升序排序尝试 {attempt+1}/{max_retries} 失败: {e}")
                    # 如果失败，等待并重试
                    time.sleep(1)
                    # 若最后一次重试仍失败，抛出异常
                    if attempt == max_retries - 1:
                        self.logger.error("升序排序最终失败，请检查UI元素")
                        raise

        elif key == "降序":
            # 当前版本不支持降序，记录日志但不执行操作
            self.logger.warning('2026_05版本不存在降序，请勿调用')
            print('2026_05版本不存在降序，请勿调用')
            # 可根据需要选择抛出异常或直接返回
            # raise NotImplementedError("降序功能未实现")
        else:
            raise ValueError(f"无效的排序参数: {key}，仅支持 '升序' 或 '降序'")


    def wr_re(self, mod, device_id, sort=None, page=None):
        file_path = self._progress_file_path(device_id)
        if not file_path:
            return None
        if mod == "写":
            try:
                data = {
                    "page": page if page else "",
                    "sort": sort if sort else "",
                }
                os.makedirs(os.path.dirname(file_path), exist_ok=True)
                with open(file_path, 'w', encoding='utf-8') as f:
                    json.dump(data, f, ensure_ascii=False, indent=2)
                print(f"进度保存成功:{sort},{page}页")
            except Exception as e:
                print("保存进度失败")
        elif mod == "读":
            self.li_or_lo()
            try:
                if not os.path.exists(file_path):
                    return None
                with open(file_path, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    i = 0
                    while True:
                        self.wait_for_ready(getattr(self, "monitor", None))
                        if i == data['page']:
                            self.page = data['page']
                            print("当前页", self.page)
                            break
                        else:
                            i += 1
                            if self.is_high_res:
                                self.d.drag(300, 2600, 300, 400, 1)
                            else:
                                self.d.drag(300, 1400, 300, 400, 1)
                return data
            except Exception as e:
                print(f"读取进度失败")
                return None
        elif mod == "删":
            try:
                if os.path.exists(file_path):
                    os.remove(file_path)
                    print(f"进度文件已删除:{file_path}")
                return True
            except Exception as e:
                print(f"删除进度文件失败: {e}")
                return False
        return None

    # 任何一个spec满足都算有效
    def is_link_spec_useful(self, product_title):
        normalized_title = normalize_match_text(product_title)
        if len(self.spec_list) == 0:
            return True
        for spec in self.spec_list:
            if normalize_match_text(spec) in normalized_title:
                return True
        return False

    # TODO 继续优化这里的判断逻辑，可以考虑搭配config的修改
    def is_link_useful(self, product_title):
        normalized_title = normalize_match_text(product_title)
        normalized_title_key = normalize_match_text(self.title_key)
        normalized_brand = normalize_match_text(self.brand)

        # 1. 药品名必须在标题里
        if normalized_title_key != "" and normalized_title_key not in normalized_title:
            print(f"当前商品名称:{product_title} 不包含{self.title_key}关键字")
            return False

        # 2. 把药品名从标题里去掉后再检查品牌
        #    （避免品牌是通用名子串的干扰，如品牌"午时" ⊂ 药名"午时茶颗粒"）
        check_target = normalized_title.replace(normalized_title_key, "") if normalized_title_key else normalized_title
        if normalized_brand != "" and normalized_brand not in check_target:
            print(f"当前商品名称:{product_title} 去掉药名后不包含{self.brand}品牌")
            return False

        if not self.is_link_spec_useful(product_title):
            print(f"当前商品名称:{product_title} 不包含{self.spec_list}品规")
            return False
        return True

    @staticmethod
    def get_sleep_time():
        # return random.randint(5, 8)
        # return 1
        return random.uniform(0.5, 1.0)

    @staticmethod
    def get_current_date():
        return datetime.datetime.now().strftime('%Y/%m/%d')

    def get_shop_name_from_current_page(self):
        """
        仅从当前商品详情页读取店铺名，不做任何页面跳转。
        """
        shop_name = self.get_first_text_by_xpaths([
            '//android.widget.ScrollView/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.widget.FrameLayout[1]/android.widget.TextView',
            '//android.widget.ScrollView/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()-1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.widget.FrameLayout[1]/android.widget.TextView',
        ])
        if shop_name:
            print(f'获取到店铺名:{shop_name}')
        return shop_name

    def get_shop_name(self):
        """
        获取店铺名
        :return:
        """
        shop_name = self.get_shop_name_from_current_page()
        if shop_name:
            return shop_name

        try:
            # 点击店铺进入后获取店铺名称
            print("点击店铺进入后获取店铺名称")
            self.enter_shop()
            shop_xpath = '//*[@resource-id="com.sankuai.meituan:id/layout_header_view"]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]//android.widget.FrameLayout[2]/android.widget.FrameLayout[1]/android.widget.TextView'
            if self.d.xpath(shop_xpath).exists:
                shop_name = self.d.xpath(shop_xpath).text
                self.swipe_back(1)
                return shop_name
            shop_name = ''
            return shop_name
        except Exception as e:
            print(f'获取店铺名出错:{e}')
            return ''

    def get_qualification_number(self):
        """
        获取资质编号
        :return:
        """
        try:
            # 方法1：精准 XPath
            elem = self.d.xpath(
                '//*[@resource-id="com.sankuai.meituan:id/mil_container"]/android.webkit.WebView[1]/android.webkit.WebView[1]/android.view.View[1]/android.view.View[1]/android.widget.TextView[2]')
            if elem.exists:
                text = elem.text
                if text:
                    return text.replace('资质编号：', '').strip()

            # 方法2：模糊匹配任意包含"资质编号"的 TextView
            elem2 = self.d.xpath('//android.widget.TextView[contains(@text, "资质编号")]')
            if elem2.exists:
                text = elem2.text
                if text:
                    return text.replace('资质编号：', '').strip()

            # 方法3：更通用的包含匹配（不限 TextView）
            elem3 = self.d.xpath('//*[contains(@text, "资质编号")]')
            if elem3.exists:
                text = elem3.text
                if text:
                    return text.replace('资质编号：', '').strip()

            return None
        except Exception as e:
            print(f"获取资质编号失败: {e}")
            return None

    def get_shop_address(self):
        try:
            shop_address_xpaths = [
                '//*[@resource-id="com.sankuai.meituan:id/wm_sc_drug_shop_content_mrn_container_id_2"]/android.widget.FrameLayout/android.widget.FrameLayout/android.view.ViewGroup/android.view.ViewGroup/android.widget.ScrollView/android.view.ViewGroup/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.TextView',
                '//*[@resource-id="com.sankuai.meituan:id/wm_sc_drug_shop_content_mrn_container_id_2"]/android.widget.FrameLayout/android.widget.FrameLayout/android.view.ViewGroup/android.view.ViewGroup/android.widget.ScrollView/android.view.ViewGroup/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.TextView'
            ]
            shop_address = self.get_first_text_by_xpaths(shop_address_xpaths)
            print(f'111-获取到店铺地址:{shop_address}')
            if '发货时间' in shop_address:
                print(f'店铺地址包含发货时间，再次获取店铺地址')
                shop_address = self._read_xpath_text(shop_address_xpaths[1])
                if shop_address:
                    print(f'222-获取到店铺地址:{shop_address}')
                else:
                    print(f'222-xpath2获取店铺地址失败')
            print(f'333-获取到店铺地址:{shop_address}')
            if "近30天平均发货" not in shop_address :
                return shop_address
            else:

                return ""
        except:
            print(f'获取店铺地址出错-get_shop_address')
            return None

    def execute_db_write(self, sql, params, action_desc, max_retries=5):
        for attempt in range(max_retries):
            conn = None
            try:
                conn = get_mysql()
                with conn.cursor() as cur:
                    cur.execute(sql, params)
                conn.commit()
                print(f"{action_desc}成功")
                return True
            except Exception as e:
                print(f'{action_desc}异常 (尝试 {attempt + 1}/{max_retries}): {e}')
                if conn:
                    conn.rollback()
                if attempt == max_retries - 1:
                    print(f"{action_desc}失败，达到最大重试次数")
                    return False
                time.sleep(2)
            finally:
                if conn:
                    conn.close()

    def query_exists(self, sql, params, error_desc):
        conn = None
        try:
            conn = get_mysql()
            with conn.cursor() as cur:
                cur.execute(sql, params)
                return bool(cur.fetchone())
        except Exception as e:
            print(f"{error_desc}错误: {str(e)}")
            return None
        finally:
            if conn:
                conn.close()

    def save_to_database(self, data):
        add_sql = f"""
            INSERT IGNORE INTO {self.table_name} (
                enterprise_id, platform_id, platform_item_id, province_id, city_id,
                province_name, city_name, area_info, product_name, product_specs,
                one_box_price, manufacture_date, expiry_date, manufacturer, approval_number,
                is_sold_out, online_posting_count, continuous_listing_count, link_url,
                store_name, store_url, shipment_province_id, shipment_province_name,
                shipment_city_id, shipment_city_name, company_name, qualification_number,
                scrape_date, min_price, number, sales, inventory, snapshot_url,
                product_brand, search_name, insert_time ,update_time, collect_config_info,
                collect_equipment_account_id ,collect_region_id ,collect_round,
                shop_id, company_id, task_id
            ) VALUES (
                %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s,
                %s, %s, %s, %s,
                %s, %s, %s, %s,
                %s, %s, %s, %s,
                %s, %s, %s, %s, %s, %s,
                %s, %s, %s, %s, %s, %s, %s,
                %s,
                %s, %s,
                %s
            )
        """
        store_name = data.get('store_name', '')
        params = (
            data['enterprise_id'],
            data['platform_id'],
            data['platform_item_id'],
            data['province_id'],
            data['city_id'],
            data['province_name'],
            data['city_name'],
            data['area_info'],
            data['product_name'],
            data['product_specs'],
            data['one_box_price'],
            data['manufacture_date'],
            data['expiry_date'],
            data['manufacturer'],
            data['approval_number'],
            data['is_sold_out'],
            data['online_posting_count'],
            data['continuous_listing_count'],
            data['link_url'],
            store_name,
            data['store_url'],
            data['shipment_province_id'],
            data['shipment_province_name'],
            data['shipment_city_id'],
            data['shipment_city_name'],
            data['company_name'],
            data['qualification_number'],
            data['scrape_date'],
            data['min_price'],
            data['number'],
            data['sales'],
            data['inventory'],
            data['snapshot_url'],
            data['product_brand'],
            data['search_name'],
            data['insert_time'],
            data['update_time'],
            data['collect_config_info'],
            data['collect_equipment_account_id'],
            data['collect_region_id'],
            data['collect_round'],
            store_name,   # shop_id = store_name
            store_name,   # company_id = store_name
            data.get('task_id'),
        )
        return self.execute_db_write(add_sql, params, "保存商品数据到数据库")

    def save_shop_info_to_database(self, data):
        print(f'保存店铺数据到数据库:{data}')
        now_str = time.strftime('%Y-%m-%d %H:%M:%S')

        def _clean(v):
            v = str(v or '').strip()
            return '' if v in ('无', '無') else v

        shop = str(data.get('shop') or '').strip()
        contact_address = _clean(data.get('contact_address') or data.get('business_license_address'))
        business_license_address = _clean(data.get('business_license_address') or contact_address)
        qualification_number = _clean(data.get('qualification_number'))
        business_license_company = _clean(data.get('business_license_company'))
        scrape_date = str(data.get('scrape_date') or self.get_current_date()).strip()
        platform = str(data.get('platform') or self.platform or '4').strip()
        province = str(data.get('province') or '').strip()
        city = str(data.get('city') or '').strip()
        create_time = str(data.get('create_time') or now_str).strip()
        update_time = str(data.get('update_time') or now_str).strip()

        if self.shop_is_exists_database(shop, platform):
            if ENABLE_SHOP_DEBUG:
                print(f"[SHOP-DEBUG] save_shop_info_to_database: 命中UPDATE, shop={shop}")
            existing = self.get_shop_info_from_database(shop, platform) or {}
            existing_contact = str(existing.get("contact_address") or "").strip()
            existing_biz_address = str(existing.get("business_license_address") or "").strip()
            existing_company = str(existing.get("business_license_company") or "").strip()
            existing_qn = str(existing.get("qualification_number") or "").strip()
            existing_province = str(existing.get("province") or "").strip()
            existing_city = str(existing.get("city") or "").strip()

            contact_address = contact_address or existing_contact or existing_biz_address
            business_license_address = business_license_address or existing_biz_address or existing_contact
            business_license_company = business_license_company or existing_company
            qualification_number = qualification_number or existing_qn
            province = province or existing_province
            city = city or existing_city
            if (not province) and (not city):
                match = self.area_service.search_area(
                    business_license_address or contact_address)
                if match:
                    province = match.province or province
                    city = match.city or city
            if ENABLE_SHOP_DEBUG:
                print(
                    f"[SHOP-DEBUG] update payload: shop={shop}, "
                    f"company={business_license_company}, qn={qualification_number}, "
                    f"contact_address={contact_address}, biz_address={business_license_address}, "
                    f"province={province}, city={city}, platform={platform}"
                )

            update_sql = f"""
                UPDATE {self.shop_table_name}
                SET contact_address = %s,
                    qualification_number = %s,
                    business_license_company = %s,
                    business_license_address = %s,
                    scrape_date = %s,
                    platform = %s,
                    province = %s,
                    city = %s,
                    update_time = %s,
                    shop_id = %s,
                    company_id = %s
                WHERE shop = %s
            """
            update_params = (
                contact_address,
                qualification_number,
                business_license_company,
                business_license_address,
                scrape_date,
                platform,
                province,
                city,
                update_time,
                shop,   # shop_id = shop
                shop,   # company_id = shop
                shop
            )
            return self.execute_db_write(update_sql, update_params, "更新店铺数据到数据库")

        if ENABLE_SHOP_DEBUG:
            print(f"[SHOP-DEBUG] save_shop_info_to_database: 命中INSERT, shop={shop}")
        if ENABLE_SHOP_DEBUG:
            print(
                f"[SHOP-DEBUG] insert payload: shop={shop}, "
                f"company={business_license_company}, qn={qualification_number}, "
                f"contact_address={contact_address}, biz_address={business_license_address}, "
                f"province={province}, city={city}, platform={platform}"
            )
        add_sql = f"""
            INSERT INTO {self.shop_table_name}
            (shop, contact_address, qualification_number, business_license_company, business_license_address, scrape_date, platform, province, city, create_time, update_time, shop_id, company_id)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        """
        add_params = (
            shop,
            contact_address,
            qualification_number,
            business_license_company,
            business_license_address,
            scrape_date,
            platform,
            province,
            city,
            create_time,
            update_time,
            shop,   # shop_id = shop
            shop,   # company_id = shop
        )
        return self.execute_db_write(add_sql, add_params, "保存店铺数据到数据库")

    def swipe_back(self, no):
        """
        返回
        :param no: 回退次数
        :return:
        """
        for idx in range(no):
            self.d.press('back')
            time.sleep(0.5)

    def drug_price(self):
        """
        获取药品价格
        :return:
        """
        time.sleep(0.5)
        try:
            price_str = ""
            price_int_xpath = '//*[@text="¥"]/../../android.widget.FrameLayout[2]/android.widget.TextView[1]'
            price_decimal_xpath = '//*[@text="¥"]/../../android.widget.FrameLayout[3]/android.widget.TextView[1]'

            price_int_node = self.d.xpath(price_int_xpath)
            if price_int_node.exists and price_int_node.text:
                price_str = price_int_node.text.strip()

            price_decimal_node = self.d.xpath(price_decimal_xpath)
            if price_decimal_node.exists and price_decimal_node.text:
                price_str += price_decimal_node.text.strip()

            if not price_str:
                print('提取价格出错-->未获取到价格文本')
                return None
            # if self.d.xpath('//*[@text="优惠"]').exists:
            #     self.d.xpath('//*[@text="优惠"]').click()
            #     time.sleep(0.5)
            #     if self.d.xpath('//*[contains(@text, "现在购买") and contains(@text, "享受以下优惠") and contains(@text, "共省")]').exists:
            #         match = re.search(r'共省¥([\d.]+)', self.d.xpath('//*[contains(@text, "现在购买") and contains(@text, "享受以下优惠") and contains(@text, "共省")]').text)
            #         self.d.press("back")
            #         if match:
            #             save_amount = match.group(1)
            #             print(f"优惠金额: {save_amount}")
            #         price = float(Decimal(str(price_str)) + Decimal(str(save_amount)))
            # if not price:
            #     price = float(price_str)
            price = float(price_str)
            print(f'获取到价格: {price}')
            return price
        except Exception as e:
            print(f'提取价格出错-->{e}')
            return None

    def drug_sale_num(self):
        """
        获取药品销量
        :return:
        """
        try:
            sales_element = self.d.xpath('//*[starts-with(@text,"已售")]')

            if sales_element.exists:
                sales_num_str = self.d.xpath('//*[starts-with(@text,"已售")]').text
                sales_num_str = sales_num_str.replace("已售", "").strip()
                # price = float(re.search(r'[\d\.]+', price_str).group())
                print(f'获取到已售数量:{sales_num_str}')
                return sales_num_str
            return None
        except Exception as e:
            print(f'提取已售数量出错-->{e}')
            return None

    def restart_uiautomator_services(self, device_id):
        """
        重启atx的uiautomator 服务
        :param device_id:
        :return:
        """
        stop_uiautomator_services = f'adb -s {device_id} shell /data/local/tmp/atx-agent server -d --stop'
        start_uiautomator_services = f'adb -s {device_id} shell /data/local/tmp/atx-agent server -d'
        subprocess.run(stop_uiautomator_services, capture_output=True, text=True, shell=True)
        time.sleep(self.get_sleep_time())
        subprocess.run(start_uiautomator_services, capture_output=True, text=True, shell=True)
        time.sleep(self.get_sleep_time())

    def reconnect_device(self):
        """重启 atx-agent 并重新连接设备"""
        try:
            # 停止 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.d = u2.connect_usb(self.device_id)
            self.is_high_res = is_high_resolution_device(self.d)
            self.restart_uiautomator_services(self.device_id)
            self.loggerMT.info("设备重连成功")
            return True
        except Exception as e:
            self.loggerMT.error(f"设备重连失败: {e}")
            return False

    def connect_devices(self, device_id):
        """
        连接设备
        :return:
        """
        try:
            self.device_id = device_id
            self.d = u2.connect_usb(device_id)
            self.is_high_res = is_high_resolution_device(self.d)
            self.restart_uiautomator_services(device_id)
            self.oss_config = {
                "access_key_id": 'LTAI5t5pWgfa1BMztEuWBjdK',
                "access_key_secret": 'wU7FLzEr1NqLg2rJrmAu7Ibn69np0u',
                "endpoint": "oss-cn-shenzhen.aliyuncs.com",  # 例：oss-cn-beijing.aliyuncs.com
                "bucket_name": "zhijiayun-jiansuo",
                "oss_prefix": "scrape_data/"  # OSS中存放截图的前缀（虚拟文件夹）
            }
            print(f'连接到设备:{device_id}')
            self.loggerMT.info(f'连接到设备:{device_id}')
        except Exception as e:
            print(f'{device_id} 连接错误: {e}')
            self.loggerMT.info(f'{device_id} 连接错误: {e}')
            raise Exception(e)

    def get_ocr_res(self, img):
        try:
            # img地址
            print(f'开始识别图片:{img}')
            request_url = "https://aip.baidubce.com/rest/2.0/ocr/v1/business_license"
            f = open(img, 'rb')
            img = base64.b64encode(f.read())
            params = {"image": img}
            request_url = request_url + "?access_token=" + self.access_token
            headers = {'content-type': 'application/x-www-form-urlencoded'}
            response = requests.post(request_url, data=params, headers=headers)
            if response:
                res = response.json()
                new_dic = dict()
                for ite in res['words_result'].keys():
                    new_dic[ite] = res['words_result'][ite]['words']
                print('资质数据信息', new_dic)
                return new_dic
            else:
                return None
        except:
            return None

    def remove_watermark(self, img_path):
        """
        图片去水印(将水印部分变成白色背景)并将数据转化为二进制数据
        :param img_path: 图片路径
        :return: 二进制图片数据
        """
        img = cv2.imdecode(np.fromfile(img_path, dtype=np.uint8), -1)
        endswith = os.path.splitext(img_path)[1]
        new = np.clip(1.4057577998008846 * img - 38.33089999653017, 0, 255).astype(np.uint8)
        _, img_binary = cv2.imencode(endswith, new)
        return img_binary

    def get_ocr_res_image(self, img):
        try:
            image = self.remove_watermark(img)
            # image_file = open(img,'wb')
            # image_file.write(images)
            # res_image = self.client.basicAccurate(images)  # 高精度
            res_image = self.client.basicGeneral(image)
            data = res_image.get('words_result', '')
            print(f'百度api返回结果:{data}')

            return data
        except:
            return None

    def write_ocr_result_sidecar(self, image_path, ocr_res, source="business_license", extra=None):
        """
        将 OCR 结果写到与图片同名的 sidecar 文件，便于逐图对照识别结果。
        """
        try:
            if not image_path:
                return ''
            image_path = Path(str(image_path))
            sidecar_path = image_path.with_suffix('.ocr.json')
            payload = {
                "image_path": str(image_path),
                "source": source,
                "success": bool(ocr_res),
                "ocr_result": ocr_res if isinstance(ocr_res, (dict, list)) else {},
                "extra": extra or {},
                "created_at": time.strftime('%Y-%m-%d %H:%M:%S')
            }
            sidecar_path.parent.mkdir(parents=True, exist_ok=True)
            with open(sidecar_path, 'w', encoding='utf-8') as f:
                json.dump(payload, f, ensure_ascii=False, indent=2)
            print(f'OCR结果已写入:{sidecar_path}')
            return str(sidecar_path)
        except Exception as e:
            print(f'写OCR结果文件失败:{e}')
            return ''

    def screenshot_the_business_license(self, qualification_number):
        screenshot_path = 'screenshot1.png'
        self.d.screenshot(screenshot_path)
        img = cv2.imread(screenshot_path)
        # 指定裁剪区域 (left, top, right, bottom)
        left = 0
        top = 1026
        right = 1220
        bottom = 1904
        cropped_img = img[top:bottom, left:right]

        # 创建目录
        SCREENSHOT_DIR = Path('screenshot')  # 注意这里的变化和py文件同一级目录即可
        SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True)

        if qualification_number:
            # cropped_screenshot_path = 'D:\\work\\dfwy_spider\\drug_data\\mt\\screenshot\\' + qualification_number + '.png'
            cropped_screenshot_path = SCREENSHOT_DIR / f'{qualification_number}.png'
        else:
            cropped_screenshot_path = 'cropped_screenshot.png'
        cv2.imwrite(str(cropped_screenshot_path), cropped_img)
        return cropped_screenshot_path

    def screenshot_instruction(self):
        # 获取当前时间
        current_time = datetime.datetime.now()

        # 格式化为时分秒
        time_str = current_time.strftime("%H-%M-%S")

        # 生成随机的 8 位字符串
        random_str = secrets.token_hex(4)  # 生成 4 个字节的随机字符串，转换为 8 位十六进制字符串

        screenshot_path = 'instructionscreenshot1-' + time_str + '-' + random_str + '.png'
        self.d.screenshot(screenshot_path)

        return screenshot_path

    def extract_specification(self, text):
        """提取药品规格信息"""
        # 方法1：简单去除到期信息
        pattern = r'^[^【]+'
        match = re.search(pattern, text)
        if match:
            return match.group(0).strip()
        return text

    # 获取商品title
    def get_title(self):

        def _inner():
            print(f'获取商品title时的搜索关键字:{self.title_key}')

            # 初始化
            drugs_name = ''
            specifications = ''
            title = ''

            # 循环的获取title为了有时间来处理人机验证
            for m in range(1, 6000):
                if self.d.xpath(f'//*[contains(@text, "{self.title_key}")]').exists:
                    title = self.safe_exec(
                        lambda: self.d.xpath(f'//*[contains(@text, "{self.title_key}")]').text
                    )
                    print(f"第{m}次获取title成功")
                    break
                else:
                    time.sleep(3)
                    # return drugs_name, specifications

            title = title[1:] if title.startswith('0') else title

            print(f'获取到药品标题:{title}')

            # match = re.match(r'(\[[^\]]+\])(.*?)\s*((?:\d+\S*|\(.+))$', title)
            match = re.match(r'^(?:0?)?(?:\[([^\]]+)\])?\s*(.*?)\s*(\d+[^\s]+)$', title)

            if match:
                # drugs_name = match.group(1).strip() + match.group(2).strip()
                drugs_name = title
                specifications = match.group(3).strip()
                print("药品名:", drugs_name)
                print("规格:", specifications)
                # 如果品规中包含到期则需要再次的正则处理
                if '到期' in specifications:
                    specifications = self.extract_specification(specifications)

                # print('完整药名:', drugs_name + specifications)
                return drugs_name, specifications
            else:
                print("没有匹配到预期格式")
                drugs_name = title
                specifications = ''
                return drugs_name, specifications

        # 用 safe_exec 包装内部逻辑，确保验证码阻塞
        return self.safe_exec(_inner)

    def enter_shop(self):
        """
        进店，方便提取资质环境
        :return:
        """
        self.d.xpath('//*[@text="店铺"]').click()
        time.sleep(0.7)

    def enter_shoper(self):
        """
        进入商家
        :return:
        """
        is_shoper_exists = 0
        for i in range(5):
            if self.d.xpath('//*[@text="商家"]').exists:
                print(f'第{i}次商家存在')
                is_shoper_exists = 1
                break
            else:
                print(f'第{i}次商家不存在')
                time.sleep(0.5)
        if is_shoper_exists == 1:
            self.d.xpath('//*[@text="商家"]').click()
            time.sleep(1)
            return True
        else:
            return False

    # 点击查看商家资质
    def scan_shoper_license(self):
        exist_shoper = 0
        for i in range(6):
            if self.d.xpath('//*[@text="查看商家资质"]').exists:
                print(f'第{i}次查看商家资质存在')
                exist_shoper = 1
                break
            else:
                print(f'第{i}次查看商家资质不存在')

        if exist_shoper == 1:
            self.d.xpath('//*[@text="查看商家资质"]').click()
            time.sleep(0.5)
        else:
            self.swipe_back(1)

    # 验证店铺信息是否在数据库中已存在
    def shop_is_exists_database(self, shop, platform=None):
        platform = str(platform or self.platform or '4').strip()
        query_sql = f"""
            SELECT 1 FROM {self.shop_table_name}
            WHERE shop = %s
            AND platform = %s
            LIMIT 1
        """
        return self.query_exists(query_sql, (shop, platform), "店铺查重")

    def query_one(self, sql, params, error_desc):
        conn = None
        try:
            conn = get_mysql()
            with conn.cursor() as cur:
                cur.execute(sql, params)
                row = cur.fetchone()
                if not row:
                    return None
                columns = [item[0] for item in cur.description]
                return dict(zip(columns, row))
        except Exception as e:
            print(f"{error_desc}错误: {str(e)}")
            return None
        finally:
            if conn:
                conn.close()

    def get_shop_info_from_database(self, shop, platform=None):
        platform = str(platform or self.platform or '4').strip()
        query_sql = f"""
            SELECT shop, contact_address, qualification_number, business_license_company,
                   business_license_address, province, city
            FROM {self.shop_table_name}
            WHERE shop = %s
              AND platform = %s
            ORDER BY scrape_date DESC
            LIMIT 1
        """
        return self.query_one(query_sql, (shop, platform), "查询店铺信息")

    def wait_for_ready(self, monitor, timeout=86400):
        """进入每一页前都先等验证码"""
        if monitor is None:
            return
        start = time.time()
        while monitor.pausing.is_set() and time.time() - start < timeout:
            time.sleep(1)

    def _wait_xpath_exists(self, xpath, timeout=25, interval=0.5):
        deadline = time.time() + timeout
        while time.time() < deadline:
            self.wait_for_ready(getattr(self, "monitor", None))
            try:
                if self.d.xpath(xpath).exists:
                    return True
            except Exception:
                pass
            time.sleep(interval)
        return False

    def _click_xpath_when_ready(self, xpath, action_desc, timeout=25, sleep_after=None):
        if not self._wait_xpath_exists(xpath, timeout=timeout):
            raise RuntimeError(f"{action_desc}失败，未找到元素: {xpath}")
        self.safe_exec(lambda: self.d.xpath(xpath).click())
        if sleep_after is None:
            sleep_after = self.get_sleep_time()
        if sleep_after > 0:
            time.sleep(sleep_after)

    def safe_exec(self, func, *args, **kwargs):
        """
        万能安全壳：执行 func 前检查验证码，
        若监控线程已置位 pausing，则一直阻塞直到放行。
        """
        self.wait_for_ready(getattr(self, "monitor", None))
        max_retries = 3
        for attempt in range(max_retries):
            try:
                result = func(*args, **kwargs)
                # 若执行过程中触发验证码，返回前继续阻塞直到监控放行。
                self.wait_for_ready(getattr(self, "monitor", None))
                return result
            except http.client.RemoteDisconnected as e:
                self.loggerMT.error(f"连接断开 (尝试 {attempt + 1}/{max_retries}): {e}")
                if attempt == max_retries - 1:
                    raise  # 最后一次失败，向上抛出
                # 尝试重连
                if self.reconnect_device():
                    self.loggerMT.info("重连成功，准备重试...")
                    time.sleep(2)  # 等待设备稳定
                    continue
                else:
                    self.loggerMT.error("重连失败，无法继续")
                    raise
            except Exception as e:
                # 其他异常直接抛出
                raise

    def get_next_data(self, data, target):
        for i, item in enumerate(data):
            if item['words'] == target:
                if i + 1 < len(data):
                    return data[i + 1]['words']
        return None

    def delete_instruction_screenshot(self, screenshot_path):
        # 删除截图文件
        try:
            os.remove(screenshot_path)
            print(f"截图文件已删除：{screenshot_path}")
        except FileNotFoundError:
            print(f"文件未找到，无法删除：{screenshot_path}")
        except Exception as e:
            print(f"删除文件时出错：{e}")

    def get_instructions_data(self, capture_only=False):
        """
        确定有说明书之后，提取所有的说明书数据
        :return:
        """
        self.d.xpath('//*[@text="说明"]').click()
        time.sleep(0.3)
        if self.d.xpath('//*[@text="查看详细说明"]').exists:
            self.d.xpath('//*[@text="查看详细说明"]').click()
        else:
            view_all_xpath = self.find_xpath_with_swipes(
                ['//*[@text="查看全部"]'],
                swipe_direction='down',
                swipe_scale=0.3,
                max_swipes=3,
                found_desc='查看全部'
            )
            if view_all_xpath:
                self.d.xpath(view_all_xpath).click()
            else:
                res_data = {
                    "有效期": '',
                    "生产单位": '',
                    "批准文号": ''
                }
                self.loggerMT.info('获取到的说明书信息为空。')
                return res_data

        for ii in range(3):
            if self.d.xpath('//*[@text="查看更多"]').exists:
                self.d.xpath('//*[@text="查看更多"]').click()
                time.sleep(0.3)
                break
            else:
                if self.is_high_res:
                    self.d.swipe(200, 2000, 200, 300, 0.3)
                else:
                    self.d.swipe(200, 1000, 200, 300, 0.3)

        for iii in range(3):
            if self.d.xpath('//*[@text="生产单位"]').exists and self.d.xpath('//*[@text="批准文号"]').exists:
                break
            else:
                if self.is_high_res:
                    self.d.swipe(200, 2000, 200, 300, 0.3)
                else:
                    self.d.swipe(200, 1300, 200, 300, 0.3)
                # self.d.swipe_ext("up", scale=0.3)

        instruction_path = self.screenshot_instruction()
        self.swipe_back(1)

        if capture_only:
            return {"screenshot_path": instruction_path}

        ocr_res = self.get_ocr_res_image(instruction_path)

        if ocr_res:
            # 获取有效期的下一个数据
            validity = self.get_next_data(ocr_res, '有效期')
            # 获取批准文号的下一个数据
            approval_number = self.get_next_data(ocr_res, '批准文号')
            # 获取生产单位的下一个数据
            manufacturer = self.get_next_data(ocr_res, '生产单位')
        else:
            validity = ''
            approval_number = ''
            manufacturer = ''
        res_data = {
            "有效期": validity,
            "生产单位": manufacturer,
            "批准文号": approval_number
        }

        print(f"res_data={res_data}")

        self.delete_instruction_screenshot(instruction_path)

        return res_data

    def has_instructions(self):
        """
        是否有说明书
        :return:
        """
        # 没有说明书的无法采集具体数据
        return bool(self.find_xpath_with_swipes(
            ['//*[@text="说明"]'],
            swipe_direction='down',
            swipe_scale=0.3,
            max_swipes=4,
            found_desc='说明'
        ))

    def has_shop(self):
        """
        是否有进店按钮
        :return:
        """
        is_has_enter_shop = self.d.xpath('//*[@text="进店"]').exists
        return is_has_enter_shop


    def get_license_info_capture(self):
        self.enter_shop()
        result = self.enter_shoper()
        if result is False:
            return {
                'need_save': False,
                'need_back': True,
                'contact_address': '',
                'qualification_number': '',
                'business_license_image_path': ''
            }

        for _ in range(5):
            if self.d.xpath('//*[@text="查看商家资质"]').exists:
                break
            time.sleep(0.5)

        contact_address = self.get_shop_address() or ''
        self.scan_shoper_license()
        time.sleep(3)
        qualification_number = self.get_qualification_number() or ''

        business_license_image_path = ''
        if qualification_number:
            self.d.click(0.603, 0.27)
            time.sleep(1.5)
            captured = self.screenshot_the_business_license(qualification_number)
            business_license_image_path = str(captured) if captured else ''

        return {
            'need_save': True,
            'need_back': True,
            'contact_address': contact_address,
            'qualification_number': qualification_number,
            'business_license_image_path': business_license_image_path
        }

    def distinct_target(self):
        list_page_xpaths = [
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]'
        ]

        exists_tasks = {
            f'list_xpath_{idx}': (lambda xp=xp: self.d.xpath(xp).exists)
            for idx, xp in enumerate(list_page_xpaths)
        }
        exists_results = self.run_parallel_tasks(exists_tasks)
        result = any(bool(v) for v in exists_results.values())

        if result == False:
            print("---检测没有回到列表页---")
        return result
        # return is_position

    def _target_flow_xpaths(self):
        return {
            "medical_entry": '//*[@content-desc="看病买药"]',
            "home_search_entry": '//*[@resource-id="com.sankuai.meituan:id/vf_search_carousel_text"]',
            "search_input": '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]',
            "search_button": '//*[@text="搜索"]',
        }

    def _detect_target_flow_stage(self):
        xpaths = self._target_flow_xpaths()
        if self.distinct_target():
            return "list_page"

        exists_tasks = {
            key: (lambda xp=xp: self.d.xpath(xp).exists)
            for key, xp in xpaths.items()
        }
        exists_results = self.run_parallel_tasks(exists_tasks)

        if exists_results.get("search_button"):
            return "search_input"
        if exists_results.get("search_input"):
            return "search_input"
        if exists_results.get("home_search_entry"):
            return "home_search_entry"
        if exists_results.get("medical_entry"):
            return "medical_entry"
        return None

    def _wait_until_list_page(self, timeout=15, interval=0.8):
        deadline = time.time() + timeout
        while time.time() < deadline:
            if self.distinct_target():
                return True
            time.sleep(interval)
        return False

    def _recover_to_list_page_from_target_flow(self):
        stage = self._detect_target_flow_stage()
        if stage is None:
            return False
        if stage == "list_page":
            return True

        xpaths = self._target_flow_xpaths()
        print(f"检测到已退回入口流程页，当前阶段: {stage}，开始顺序恢复到列表页")

        if stage == "medical_entry":
            self._click_xpath_when_ready(
                xpaths["medical_entry"],
                "进入看病买药页",
                timeout=20
            )
            stage = "home_search_entry"

        if stage == "home_search_entry":
            self._click_xpath_when_ready(
                xpaths["home_search_entry"],
                "点击首页搜索入口",
                timeout=30
            )
            stage = "search_input"

        if stage == "search_input":
            self._click_xpath_when_ready(
                xpaths["search_input"],
                "点击搜索输入框",
                timeout=20,
                sleep_after=0.5
            )
            self.safe_exec(lambda: self.d.send_keys(self.search_key, clear=True))
            time.sleep(0.5)
            self._click_xpath_when_ready(
                xpaths["search_button"],
                "点击搜索按钮",
                timeout=20
            )
            self.safe_exec(self.click_express_send)
            time.sleep(0.5)
            self.wr_re("读", self.device_id)

        if self._wait_until_list_page(timeout=18, interval=0.8):
            print("已从入口流程页恢复到列表页")
            return True
        print("入口流程页恢复后仍未到列表页")
        return False

    def enter_target_page(self):
        self._click_xpath_when_ready(
            '//*[@content-desc="看病买药"]',
            "进入看病买药页",
            timeout=20
        )
        self._click_xpath_when_ready(
            '//*[@resource-id="com.sankuai.meituan:id/vf_search_carousel_text"]',
            "点击首页搜索入口",
            timeout=30
        )
        search_input_xpaths = [
            '//*[@resource-id="com.sankuai.meituan:id/dynamic_hint_layout"]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]',
        ]
        for search_input_xpath in search_input_xpaths:
            if self._wait_xpath_exists(search_input_xpath, timeout=20):
                self._click_xpath_when_ready(
                    search_input_xpath,
                    "点击搜索输入框",
                    timeout=20,
                    sleep_after=0.5
                )
                break
        else:
            raise RuntimeError(f"点击搜索输入框失败，未找到元素: {search_input_xpaths}")
        self.safe_exec(lambda: self.d.send_keys(self.search_key, clear=True))
        time.sleep(0.5)
        self._click_xpath_when_ready(
            '//*[@text="搜索"]',
            "点击搜索按钮",
            timeout=20
        )
        self.safe_exec(self.click_express_send)
        time.sleep(0.5)
        self.wr_re("读", self.device_id)
        time.sleep(2)

    def click_express_send(self):

        slide_xpaths = [
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]',
        ]

        for i in range(1, 3):
            self.wait_for_ready(getattr(self, "monitor", None))
            matched_slide_xpath = self.get_first_existing_xpath(slide_xpaths)
            if not matched_slide_xpath:
                time.sleep(self.get_sleep_time())
                continue
            bounds = self.d.xpath(matched_slide_xpath).info['bounds']
            top = bounds['top']
            bottom = bounds['bottom']
            print(f'top={top}')
            print(f'bottom={bottom}')

            y = (top + bottom) // 2
            print(f'y={y}')
            self.loggerMT.info(f'开始滑动{i}')
            self.safe_exec(lambda: self.d.swipe(500, y, 100, y, 0.5))
            time.sleep(self.get_sleep_time())
            break

        express_send_xpaths = [
            '//*[@text="快递送"]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.ScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]',
            '//*[@resource-id="com.sankuai.meituan:id/container"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]',
            '//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout[3]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]',
            '//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout[3]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.view.ViewGroup[1]/android.view.ViewGroup[1]',
            '//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.HorizontalScrollView[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[3]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]',
        ]
        self.click_candidate_xpaths(
            express_send_xpaths,
            action_desc="点击快递送",
            max_retries=5,
            sleep_after=self.get_sleep_time(),
        )

    def get_clipboard(self):
        time.sleep(0.5)
        clipboard_content = self.d.clipboard
        if clipboard_content is None:
            return ''
        return clipboard_content.strip()

    def get_product_link(self):
        try:

            product_link = ''
            self.safe_exec(self.d.xpath(
                '//android.widget.ScrollView/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]').click)
            if self._check_request_error_after_click():
                print('进入商品链接分享失败，目前没测试原因')
                self.back_to_list_page()
                return product_link

            max_retry = 5  # 最多尝试次数
            for idx in range(1, max_retry + 1):

                time.sleep(random.uniform(0.8, 1))
                x = int(random.uniform(0.2, 0.7) * 720)
                y = int(random.uniform(0.4, 0.7) * 1640)
                self.d.touch.down(x, y)
                time.sleep(random.uniform(0.45, 0.65))
                self.d.touch.up(x, y)

                if self.d(resourceId="com.sankuai.meituan:id/share_title").exists:

                    time.sleep(1.5)
                    save_png = str(self.device_id) + '.png'
                    self.d.screenshot(save_png)
                    product_link = decode_qr(save_png)
                    # 清洗：从元组中提取纯链接
                    if isinstance(product_link, (tuple, list)):
                        product_link = str(product_link[0]) if product_link else ''
                    else:
                        product_link = str(product_link or '')
                    # 只要合法的美团二维码链接
                    if not product_link.startswith('https://'):
                        product_link = ''
                    print(f'{idx}-商品链接:{product_link}')
                    self.loggerMT.info(f'{idx}-商品链接:{product_link}')
                    break

            if not product_link and idx < max_retry:
                time.sleep(0.1)  # 最后一次不需要再等待
            try:
                if product_link:
                    os.makedirs('save_png', exist_ok=True)
                    os.makedirs('error', exist_ok=True)

                    match = re.search(r'[?&]scPid=([^&\s]+)', product_link)

                    if match:
                        code = match.group(1)
                        self.d.screenshot('save_png/'+code+'.png')

                else:
                    self.d.screenshot('error/'+str(time.time())[-7:]+'.png')

            except Exception as e:

                print(e)


            if product_link == '':
                self.swipe_back(1)
            else:
                self.swipe_back(2)
            return product_link
        except Exception as e:
            raise RuntimeError(f"get_product_link 失败: {e}")

    def run_parallel_tasks(self, task_map):
        """
        并行执行相互独立的只读任务。
        任务本身不能包含点击、滑动、返回等会改变页面状态的操作。
        """
        if not task_map:
            return {}

        results = {}
        with ThreadPoolExecutor(max_workers=len(task_map)) as executor:
            future_map = {
                task_name: executor.submit(self.safe_exec, task_func)
                for task_name, task_func in task_map.items()
            }
            for task_name, future in future_map.items():
                try:
                    results[task_name] = future.result()
                except Exception as e:
                    print(f'并行采集任务 {task_name} 执行失败: {e}')
                    results[task_name] = None
        return results

    def _cleanup_post_process_futures(self):
        with self.post_process_lock:
            self.post_process_futures = [f for f in self.post_process_futures if not f.done()]

    def _wait_for_post_task_slot(self):
        while True:
            self._cleanup_post_process_futures()
            with self.post_process_lock:
                pending_count = len(self.post_process_futures)
                futures_snapshot = list(self.post_process_futures)
            if pending_count < self.max_pending_post_tasks:
                return
            if not futures_snapshot:
                return
            wait(futures_snapshot, timeout=5, return_when=FIRST_COMPLETED)

    def _submit_post_process_task(self, save_data, instruction_screenshot_path='', shop_payload=None):
        self._cleanup_post_process_futures()
        payload = dict(save_data)
        shop_payload_copy = dict(shop_payload) if isinstance(shop_payload, dict) else None
        future = self.post_process_executor.submit(
            self._async_finalize_and_store_data,
            payload,
            instruction_screenshot_path,
            shop_payload_copy
        )
        with self.post_process_lock:
            self.post_process_futures.append(future)

    def _wait_post_process_tasks(self):
        with self.post_process_lock:
            futures_snapshot = list(self.post_process_futures)
        for future in futures_snapshot:
            try:
                future.result()
            except Exception as e:
                print(f'后台任务异常: {e}')
        self._cleanup_post_process_futures()

    def _shutdown_post_process_executor(self):
        try:
            self.post_process_executor.shutdown(wait=True)
        except Exception as e:
            print(f'关闭后台线程池异常: {e}')

    def _async_finalize_and_store_data(self, save_data, instruction_screenshot_path='', shop_payload=None):
        final_data = dict(save_data)
        try:
            if instruction_screenshot_path and os.path.exists(instruction_screenshot_path):
                ocr_res = self.get_ocr_res_image(instruction_screenshot_path)
                if ocr_res:
                    validity = self.get_next_data(ocr_res, '有效期')
                    manufacturer = self.get_next_data(ocr_res, '生产单位')
                    approval_number = self.get_next_data(ocr_res, '批准文号')
                    if validity:
                        final_data['expiry_date'] = str(validity).strip('。')
                    if manufacturer:
                        final_data['manufacturer'] = str(manufacturer).strip('。')
                    if approval_number:
                        final_data['approval_number'] = str(approval_number).strip('。')
        except Exception as e:
            print(f'后台处理说明书OCR异常: {e}')
        finally:
            if instruction_screenshot_path:
                self.delete_instruction_screenshot(instruction_screenshot_path)

        try:
            if isinstance(shop_payload, dict) and shop_payload.get("need_save"):
                company = str(shop_payload.get('business_license_company') or '').strip()
                address = str(shop_payload.get('business_license_address') or '').strip()
                contact_address = str(shop_payload.get('contact_address') or address).strip()
                province = str(shop_payload.get('province') or '').strip()
                city = str(shop_payload.get('city') or '').strip()
                biz_img = shop_payload.get("business_license_image_path", "")
                ocr_res = None
                should_ocr = (not company)
                if ENABLE_SHOP_DEBUG:
                    print(
                        f"[SHOP-DEBUG] async shop payload: shop={shop_payload.get('shop', '')}, "
                        f"should_ocr={should_ocr}, company={company}, contact_address={contact_address}, "
                        f"biz_address={address}, province={province}, city={city}, biz_img_exists={bool(biz_img and os.path.exists(biz_img))}"
                    )
                if should_ocr and biz_img and os.path.exists(biz_img):
                    ocr_res = self.get_ocr_res(biz_img)
                    if ocr_res:
                        company = ocr_res.get('单位名称', '') if isinstance(ocr_res, dict) else ''
                        address = ocr_res.get('地址', '') if isinstance(ocr_res, dict) else ''
                        contact_address = str(address or '').strip()
                        if ENABLE_SHOP_DEBUG:
                            print(f"[SHOP-DEBUG] OCR结果: company={company}, biz_address={address}")
                if biz_img and os.path.exists(biz_img):
                    self.write_ocr_result_sidecar(
                        biz_img,
                        ocr_res if should_ocr else {},
                        source="business_license",
                        extra={
                            "shop": shop_payload.get('shop', ''),
                            "qualification_number": shop_payload.get('qualification_number', ''),
                            "should_ocr": should_ocr,
                            "attempted": bool(should_ocr),
                            "parsed_company": company,
                            "parsed_address": address,
                            "contact_address": contact_address
                        }
                    )
                if not address and contact_address:
                    address = contact_address
                if not contact_address and address:
                    contact_address = address
                if (not province) or (not city):
                    match = self.area_service.search_area(address or contact_address)
                    if match:
                        province = province or match.province
                        city = city or match.city
                if ENABLE_SHOP_DEBUG:
                    print(
                        f"[SHOP-DEBUG] region推断后: province={province}, city={city}, by_address={address or contact_address}")
                save_shop_data = {
                    'shop': shop_payload.get('shop', ''),
                    'contact_address': contact_address,
                    'qualification_number': shop_payload.get('qualification_number', ''),
                    'scrape_date': shop_payload.get('scrape_date', self.get_current_date()),
                    'business_license_company': company,
                    'business_license_address': address,
                    'platform': str(shop_payload.get('platform') or self.platform),
                    'province': province,
                    'city': city,
                    'create_time': time.strftime('%Y-%m-%d %H:%M:%S'),
                    'update_time': time.strftime('%Y-%m-%d %H:%M:%S')
                }
                self.save_shop_info_to_database(save_shop_data)
                # 用 area_service 获取 province_id / city_id
                match = self.area_service.search_area(province + city) if (province or city) else None
                derived_province_id = match.province_id if match else 0
                derived_city_id = match.city_id if match else 0
                if ENABLE_SHOP_DEBUG:
                    print(f"[SHOP-DEBUG] derived ids: province_id={derived_province_id}, city_id={derived_city_id}")
                if not final_data.get('company_name'):
                    final_data['company_name'] = company
                if not final_data.get('qualification_number'):
                    final_data['qualification_number'] = str(shop_payload.get('qualification_number', '') or '').strip()
                if province:
                    final_data['province_name'] = province
                if city:
                    final_data['city_name'] = city
                if derived_province_id:
                    final_data['province_id'] = derived_province_id
                if derived_city_id:
                    final_data['city_id'] = derived_city_id
                if ENABLE_SHOP_DEBUG:
                    print(
                        f"[SHOP-DEBUG] final_data回填: company_name={final_data.get('company_name')}, "
                        f"qualification_number={final_data.get('qualification_number')}, "
                        f"province_name={final_data.get('province_name')}, city_name={final_data.get('city_name')}, "
                        f"province_id={final_data.get('province_id')}, city_id={final_data.get('city_id')}"
                    )
        except Exception as e:
            print(f'后台处理店铺OCR异常: {e}')
        print(final_data)
        saved = self.save_to_database(final_data)
        if saved:
            current_count = self.mark_collected_data_saved()
            if self.count is not None:
                print(f"当前已成功采集 {current_count}/{self.count} 条")

    def get_available_xpaths(self, xpaths):
        check_tasks = {
            f'xpath_{idx}': (lambda xp=xp: self.d.xpath(xp).exists)
            for idx, xp in enumerate(xpaths)
        }
        exists_results = self.run_parallel_tasks(check_tasks)
        return [
            xpath for idx, xpath in enumerate(xpaths)
            if exists_results.get(f'xpath_{idx}')
        ]

    def get_first_existing_xpath(self, xpaths):
        available_xpaths = self.get_available_xpaths(xpaths)
        if not available_xpaths:
            return None
        return available_xpaths[0]

    def get_first_text_by_xpaths(self, xpaths):
        text_tasks = {
            f'xpath_{idx}': (lambda xp=xp: self._read_xpath_text(xp))
            for idx, xp in enumerate(xpaths)
        }
        text_results = self.run_parallel_tasks(text_tasks)
        for idx, _ in enumerate(xpaths):
            text = text_results.get(f'xpath_{idx}')
            if text:
                return text
        return ''

    def get_first_texts_by_xpath_groups(self, xpath_groups):
        tasks = {}
        group_keys = {}
        for group_name, xpaths in xpath_groups.items():
            group_keys[group_name] = []
            for idx, xpath in enumerate(xpaths):
                task_name = f'{group_name}_{idx}'
                group_keys[group_name].append(task_name)
                tasks[task_name] = (lambda xp=xpath: self._read_xpath_text(xp))

        text_results = self.run_parallel_tasks(tasks)
        grouped_results = {}
        for group_name, task_names in group_keys.items():
            grouped_results[group_name] = ''
            for task_name in task_names:
                text = text_results.get(task_name)
                if text:
                    grouped_results[group_name] = text
                    break
        return grouped_results

    def _read_xpath_text(self, xpath):
        selector = self.d.xpath(xpath)
        if not selector.exists:
            return ''
        try:
            text = selector.text
            return text.strip() if isinstance(text, str) else text
        except Exception:
            return ''

    def click_candidate_xpaths(self, xpaths, action_desc, max_retries=1, sleep_after=0):
        for attempt in range(1, max_retries + 1):
            available_xpaths = self.get_available_xpaths(xpaths)
            if not available_xpaths:
                print(f'{action_desc}失败，第{attempt}次没有匹配到可点击的xpath')
                time.sleep(self.get_sleep_time())
                continue

            rotate_offset = (attempt - 1) % len(available_xpaths)
            candidate_xpaths = available_xpaths[rotate_offset:] + available_xpaths[:rotate_offset]
            for xpath in candidate_xpaths:
                try:
                    self.safe_exec(lambda xp=xpath: self.d.xpath(xp).click())
                    print(f'{action_desc}成功')
                    if sleep_after:
                        time.sleep(sleep_after)
                    return xpath
                except Exception as e:
                    print(f'{action_desc}点击异常: {e}')
            time.sleep(0.5)
        return None

    def find_xpath_with_swipes(self, xpaths, swipe_direction='down', swipe_scale=0.3, max_swipes=8, found_desc=''):
        for idx in range(max_swipes):
            matched_xpath = self.get_first_existing_xpath(xpaths)
            if matched_xpath:
                if found_desc:
                    print(f'第{idx}次找到{found_desc}')
                return matched_xpath
            self.d.swipe_ext(swipe_direction, swipe_scale)

        matched_xpath = self.get_first_existing_xpath(xpaths)
        if matched_xpath and found_desc:
            print(f'第{max_swipes}次找到{found_desc}')
        return matched_xpath

    def _collect_detail_core_data(self, prefetched, ctx):
        """步骤1: 采集详情页核心字段（标题、价格、销量、自营状态）"""
        prefetched = prefetched if isinstance(prefetched, dict) else {}
        prefetched_product = str(prefetched.get("product") or "").strip()
        prefetched_specifications = str(prefetched.get("specifications") or "").strip()
        prefetched_shop = str(prefetched.get("shop") or "").strip()
        detail_tasks = {
            "sales_num": self.drug_sale_num,
            "is_self_operated": lambda: self.d.xpath('//*[@text="自营"]').exists,
            "min_price": self.drug_price,
        }
        if not prefetched_product:
            detail_tasks["title_info"] = self.get_title

        detail_data = self.run_parallel_tasks(detail_tasks)
        if not prefetched_shop:
            detail_data["shop_inline"] = self.safe_exec(self.get_shop_name_from_current_page)

        if prefetched_product:
            product = prefetched_product
            specifications = prefetched_specifications
        else:
            title_info = detail_data.get("title_info")
            if isinstance(title_info, (list, tuple)) and len(title_info) >= 2:
                product, specifications = title_info[0], title_info[1]
            else:
                product, specifications = "", ""

        if not product:
            self.swipe_back(1)
            return False


        min_price = detail_data.get("min_price")
        if min_price in (None, ""):
            print("详情页未获取到价格，返回列表页采集下一条")
            return False

        if self.collect_range:
            range_start = self.collect_range["start"]
            range_end = self.collect_range["end"]
            if not (range_start <= min_price <= range_end):
                print(f"detail price {min_price} not in range {range_start}-{range_end}, skip")
                return False

        ctx.update({
            "detail_data": detail_data,
            "product": product,
            "specifications": specifications,
            "min_price": min_price,
            "shop": prefetched_shop or (detail_data.get("shop_inline") or "").strip(),
            "sales_num": detail_data.get("sales_num"),
            "scrape_date": self.get_current_date(),
            "product_link": "",
            "shop_async_payload": None,
            "shop_db_info": None,
            "need_collect_shop_ocr": False,
            "instruction_screenshot_path": "",
            "manufacture_date": "",
            "expiry_date": "",
            "manufacturer": "",
            "approval_number": "",
        })
        return True

    def _is_official_shop(self, shop="", detail_data=None):
        shop = str(shop or "").strip()
        detail_data = detail_data if isinstance(detail_data, dict) else {}
        return bool(
            detail_data.get("is_self_operated")
            or ("美团官方" in shop)
            or ("美团自营" in shop)
        )

    def _handle_detail_shop(self, ctx):
        """步骤2: 处理店铺、店铺数据库信息与商品链接"""
        detail_data = ctx.get("detail_data") or {}
        product = ctx["product"]
        min_price = ctx["min_price"]
        scrape_date = ctx["scrape_date"]
        ctx["platform_item_id"] = ctx["shop"] + product
        if detail_data.get("is_self_operated"):
            ctx["shop"] = "美团自营大药房（快递电商）"
            ctx["platform_item_id"] = ctx["shop"] + product
            self._reset_product_link_missing_counter()
            return True

        if not ctx["shop"]:
            self.find_xpath_with_swipes(
                ['//*[@text="进店"]'],
                swipe_direction='up',
                swipe_scale=0.3,
                max_swipes=8,
                found_desc='进店'
            )
            ctx["shop"] = (self.get_shop_name_from_current_page() or self.get_shop_name() or "").strip()

        if not ctx["shop"]:
            print('未获取到店铺名：开始回退')
            self.back_to_list_page()
            return False

        ctx["platform_item_id"] = ctx["shop"] + product

        db_check_tasks = {
            "shop_exists": lambda: self.shop_is_exists_database(ctx["shop"], self.platform),
        }
        db_check_results = self.run_parallel_tasks(db_check_tasks)

        shop_db_info = None
        shop_is_exists = bool(db_check_results.get("shop_exists"))
        if shop_is_exists:
            shop_db_info = self.get_shop_info_from_database(ctx["shop"], self.platform)
        company = str((shop_db_info or {}).get("business_license_company") or "").strip()
        biz_address = str((shop_db_info or {}).get("business_license_address") or "").strip()
        contact = str((shop_db_info or {}).get("contact_address") or "").strip()
        need_collect_shop_ocr = (not shop_is_exists) or (not company)
        if ENABLE_SHOP_DEBUG:
            print(
                f"[SHOP-DEBUG] should_collect_shop_ocr: "
                f"company_empty={not bool(company)}, biz_address_empty={not bool(biz_address)}, contact_empty={not bool(contact)}"
            )
            print(
                f"[SHOP-DEBUG] 主流程店铺判定: shop={ctx['shop']}, "
                f"shop_is_exists={shop_is_exists}, need_collect_shop_ocr={need_collect_shop_ocr}, "
                f"shop_db_info={shop_db_info}"
            )

        ctx["shop_db_info"] = shop_db_info
        ctx["need_collect_shop_ocr"] = need_collect_shop_ocr

        ctx["product_link"] = self._record_product_link_result(self.get_product_link())
        if not ctx["product_link"]:
                print("当前商品获取不到商品链接，返回列表页")
                self.back_to_list_page()
                return False
        return True

    def _collect_detail_instruction_info(self, ctx):
        """步骤3: 采集说明书截图信息"""
        if not self.safe_exec(self.has_instructions):
            return
        print('开始获取说明书信息')
        try:
            instructions_info = self.safe_exec(lambda: self.get_instructions_data(capture_only=True))
            if isinstance(instructions_info, dict):
                ctx["instruction_screenshot_path"] = instructions_info.get("screenshot_path", "")
        except Exception as e:
            print(f'说明书采集跳过: {e}')

    def _collect_detail_shop_ocr(self, ctx):
        """步骤4: 采集店铺资质 OCR"""
        if self._is_official_shop(ctx.get("shop", ""), ctx.get("detail_data")):
            return

        print(f"已采集{self.shop_data_num}家店铺数据")
        is_has_enter_shop = bool(self.safe_exec(self.has_shop))
        shop = ctx["shop"]
        if is_has_enter_shop and ctx["need_collect_shop_ocr"] and self.shop_data_num < 500:
            license_capture = self.safe_exec(self.get_license_info_capture)
            if license_capture.get("need_back"):
                self.swipe_back(2)
            ctx["shop_async_payload"] = {
                "need_save": bool(license_capture.get("need_save")),
                "shop": shop,
                "contact_address": license_capture.get("contact_address", ""),
                "qualification_number": license_capture.get("qualification_number", ""),
                "business_license_image_path": license_capture.get("business_license_image_path", ""),
                "scrape_date": ctx["scrape_date"],
                "platform": str(self.platform),
            }
            if ENABLE_SHOP_DEBUG:
                print(f"[SHOP-DEBUG] 走OCR采集分支, shop_async_payload={ctx['shop_async_payload']}")
            if ctx["shop_async_payload"]["need_save"]:
                self.shop_data_num += 1
        else:
            print('不采集店铺信息')

    def _build_detail_save_data(self, ctx):
        """步骤5: 组装详情页保存数据"""
        shop_db_info = ctx.get("shop_db_info")
        shop = ctx["shop"]
        province_id, city_id, province, city = 0, 0, '', ''

        # 优先从 DB 缓存取，没有则从地址中推断
        addr_for_region = ''
        if isinstance(shop_db_info, dict):
            province = str(shop_db_info.get('province') or '').strip()
            city = str(shop_db_info.get('city') or '').strip()
            addr_for_region = str(
                shop_db_info.get('business_license_address')
                or shop_db_info.get('contact_address')
                or ''
            ).strip()

        if (not province or not city) and addr_for_region:
            match = self.area_service.search_area(addr_for_region)
            if match:
                province_id, city_id = match.province_id, match.city_id
                province = province or match.province
                city = city or match.city
        elif province or city:
            # DB 已有省市，只查 ID
            match = self.area_service.search_area(province + city)
            if match:
                province_id, city_id = match.province_id, match.city_id

        if ENABLE_SHOP_DEBUG:
            print(
                f"[SHOP-DEBUG] save_data省市计算: shop={shop}, province={province}, city={city}, "
                f"province_id={province_id}, city_id={city_id}"
            )

        save_data = {
            'enterprise_id': self.enterprise_id,
            'platform_id': 4,
            'platform_item_id': ctx["platform_item_id"],
            'province_id': province_id,
            'city_id': city_id,
            'province_name': '',
            'city_name': '',
            'area_info': "",
            'product_brand': self.brand,
            'product_name': ctx["product"],
            'product_specs': ctx["specifications"],
            'one_box_price': 0.00,
            'manufacture_date': ctx["manufacture_date"],
            'expiry_date': ctx["expiry_date"],
            'manufacturer': ctx["manufacturer"],
            'approval_number': ctx["approval_number"],
            'is_sold_out': 0,
            'online_posting_count': 1,
            'continuous_listing_count': 1,
            'link_url': ctx["product_link"],
            'store_name': shop,
            'store_url': '',
            'shipment_province_id': 0,
            'shipment_province_name': "",
            'shipment_city_id': 0,
            'shipment_city_name': "",
            'company_name': "",
            'qualification_number': "",
            'search_name': self.search_key,
            'scrape_date': ctx["scrape_date"],
            'min_price': ctx["min_price"],
            'number': 1,
            'sales': ctx["sales_num"],
            'inventory': "",
            'snapshot_url': str(ctx.get("snapshot_url") or ""),
            'collect_equipment_account_id': self.collect_equipment_account_id,
            'collect_region_id': self.collect_region_id,
            'collect_round': self.collect_round,
            'insert_time': time.strftime('%Y-%m-%d %H:%M:%S'),
            'update_time': time.strftime('%Y-%m-%d %H:%M:%S'),
            'collect_config_info': {
                "sampling_cycle": self.sampling_cycle,
                "sampling_start_time": self.sampling_start_time,
                "sampling_end_time": self.sampling_end_time,
            },
            'task_id': self.task_id,
        }
        if isinstance(shop_db_info, dict):
            save_data['province_name'] = province
            save_data['city_name'] = city
            save_data['province_id'] = province_id
            save_data['city_id'] = city_id
            save_data['company_name'] = str(shop_db_info.get('business_license_company') or '')
            save_data['qualification_number'] = str(shop_db_info.get('qualification_number') or '')
            if ENABLE_SHOP_DEBUG:
                print(
                    f"[SHOP-DEBUG] save_data DB回填结果: company_name={save_data['company_name']}, "
                    f"qualification_number={save_data['qualification_number']}, "
                    f"province_name={save_data['province_name']}, city_name={save_data['city_name']}, "
                    f"province_id={save_data['province_id']}, city_id={save_data['city_id']}"
                )
        if isinstance(save_data.get('collect_config_info'), dict):
            save_data['collect_config_info'] = json.dumps(save_data['collect_config_info'], ensure_ascii=False)
        ctx["save_data"] = save_data
        return save_data

    def _submit_detail_record(self, ctx):
        """步骤6: 提交后台处理并回到列表页"""
        self._submit_post_process_task(
            ctx["save_data"],
            instruction_screenshot_path=ctx["instruction_screenshot_path"],
            shop_payload=ctx["shop_async_payload"]
        )
        print(f'[{datetime.datetime.now().strftime("%H:%M:%S.%f")}] 已提交后台异步处理OCR与入库')
        print(f'[{datetime.datetime.now().strftime("%H:%M:%S.%f")}] 开始回列表页')
        return self.back_to_list_page()

    def integrate_data(self, prefetched=None):
        """
        整合详情页数据
        """
        ctx = {}
        prefetched_title = ""
        if isinstance(prefetched, dict):
            prefetched_title = str(prefetched.get("product") or "").strip()
        mt_screenshot = MTScreenshot(
            d=self.d,
            oss_config=self.oss_config,
            search_key=self.search_key,  # 添加这行
            title_key=self.title_key,
            device_id=self.device_id,
            monitor=getattr(self, "monitor", None)
        )

        # 1. 采集详情页核心字段
        if not self._collect_detail_core_data(prefetched, ctx):
            return
        # 2. 处理店铺、去重、店铺数据库信息与商品链接
        if not self._handle_detail_shop(ctx):
            return
        # 3. 采集说明书截图
        self._collect_detail_instruction_info(ctx)

        # 4.网页快照
        snapshot_url = mt_screenshot.get_oss_url(title=prefetched_title)
        ctx["snapshot_url"] = str(snapshot_url or "")

        # 5. 采集店铺资质 OCR
        self._collect_detail_shop_ocr(ctx)
        # 6. 组装保存数据
        self._build_detail_save_data(ctx)
        # 7. 提交后台处理并回列表页
        return self._submit_detail_record(ctx)

    def back_to_list_page(self):
        for i in range(5):
            if self.distinct_target():
                return True
            if i >= 3 and self._recover_to_list_page_from_target_flow():
                return True
            print(f'第{i}次尝试退回到列表页')
            self.swipe_back(1)
            time.sleep(0.5)

        if self._recover_to_list_page_from_target_flow():
            return True
        print('页面出错，没有退回到列表页')
        return False

    def reset_collection_cursor(self):
        self.collection_cursor["page_no"] = 1
        self.collection_cursor["item_index"] = 0

    def get_current_page_no(self):
        return self.page + self.collection_cursor["page_no"]

    def jump_to_page(self, target_page):
        current_page = self.get_current_page_no()
        if target_page <= current_page:
            return

        while current_page < target_page:
            if self.d.xpath('//*[@text="已经到底啦"]').exists:
                print(f"列表实际页数不足，当前停留在第{current_page}页，无法跳转到第{target_page}页")
                return
            print(f"跳过第{current_page}页，前往第{current_page + 1}页")
            if self.is_high_res:
                self.d.drag(300, 2600, 300, 400, 1)
            else:
                self.d.drag(300, 1400, 300, 400, 1)
            time.sleep(1)
            self.collection_cursor["page_no"] += 1
            self.collection_cursor["item_index"] = 0
            current_page = self.get_current_page_no()

    def move_to_page_range_start(self):
        if not self.page_range:
            return

        start_page = self.page_range["start"]
        current_page = self.get_current_page_no()
        if current_page < start_page:
            self.jump_to_page(start_page)

    def start_collection_app(self):
        self.sort_key = 0
        self.restart_app()

    def open_product_list_page(self):
        last_error = None
        for attempt in range(1, OPEN_PRODUCT_LIST_PAGE_RETRY + 1):
            try:

                self.safe_exec(self.enter_target_page)
                self.reset_collection_cursor()
                if self.sort and self.sort_key == 0:
                    self.li_or_lo(self.sort)
                self.move_to_page_range_start()
                return
            except Exception as e:
                last_error = e
                self.loggerMT.warning(
                    f"open_product_list_page 第{attempt}/{OPEN_PRODUCT_LIST_PAGE_RETRY}次失败: {e}"
                )
                if attempt >= OPEN_PRODUCT_LIST_PAGE_RETRY:
                    break
                ready = self._wait_xpath_exists('//*[@content-desc="看病买药"]', timeout=6, interval=0.5)
                if not ready:
                    record_restart_reason(
                        reason="open_product_list_page 入口未就绪，执行 restart_app 后重试",
                        source="workflow",
                        device_id=getattr(self, "device_id", None),
                        task_id=getattr(self, "task_id", None),
                        step="open_product_list_page",
                        action="restart_app",
                        fail_count=attempt,
                        retry_limit=OPEN_PRODUCT_LIST_PAGE_RETRY,
                        search_key=getattr(self, "search_key", None),
                        exc=e,
                        traceback_text=traceback.format_exc(),
                    )
                    self.restart_app()
                time.sleep(max(2, self.get_sleep_time()))
        raise RuntimeError(f"open_product_list_page 重试{OPEN_PRODUCT_LIST_PAGE_RETRY}次仍失败: {last_error}")

    def handle_workflow_error(self, step_name):
        action = self.workflow_error_action.get(step_name)
        if action == "back_to_list_page":
            if not self.back_to_list_page():
                raise RuntimeError("退回列表页失败")
            return "collect_single_product"
        if action == "open_product_list_page":
            return "open_product_list_page"
        if action == "start_app":
            return "start_app"
        raise RuntimeError(f"未配置步骤 {step_name} 的错误处理动作")

    def get_list_items(self):
        for _ in range(10):
            items = self.safe_exec(
                self.d.xpath('//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout').all
            )
            if items:
                return items
            time.sleep(1)
        raise RuntimeError("列表页商品加载失败")

    def _get_list_visible_params(self):
        """根据分辨率返回列表可见区参数"""
        if self.is_high_res:
            return {"visible_top": 509, "visible_bottom": 2646, "target_top": 519}
        else:
            return {"visible_top": 304, "visible_bottom": 1475, "target_top": 314}

    def _get_sorted_visible_items(self):
        """获取 RecyclerView 中所有可见 item，按 top 从小到大排序"""
        items = self.d.xpath(
            '//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout'
        ).all()
        result = []
        for item in (items or []):
            try:
                bounds = item.info.get("bounds") or item.info.get("visibleBounds") or {}
                top = int(bounds.get("top", 0))
                bottom = int(bounds.get("bottom", 0))
                if top >= 0 and bottom > top:
                    result.append((top, bottom, item))
            except Exception:
                continue
        result.sort(key=lambda x: x[0])
        return result

    def _anchor_scroll_to_next_page(self):
        """锚点滑动：把最后一个可见 item 滑到可见区顶部，确保翻页不漏商品"""
        sorted_items = self._get_sorted_visible_items()
        if len(sorted_items) < 1:
            return False

        params = self._get_list_visible_params()
        anchor_top = sorted_items[-1][0]
        target_top = params["target_top"]
        scroll_distance = anchor_top - target_top

        if scroll_distance <= 0:
            return False

        screen_width = self.d.info.get("displayWidth", 1220)
        start_x = screen_width // 2
        start_y = params["visible_bottom"] - 200
        end_y = max(start_y - scroll_distance, 100)

        self.d.drag(start_x, start_y, start_x, end_y, duration=1)
        return True

    def move_to_next_list_page(self):
        current_page = self.get_current_page_no()
        # 逐页回告：当前页采集完毕，上报进度
        if self.scheduler and self.task_id:
            resp = self.scheduler.post_report({
                "task_id": self.task_id,
                "platform": str(self.platform),
                "username": self.scheduler.username,
                "current_page": current_page,
                "crawled_count": self.get_collected_data_count(),
                "is_finished": 0,
            })
            if isinstance(resp, dict) and resp.get("code") == "error":
                logging.warning(f"调度器要求停止: {resp.get('msg', '')}")
                return False
        if self.page_range and current_page >= self.page_range["end"]:
            self.wr_re("写", self.device_id, self.sort, current_page)
            print(f'已完成第{current_page}页采集，达到结束页{self.page_range["end"]}，停止采集')
            return False
        if self.d.xpath('//*[@text="已经到底啦"]').exists:
            return False
        # 翻页前检查"加载更多"按钮，检测风控卡死
        if self.d.xpath('//*[@text="加载更多"]').exists:
            self._check_wind_control_stuck()
        self.wr_re("写", self.device_id, self.sort, current_page)
        print(f'当前第{current_page}页采集完成，开始滑动到下一页')
        self._anchor_scroll_to_next_page()
        time.sleep(1)
        self.collection_cursor["page_no"] += 1
        self.collection_cursor["item_index"] = 0
        return True

    def _get_list_item_snapshot(self, drug_idx, drug_one):
        """步骤1: 读取列表商品的可见区域和文本快照"""
        bounds = drug_one.info['bounds']
        top = bounds['top']
        bottom = bounds['bottom']

        # print(f'当前商品bottom:{bottom}')
        # print(f'当前商品top:{top}')
        if self.is_high_res:
            if not (509 <= top and bottom <= 2646):
                return None
        else:
            if not (304 < top and bottom <= 1475):
                return None

        item_text_data = self.get_first_texts_by_xpath_groups({
            "product_title": [
                f'//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView[1]'],
            "price_str": [
                f'//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView[1]',
            ],
            "shop_name": [
                f'//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]//*[contains(@text, "快递电商")]',
                f'//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup/android.view.ViewGroup/android.view.ViewGroup[2]/android.view.ViewGroup/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup/android.view.ViewGroup/android.widget.FrameLayout/android.widget.TextView',
                f'//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout[{drug_idx}]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView[1]',
                f'/hierarchy/android.widget.FrameLayout[2]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[3]/android.widget.FrameLayout[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/androidx.recyclerview.widget.RecyclerView[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[1]/android.view.ViewGroup[2]/android.view.ViewGroup[3]/android.view.ViewGroup[1]/android.view.ViewGroup[1]/android.widget.FrameLayout[1]/android.widget.TextView[1]'
            ],
        })
        return {
            "drug_idx": drug_idx,
            "drug_one": drug_one,
            "item_text_data": item_text_data,
        }

    def _prepare_list_item_collect(self, item_ctx):
        """步骤2: 解析列表商品信息并完成过滤/去重准备"""
        drug_idx = item_ctx["drug_idx"]
        item_text_data = item_ctx["item_text_data"]
        print(f"这页的第几个商品:{drug_idx}")

        product_title = item_text_data.get("product_title", "")
        if not product_title:
            print("列表当前商品名称不存在")
            self.unrelated_data += 1
            return "continue"

        product_title = product_title[1:] if product_title.startswith('0') else product_title
        print(f"列表当前商品名称:{product_title}")
        if not self.is_link_useful(product_title):
            print(f"is_link_useful 没通过:{product_title}")
            self.unrelated_data += 1
            return "continue"

        self.unrelated_data = 0

        price = ''
        price_str = item_text_data.get("price_str", "")
        print(f"列表当前商品价格:{price_str}")
        if price_str:
            price = float(re.search(r'[\d\.]+', price_str).group())

        shop_name = item_text_data.get("shop_name", "")
        print(f"列表当前商品店铺名称:{shop_name}")

        if price == '' or shop_name == '':
            print("列表当前商品价格或店铺名称不存在","价格：",price,"店铺名：",shop_name)

            return "continue"

        scrape_date = self.get_current_date()
        item_ctx["product_title"] = product_title
        item_ctx["shop_name"] = shop_name
        item_ctx["prefetched_detail"] = {
            "product": product_title,
            "specifications": "",
            "shop": shop_name,
        }
        return "ready"

    def _reset_request_error_counter(self):
        if self.request_error_count:
            print(f"请求错误计数已重置，上一轮累计: {self.request_error_count}")
        self.request_error_count = 0

    def _reset_product_link_missing_counter(self):
        if self.product_link_missing_count:
            print(f"商品链接缺失计数已重置，上一轮累计: {self.product_link_missing_count}")
        self.product_link_missing_count = 0

    def _check_wind_control_stuck(self):
        """
        风控卡死检测：
        每隔30秒检查一次 //*[@text="加载更多"] 是否存在，共检查4次。
        如果4次都存在，判定为风控卡死，抛出异常终止任务。
        如果任意一次不存在，判定为非风控，继续正常采集。
        """
        self.loggerMT.warning(
            f"开始风控卡死检测，将检查'加载更多'按钮{self.load_more_check_rounds}次，"
            f"间隔{self.load_more_check_interval}秒"
        )

        load_more_xpath = '//*[@text="加载更多"]'
        load_more_exists_count = 0

        for check_round in range(1, self.load_more_check_rounds + 1):
            self.wait_for_ready(getattr(self, "monitor", None))
            try:
                load_more_exists = self.d.xpath(load_more_xpath).exists
            except Exception:
                load_more_exists = False

            if load_more_exists:
                load_more_exists_count += 1
                print(f"第{check_round}/{self.load_more_check_rounds}次检测: '加载更多' 存在 "
                      f"(累计{load_more_exists_count}次)")
                self.loggerMT.info(
                    f"风控卡死检测 第{check_round}/{self.load_more_check_rounds}次: '加载更多' 存在"
                )
            else:
                print(f"第{check_round}/{self.load_more_check_rounds}次检测: '加载更多' 不存在，"
                      f"判定为非风控")
                self.loggerMT.info(
                    f"风控卡死检测 第{check_round}/{self.load_more_check_rounds}次: '加载更多' 不存在"
                )
                return

            if check_round < self.load_more_check_rounds:
                print(f"等待{self.load_more_check_interval}秒后进行下一次检测...")
                time.sleep(self.load_more_check_interval)

        # 所有轮次都存在"加载更多"，判定为风控卡死
        raise WindControlStuckError(
            f"风控卡死确认：'加载更多'按钮{self.load_more_check_rounds}次检测均存在"
            f"（间隔{self.load_more_check_interval}秒），疑似被风控限制无法获取新数据"
        )

    def _record_product_link_result(self, product_link):
        product_link = str(product_link or "").strip()
        if product_link:
            self._reset_product_link_missing_counter()
            return product_link

        self.product_link_missing_count += 1
        print(
            f"当前商品获取不到商品链接，第{self.product_link_missing_count}/"
            f"{self.product_link_missing_threshold}次"
        )
        if self.product_link_missing_count >= self.product_link_missing_threshold:
            raise ProductLinkUnavailableError(
                f"连续{self.product_link_missing_count}次获取不到商品链接，停止采集"
            )
        return ""

    def _check_request_error_after_click(self, timeout=3, interval=0.5):
        deadline = time.time() + timeout
        while time.time() < deadline:
            self.wait_for_ready(getattr(self, "monitor", None))
            try:
                if self.d.xpath('//*[@text="请求错误"]').exists:
                    self.request_error_count += 1
                    print(
                        f"点击商品后检测到请求错误，第{self.request_error_count}/"
                        f"{self.request_error_threshold}次"
                    )
                    if self.request_error_count >= self.request_error_threshold:
                        raise AccountBlockedError(
                            f"连续点击{self.request_error_count}个商品均出现请求错误，疑似账号被封禁，停止采集"
                        )
                    return True
            except AccountBlockedError:
                raise
            except Exception:
                pass
            time.sleep(interval)
        return False

    def _enter_list_item_detail(self, item_ctx):
        """步骤3: 点击商品并交给详情页采集"""
        self.safe_exec(item_ctx["drug_one"].click)
        print('点击目标药品完毕')
        if self._check_request_error_after_click():
            self.back_to_list_page()
            return "continue"

        self._reset_request_error_counter()
        integrate_ok = self.safe_exec(lambda: self.integrate_data(prefetched=item_ctx["prefetched_detail"]))
        print('integrate_data结束')
        if not integrate_ok:
            if self._check_request_error_after_click(timeout=1, interval=0.3):
                self.safe_exec(self.back_to_list_page)
                return "continue"
            self.safe_exec(self.back_to_list_page)
        return "collected"

    def _collect_list_item(self, drug_idx, drug_one):
        item_ctx = self._get_list_item_snapshot(drug_idx, drug_one)
        if not item_ctx:
            return "skip"

        collect_state = self._prepare_list_item_collect(item_ctx)
        if collect_state != "ready":
            return collect_state

        return self._enter_list_item_detail(item_ctx)

    def collect_single_product(self):
        if self.monitor and self.monitor.captcha_appearance_limit_reached:
            raise CollectionStopError(
                f"验证码出现过多，1小时内累计出现{len(self.monitor.captcha_appearance_timestamps)}次验证码，终止任务"
            )
        if self.monitor.verification_count >= self.monitor.MAX_VERIFICATION_RETRY:
            raise RuntimeError("验证码触发过多，暂停程序")
        if self.has_reached_target_count():
            print(f"已达到目标采集数量 {self.count} 条，停止采集")
            return False
        if self.page_range:
            self.move_to_page_range_start()
            current_page = self.get_current_page_no()
            if current_page > self.page_range["end"]:
                print(f"当前已在第{current_page}页，超过结束页{self.page_range['end']}，停止采集")
                return False
        items = self.get_list_items()
        print(f'当前第{self.get_current_page_no()}页，共有{len(items)}个商品')
        while self.collection_cursor["item_index"] < len(items):
            item_index = self.collection_cursor["item_index"]
            self.collection_cursor["item_index"] += 1
            result = self._collect_list_item(item_index + 1, items[item_index])
            if result == "collected":
                if self.count is not None:
                    self._wait_post_process_tasks()
                    if self.has_reached_target_count():
                        print(f"已达到目标采集数量 {self.count} 条，停止采集")
                        return False
                return True
            if result == "continue":
                return True

        if not self.move_to_next_list_page():
            print('已经到达列表页最底部')
            return False
        return True

    def execute_workflow_step(self, step_name):
        if step_name == "start_app":
            self.safe_exec(self.start_collection_app)
            return "open_product_list_page"
        if step_name == "open_product_list_page":
            self.safe_exec(self.open_product_list_page)
            MT.open_page_fail_count = 0  # 成功则重置封号计数
            return "collect_single_product"
        if step_name == "collect_single_product":
            has_next = self.safe_exec(self.collect_single_product)
            if not has_next:
                return None
            print('目前连续无关数据量: ', self.unrelated_data)
            if self.unrelated_data > self.max_unrelated_data:
                print(f"连续超过{self.max_unrelated_data}个不达标的数据则停止采集")
                self.finish_task_normally(
                    self.get_current_page_no(),
                    f"连续超过{self.max_unrelated_data}个不达标的数据则停止采集",
                )
                return None
            return "collect_single_product"
        raise RuntimeError(f"未知流程步骤: {step_name}")

    def main(self, device_id):
        self.device_id = device_id
        self.connect_devices(device_id)
        time.sleep(self.get_sleep_time())
        self.monitor = SpiderMonitor(self)
        self.monitor.start()
        current_step = "start_app"
        step_failures = {step: 0 for step in self.workflow_retry_limit}
        try:
            while current_step:
                try:
                    next_step = self.execute_workflow_step(current_step)
                    step_failures[current_step] = 0
                    current_step = next_step
                except CollectionStopError:
                    raise
                except Exception as e:
                    # 验证码重试超限，直接终止任务并回告
                    if "验证码触发过多" in str(e):
                        self.finish_task_abnormally(self.get_current_page_no(), f"验证码处理失败: {e}")
                        raise CollectionStopError(str(e))
                    print(f'{current_step} 执行异常: {e}')
                    time.sleep(3)
                    step_failures[current_step] += 1
                    # 封号检测：open_product_list_page 连续失败
                    if current_step == "open_product_list_page":
                        MT.open_page_fail_count += 1
                        print(f"⚠️ open_product_list_page 连续失败 {MT.open_page_fail_count}/{MT.open_page_fail_threshold} 次")
                        if MT.open_page_fail_count >= MT.open_page_fail_threshold:
                            msg = f"疑似账号被封禁：open_product_list_page 连续失败 {self.open_page_fail_threshold} 次"
                            logging.error(msg)
                            self.finish_task_abnormally(self.get_current_page_no(), msg)
                            raise CollectionStopError(msg)
                    retry_limit = self.workflow_retry_limit.get(current_step)
                    next_action = self.workflow_error_action.get(current_step)
                    record_restart_reason(
                        reason="工作流步骤异常，准备按配置重新开始",
                        source="workflow",
                        device_id=self.device_id,
                        task_id=self.task_id,
                        step=current_step,
                        action=next_action,
                        fail_count=step_failures[current_step],
                        retry_limit=retry_limit,
                        search_key=self.search_key,
                        exc=e,
                        traceback_text=traceback.format_exc(),
                    )
                    if step_failures[current_step] > self.workflow_retry_limit[current_step]:
                        raise
                    current_step = self.handle_workflow_error(current_step)
            return self.finish_task_normally(self.get_current_page_no(), "美团任务执行完成")
        finally:
            self._wait_post_process_tasks()
            self._shutdown_post_process_executor()
            self.monitor.stop()
            self.monitor.join()


def fetch_task_from_scheduler(scheduler, device_id):
    """从已有的调度器获取一个任务，转换为 device_list 兼容的格式。没有任务返回 None。"""
    task = scheduler.get_task()
    if not task:
        return None
    start_offset = task.get("current_page", 0)  # 移动端起始偏移量，0=从头开始
    # start_offset = 0  # 移动端起始偏移量，0=从头开始
    start_page = start_offset if start_offset > 0 else 1
    end_page = task.get("end_page", 0)
    # 转成 page_range 格式，MT.open_product_list_page → move_to_page_range_start 会跳页
    if start_page > 1 or end_page > 0:
        page_range = {"start": start_page, "end": end_page if end_page > 0 else 200}
    else:
        page_range = []
    return {
        "search_key": f"{task.get('product_brand', '')} {task.get('product_name', '')}".strip(),
        "title_key": task.get("product_name", ""),
        "spec_list": parse_spec_list(task.get("product_specs")),
        "brand": task.get("product_brand", ""),
        "sort": "",
        "collect_range": [],
        "page_range": page_range,
        "platform": PLATFORM_MT,
        "task_id": task.get("id"),
        "enterprise_id": task.get("company_id"),
        "equipment_id": task.get("collect_equipment_id", 0),
        "device_name": device_id,
        "collect_equipment_account_id": 0,
        "collect_region_id": 0,
        "collect_round": task.get("collect_round", 1),
        "sampling_cycle": "",
        "sampling_start_time": "",
        "sampling_end_time": "",
        "count": 150,
        "search_task_mode": SEARCH_TASK_MODE,
        "workflow_retry_limit": {
            "start_app": 3,
            "open_product_list_page": 3,
            "collect_single_product": 3,
        },
        "workflow_error_action": {
            "start_app": "start_app",
            "open_product_list_page": "start_app",
            "collect_single_product": "back_to_list_page",
        },
    }

def run_device(device_id, scheduler=None):
    """单个设备的采集任务。scheduler 由外部传入复用，避免重复启停心跳。"""
    if device_id not in device_list:
        logging.error(f"设备id没有配置: {device_id}")
        return

    own_scheduler = False
    if scheduler is None:
        scheduler = CrawlerScheduler(DEVICE_ID=device_id, platform=str(PLATFORM_MT))
        scheduler.start()
        time.sleep(2)
        own_scheduler = True
    logging.info(f"[设备 {device_id}] 调度器已启动（心跳+回告）")

    tasks = device_list[device_id]
    logging.info(f"[设备 {device_id}] 开始执行，共 {len(tasks)} 个任务")

    for task in tasks:
        mode = task.get("search_task_mode", SEARCH_TASK_MODE)
        variants = build_search_variants(task.get("search_key"), task.get("spec_list"), mode)
        logging.info(
            f"[设备 {device_id}] 任务 {task.get('search_key')} 使用搜索模式 {mode}，共 {len(variants)} 组搜索词")
        for variant_idx, variant in enumerate(variants, start=1):
            cycle_no = 0
            while True:
                cycle_no += 1
                mt = None
                variant_search_key = variant["search_key"]
                variant_specs = variant["spec_list"]
                logging.info(
                    f'[设备 {device_id}] ========== 搜索组 {variant_idx}/{len(variants)} '
                    f'{variant_search_key} 第 {cycle_no} 轮采集开始 =========='
                )
                counter_key = _build_failure_counter_key(
                    "local",
                    device_id,
                    search_key=variant_search_key,
                )
                try:
                    mt = MT(
                        variant_search_key,
                        task["title_key"],
                        variant_specs,
                        task["brand"],
                        task.get("sort"),
                        task.get("collect_range"),
                        task.get("page_range"),
                        task.get("workflow_retry_limit"),
                        task.get("workflow_error_action"),
                        platform=task.get("platform"),
                        task_id=task.get("task_id"),
                        enterprise_id=task.get("enterprise_id"),
                        sampling_cycle=task.get("sampling_cycle"),
                        sampling_start_time=task.get("sampling_start_time"),
                        sampling_end_time=task.get("sampling_end_time"),
                        count=task.get("count"),
                        collect_equipment_id=task.get("collect_equipment_id"),
                        device_name=task.get("device_name"),
                        collect_equipment_account_id=task.get("collect_equipment_account_id"),
                        collect_region_id=task.get("collect_region_id"),
                        collect_round=task.get("collect_round"),
                        scheduler=scheduler,
                    )
                    mt.main(device_id)
                    logging.info(f'[设备 {device_id}] 搜索组 {variant_idx}/{len(variants)} 本轮采集完成')
                    reset_failure_notice_counter(counter_key)
                    reset_captcha_restart_count(counter_key)
                    break  # 成功则跳出当前搜索组重试循环
                except CollectionStopError as e:
                    tb_text = traceback.format_exc()
                    logging.exception(
                        f'[设备 {device_id}] 搜索组 {variant_idx}/{len(variants)} 关键字 {variant_search_key} '
                        f'检测到致命停止条件：{e}'
                    )
                    record_restart_reason(
                        reason="run_device 检测到致命停止条件，终止程序",
                        source="collection_stop_local",
                        device_id=device_id,
                        task_id=task.get("task_id"),
                        search_key=variant_search_key,
                        cycle_no=cycle_no,
                        exc=e,
                        traceback_text=tb_text,
                    )
                    if mt is not None:
                        end_page = mt.get_current_page_no() if hasattr(mt, "get_current_page_no") else getattr(mt,
                                                                                                               "page",
                                                                                                               0)
                        mt.finish_task_abnormally(end_page, f"任务终止: {e}")
                    raise
                except Exception as e:
                    tb_text = traceback.format_exc()
                    logging.exception(
                        f'[设备 {device_id}] 搜索组 {variant_idx}/{len(variants)} 关键字 {variant_search_key} 采集异常：{e}')
                    record_restart_reason(
                        reason="run_device 捕获异常后进入下一轮重试",
                        source="run_device",
                        device_id=device_id,
                        task_id=task.get("task_id"),
                        search_key=variant_search_key,
                        cycle_no=cycle_no,
                        exc=e,
                        traceback_text=tb_text,
                    )
                    if _is_captcha_related_error(str(e), tb_text):
                        captcha_restart_count = increase_captcha_restart_count(counter_key)
                        record_restart_reason(
                            reason=f"验证码导致重启，1小时窗口内累计第{captcha_restart_count}次",
                            source="captcha_restart_alert_local",
                            device_id=device_id,
                            task_id=task.get("task_id"),
                            search_key=variant_search_key,
                            fail_count=captcha_restart_count,
                            retry_limit=CAPTCHA_STOP_THRESHOLD,
                            cycle_no=cycle_no,
                            exc=e,
                            traceback_text=tb_text,
                        )
                        # 1小时内验证码导致重启超过阈值 → 停止任务并回告
                        if captcha_restart_count >= CAPTCHA_STOP_THRESHOLD:
                            logging.error(
                                "1小时内验证码导致重启 %s 次，达到阈值 %s，终止当前任务并回告",
                                captcha_restart_count, CAPTCHA_STOP_THRESHOLD
                            )
                            if mt is not None:
                                end_page = mt.get_current_page_no() if hasattr(mt, "get_current_page_no") else getattr(mt, "page", 0)
                                mt.finish_task_abnormally(
                                    end_page,
                                    f"1小时内验证码过多，累计重启{captcha_restart_count}次"
                                )
                            raise CollectionStopError(
                                f"1小时内验证码过多，累计重启{captcha_restart_count}次，终止任务"
                            )
                    else:
                        reset_captcha_restart_count(counter_key)
                    _, fail_count, _ = should_send_failure_notice(counter_key, str(e))
                    logging.warning(
                        "[设备 %s] 搜索组 %s/%s 瞬时入口异常，第%s/%s次",
                        device_id, variant_idx, len(variants), fail_count, FAILURE_NOTICE_THRESHOLD
                    )
                    # 发生异常后继续循环重试，超过上限则回告并停止
                    if cycle_no >= MAX_RUN_DEVICE_RETRIES:
                        logging.error(
                            f'[设备 {device_id}] 搜索组 {variant_idx}/{len(variants)} '
                            f'重试{cycle_no}次仍失败，停止: {e}'
                        )
                        if mt is not None:
                            end_page = mt.get_current_page_no() if hasattr(mt, "get_current_page_no") else getattr(mt, "page", 0)
                            mt.finish_task_abnormally(end_page, f"重试{cycle_no}次仍失败: {e}")
                        break
                finally:
                    if mt and hasattr(mt, 'close'):
                        mt.close()

    logging.info(f"[设备 {device_id}] 所有任务执行完毕")
    if own_scheduler and scheduler is not None:
        scheduler.stop()


class DeviceIdLogFilter(logging.Filter):
    """给每条日志注入当前设备ID（运行时读取全局 DEVICE_ID，终端传参后自动生效）"""
    def filter(self, record):
        record.device_id = DEVICE_ID
        return True


def main():
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s [%(threadName)s] [%(device_id)s] %(levelname)s: %(message)s'
    )
    # 给所有 root handler 挂上设备ID过滤器
    _device_filter = DeviceIdLogFilter()
    for _h in logging.getLogger().handlers:
        _h.addFilter(_device_filter)

    # 终端传入设备 ID：python main.py T4VK4LM7AAUOV8AY
    global DEVICE_ID
    if len(sys.argv) > 1 and sys.argv[1].strip():
        DEVICE_ID = sys.argv[1].strip()
        logging.info(f"使用终端传入设备: {DEVICE_ID}")

    # 同步设备ID到 commons.Logger（scheduler 等模块的日志带上设备ID）
    try:
        from commons.Logger import set_device_id
        set_device_id(DEVICE_ID)
    except Exception:
        pass

    # 自动模式：全局只创建一个调度器 + 一个心跳线程，所有任务复用
    scheduler = None
    if not MANUAL_MODE:
        scheduler = CrawlerScheduler(DEVICE_ID=DEVICE_ID, platform=str(PLATFORM_MT))
        scheduler.start()
        time.sleep(2)
        logging.info(f"[{DEVICE_ID}] 调度器已启动（全局心跳线程），开始获取任务...")

    try:
        while True:
            try:
                if MANUAL_MODE:
                    if DEVICE_ID not in device_list:
                        logging.error(f"设备id没有配置在 device_list 中: {DEVICE_ID}")
                        return
                    run_device(DEVICE_ID)
                else:
                    task = None
                    try:
                        task = fetch_task_from_scheduler(scheduler, DEVICE_ID)
                    except Exception as e:
                        logging.exception(f"获取任务失败: {e}，{LOOP_INTERVAL_SECONDS}秒后重试...")
                        time.sleep(LOOP_INTERVAL_SECONDS)
                        continue
                    if task is None:
                        logging.info(f"当前没有可执行的任务，{LOOP_INTERVAL_SECONDS}秒后重试...")
                        time.sleep(LOOP_INTERVAL_SECONDS)
                        continue
                    logging.info(f"获取到任务: {task.get('search_key')} (task_id={task.get('task_id')})")
                    device_list[DEVICE_ID] = [task]
                    run_device(DEVICE_ID, scheduler=scheduler)
                logging.info(f"本轮任务完成，等待 {LOOP_INTERVAL_SECONDS} 秒后开始下一轮...")
            except CollectionStopError as e:
                logging.exception(f"检测到致命停止条件，本轮终止: {e}")
            except Exception as e:
                logging.exception(f"本轮任务异常: {e}")
            time.sleep(LOOP_INTERVAL_SECONDS)
    finally:
        if scheduler is not None:
            scheduler.stop()
            logging.info("调度器已停止")

device_list = {
    "T4VK4LM7AAUOV8AY": [
        {
            "search_key": "金活 依马打正红花油",
            "title_key": "依马打正红花油",
            "spec_list": [''],
            "brand": "金活",
            "sort": "",
            "collect_range": [],
            "page_range": [],
            "platform": 4,
            "task_id": "",
            "enterprise_id": 5,
            "equipment_id": 39,
            "device_name": None,
            "collect_equipment_account_id": 13,
            "collect_region_id": 0,
            "collect_round": 1,
            "sampling_cycle": '1,4',
            "sampling_start_time": 1778774400,
            "sampling_end_time": 1793116799,
            "search_task_mode": "name_with_each_spec",
            "workflow_retry_limit": {
                "start_app": 3,
                "open_product_list_page": 3,
                "collect_single_product": 3,
            },
            "workflow_error_action": {
                "start_app": "start_app",
                "open_product_list_page": "start_app",
                "collect_single_product": "back_to_list_page",
            }
        },
    ],
}
if __name__ == '__main__':
    main()