2.0/spiders/pdd/main.py

# coding=utf-8
import requests
import base64
import uiautomator2 as u2
import time
import sys
import builtins
import subprocess
import re
import random
import json
from aip import AipOcr
import numpy as np
import cv2
import os
from pdd_config import Config
import logging
import pymysql
import datetime
import threading
import oss2
import uuid
from PIL import Image
from pathlib import Path
from test_captcha import solve as captcha_solve

from commons.Logger import get_spider_logger
from commons.feishu_webhook import send_text, send_error_card
from commons.scheduler import CrawlerScheduler

_DEFAULT_PATH = Path(__file__).with_name("city.json")

# 初始化统一日志（控制台 + logs/pdd_new2.log）
_pdd_logger = get_spider_logger("pdd")
logging.root.handlers = _pdd_logger.handlers
logging.root.setLevel(_pdd_logger.level)

# 保留原 print 的调用方式，但统一写入日志，避免到处翻控制台输出。
# 说明：
# 1. 现有代码里的 print(...) 不需要逐个改；
# 2. 默认输出会走 logging，自动带时间、级别并写文件；
# 3. 只有当 print(file=某文件对象) 时，才回退到原生 print，避免改变文件写入语义。
_ORIGINAL_PRINT = builtins.print


def print(*args, sep=" ", end="\n", file=None, flush=False):
    message = sep.join(str(arg) for arg in args)
    if end and end != "\n":
        message = f"{message}{end}"
    logged = False
    try:
        root_logger = logging.getLogger()
        if root_logger.handlers:
            root_logger.info(message.rstrip("\n"))
            logged = True
    except Exception:
        logged = False

    # 日志链路异常或未初始化时，至少保证能在控制台看到输出
    if not logged:
        _ORIGINAL_PRINT(*args, sep=sep, end=end, file=file if file else sys.stdout, flush=flush)
    if file not in (None, sys.stdout, sys.stderr):
        _ORIGINAL_PRINT(*args, sep=sep, end=end, file=file, flush=flush)




# 功能：这个模块负责从数据库拉取拼多多待执行任务，把任务分发到空闲设备，
# 然后在单设备线程中驱动 App 完成搜索、采集、校验、去重和落库。
# 边界：这里主要做调度和采集流程编排，不负责数据库表结构定义，也不负责
# OCR 服务、滑块识别服务或盒数提取逻辑本身的实现，它们都通过外部依赖完成。

# 数据库连接池（与 YBM 同库：120.24.26.108:3307/drug_retrieve_test，但不用 DictCursor）
from dbutils.pooled_db import PooledDB

_db_pool = PooledDB(
    creator=pymysql,
    maxconnections=10,
    mincached=2,
    maxcached=5,
    blocking=True,
    charset="utf8mb4",
    host="120.24.26.108",
    port=3307,
    user="root",
    password="zhijiayun123456",
    database="drug_retrieve_test",
)


def get_mysql():
    """从连接池获取数据库连接（兼容旧接口）。调用方负责 conn.close() 归还连接。"""
    return _db_pool.connection()


def get_access_token():
    AppKey = "1gAzACJOAr7BeILKqkqPOETh"
    AppSrcret = "ZNArANb9GwJYgLKg4EfYhukKBfPdl1n3"
    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



# 心跳上报配置
HEARTBEAT_INTERVAL_SECONDS = 60  # 心跳上报间隔（秒）
HEARTBEAT_API_URL = "http://120.24.26.108:8082/api/collect_task/heartbeat"  # 与 YBM 共用
SCHEDULER_INTERVAL_SECONDS = 600
MANUAL_SCHEDULER_INTERVAL_SECONDS = 300
PLATFORM_PDD = 3
TASK_STATUS_PENDING = 1
DEVICE_STATUS_IDLE = 0
DEFAULT_MAX_COUNTS_LIMIT = 300
PLATFORM_NAME_MAP = {
    PLATFORM_PDD: "拼多多",
    str(PLATFORM_PDD): "拼多多",
}


BRAND_ALIAS_MAP = {
    "小葵花": ["葵花"],
    "999": ['三九'],
}

# 手动任务模式：
# False -> 保持原有数据库取任务逻辑
# True  -> 不从数据库取任务，直接使用 MANUAL_TASKS
USE_MANUAL_TASKS = False
# 手动任务示例（按需修改，可配置多条）
# 必填：search_key；device_id 不需要填写，手动模式会自动选择空闲 PDD 设备。
MANUAL_TASKS = []

task_list = [
    ("金活依马打正红花油", "依马打正红花油", [], "金活"),("金活依马打正红花油", "依马打正红花油", [], "金活"),("金活依马打正红花油", "依马打正红花油", [], "金活"),("金活依马打正红花油", "依马打正红花油", [], "金活"),("金活依马打正红花油", "依马打正红花油", [], "金活"),("金活依马打正红花油", "依马打正红花油", [], "金活"),("金活依马打正红花油", "依马打正红花油", [], "金活"),("金活依马打正红花油", "依马打正红花油", [], "金活"),


]

for search_key, title_key, spec_list, brand in task_list:
    MANUAL_TASKS.append({
        "search_key": search_key,
        "title_key": title_key,
        "spec_list": spec_list,
        "brand": brand,
        "save_search_key": search_key,
        "enterprise_id": 8,
        "start_page": 0,
        "end_page": 500,
        "max_counts_limit": 500,
        "sort": "升序",
        "collect_round": 2,
        "collect_equipment_account_id": 1,
        "collect_region_id": 1,
    })

# 这些集合只表示"当前进程里的占用状态"。
# 数据库里设备仍可能显示空闲，因此调度前后都要结合这几份内存状态做去重。
dispatch_lock = threading.Lock()
running_task_ids = set()
running_device_ids = set()
worker_threads = {}
scheduler_stop_event = threading.Event()
scheduler_timer = None
manual_dispatched_task_ids = set()


def get_platform_name(platform):
    if platform in (None, ""):
        return "拼多多"
    return PLATFORM_NAME_MAP.get(platform, PLATFORM_NAME_MAP.get(str(platform), str(platform)))



# 功能：把外部输入安全转成整数；当值为空或格式不合法时回退到默认值。
# 输入约束：允许传入 None、空字符串、数字字符串或整数。
# 返回：成功时返回 int，失败时返回 default。
def parse_optional_int(value, default=None):
    if value in (None, ""):
        return default
    try:
        return int(value)
    except (TypeError, ValueError):
        return default


def get_adb_device_status_map():
    # 功能：读取 ADB 真实设备状态；只有状态为 device 才认为可执行自动化。
    try:
        result = subprocess.run(
            ["adb", "devices"],
            capture_output=True,
            text=True,
            timeout=10,
        )
    except Exception as e:
        logging.exception(f"读取 ADB 设备状态失败: {e}")
        return {}

    status_map = {}
    for line in result.stdout.splitlines():
        line = line.strip()
        if not line or line.startswith("List of devices"):
            continue
        parts = line.split()
        if len(parts) >= 2:
            status_map[parts[0]] = parts[1]
    return status_map


def is_adb_device_online(device_id, adb_status_map=None):
    device_id = str(device_id or "").strip()
    if not device_id:
        return False
    if adb_status_map is None:
        adb_status_map = get_adb_device_status_map()
    return adb_status_map.get(device_id) == "device"


def fetch_idle_pdd_devices():
    # 功能：读取数据库里所有空闲的拼多多设备；真实在线状态由 ADB 再二次判断。
    conn = None
    try:
        conn = get_mysql()
        with conn.cursor() as cursor:
            sql = """
                SELECT *
                FROM retrieve_collect_equipment_account
                WHERE platform = %s AND status = %s
                ORDER BY id ASC
            """
            cursor.execute(sql, (PLATFORM_PDD, DEVICE_STATUS_IDLE))
            return cursor.fetchall()
    except Exception as e:
        logging.exception(f"读取空闲 PDD 设备失败: {e}")
        return []
    finally:
        if conn:
            conn.close()


def fetch_manual_task_payloads():
    payloads = []
    idle_devices = list(fetch_idle_pdd_devices())
    adb_status_map = get_adb_device_status_map()
    idle_device_idx = 0
    for idx, task in enumerate(MANUAL_TASKS):
        task_id = parse_optional_int(task.get("task_id"), 900000 + idx + 1)
        if task_id in manual_dispatched_task_ids:
            continue

        search_key = str(task.get("search_key", "")).strip()
        if not search_key:
            logging.warning(f"跳过手动任务，缺少 search_key: {task}")
            continue

        device_row = None
        while idle_device_idx < len(idle_devices):
            candidate_device = idle_devices[idle_device_idx]
            idle_device_idx += 1
            candidate_device_id = str(candidate_device[11]).strip()
            if not is_adb_device_online(candidate_device_id, adb_status_map):
                logging.info(
                    f"手动任务 {task_id} 跳过 ADB 非在线设备 {candidate_device_id}, "
                    f"status={adb_status_map.get(candidate_device_id, 'missing')}"
                )
                continue
            with dispatch_lock:
                if candidate_device_id in running_device_ids:
                    continue
                running_task_ids.add(task_id)
                running_device_ids.add(candidate_device_id)
            device_row = candidate_device
            break

        if not device_row:
            logging.info(f"手动任务 {task_id} 没有可用空闲设备，本轮跳过")
            continue

        device_id = str(device_row[11]).strip()
        title_key = task.get("title_key")
        save_search_key = task.get("save_search_key")
        payloads.append({
            "task_id": task_id,
            "equipment_id": device_row[1],
            "enterprise_id": parse_optional_int(task.get("enterprise_id"), 0),
            "platform": parse_optional_int(task.get("platform"), PLATFORM_PDD),
            "title_key": title_key if title_key not in (None, "") else search_key,
            "spec_list": task.get("spec_list", ""),
            "brand": task.get("brand", ""),
            "search_key": search_key,
            "save_search_key": save_search_key if save_search_key not in (None, "") else search_key,
            "start_page": parse_optional_int(task.get("start_page"), 0),
            "end_page": parse_optional_int(task.get("end_page"), None),
            "max_counts_limit": parse_optional_int(task.get("max_counts_limit"), DEFAULT_MAX_COUNTS_LIMIT),
            "collect_config_info": task.get("collect_config_info", ""),
            'collect_equipment_account_id': task.get("collect_equipment_account_id", ""),
            'collect_region_id':task.get("collect_region_id", ""),
            'collect_round': task.get("collect_round", ""),
            "sort": task.get("sort", "默认"),
            "device_id": device_id,
            "task_row": None,
            "direct_shop_lookup": bool(task.get("direct_shop_lookup", False)),
        })
        manual_dispatched_task_ids.add(task_id)

    if payloads:
        logging.info(f"手动任务模式启用，本轮分发 {len(payloads)} 条任务")
    else:
        logging.info("手动任务模式启用，但没有可分发的新任务")
    return payloads

def fetch_pending_tasks():
    """通过调度 API 为每个空闲 PDD 设备拉取任务（替代原 DB 查询 retrieve_collect_task_allocate）。"""
    tasks = []
    idle_devices = fetch_idle_pdd_devices()
    if not idle_devices:
        logging.info("当前没有空闲 PDD 设备")
        return tasks

    for device_row in idle_devices:
        username = device_row[4]   # 调度系统注册的账号名（如 pdd_1）
        device_id = device_row[11] or username  # ADB 序列号，空时用 username 兜底
        if not username:
            continue
        with dispatch_lock:
            if device_id in running_device_ids:
                continue
        # 用 username 向调度 API 拉任务（先发心跳注册账号，跟 YBM 一样）
        scheduler = CrawlerScheduler(username, str(PLATFORM_PDD))
        requests.post(
            scheduler.heartbeat_url,
            json={"platform": scheduler.platform, "username": scheduler.username},
            headers={'X-Crawler-Token': 'zhijiayun_crawler_2026'},
            timeout=5,
        )
        task = scheduler.get_task()
        if task:
            task['_device_row'] = device_row
            task['_device_id'] = device_id
            tasks.append(task)

    return tasks

def fetch_idle_device_by_equipment_id(equipment_id):
    # 功能：按设备 id 查询指定终端是否空闲，避免任务和设备错配。
    # 调用 get_mysql() 的目的是查询指定设备是否空闲，避免把任务错误派发到其他终端。
    conn = None
    try:
        conn = get_mysql()
        with conn.cursor() as cursor:
            sql = """
                SELECT *
                FROM retrieve_collect_equipment_account
                WHERE  collect_equipment_id = %s AND status = %s
                LIMIT 1
            """
            cursor.execute(sql, ( equipment_id, DEVICE_STATUS_IDLE))
            return cursor.fetchone()
    except Exception as e:
        logging.exception(f"读取空闲设备失败 equipment_id={equipment_id}: {e}")
        return None
    finally:
        if conn:
            conn.close()

def build_task_payload(task_row, device_row):
    # 功能：把数据库原始任务行和设备行整理成线程入口可直接消费的任务上下文。
    # 返回：统一字段名的字典，避免后续线程逻辑继续依赖固定列下标。
    sampling_cycle = str(task_row[12]).strip() if len(task_row) > 12 else ''
    sampling_start_time = parse_optional_int(task_row[13] if len(task_row) > 13 else None, 0)
    sampling_end_time = parse_optional_int(task_row[14] if len(task_row) > 14 else None, 0)
    start_page = parse_optional_int(task_row[15] if len(task_row) > 15 else None, 0)
    end_page = parse_optional_int(task_row[16] if len(task_row) > 16 else None, 300)
    max_counts_limit = parse_optional_int(
        task_row[17] if len(task_row) > 17 else None,
        DEFAULT_MAX_COUNTS_LIMIT
    )
    collect_config_info = json.dumps({
        "sampling_cycle": sampling_cycle,
        "sampling_start_time": sampling_start_time,
        "sampling_end_time": sampling_end_time,
    }, ensure_ascii=False)


    return {
        "task_id": task_row[0],
        "task_name": task_row[1],
        "collect_equipment_account_id": task_row[3],
        "collect_region_id": task_row[4],
        "equipment_id": task_row[2],
        "enterprise_id": task_row[5],
        "platform": task_row[6],
        "title_key": task_row[7],
        "spec_list": task_row[8],
        "brand": task_row[9],
        "search_key": f"{task_row[9]}{task_row[7]} {task_row[8] or ''}".strip(),
        "save_search_key": f"{task_row[9]}{task_row[7]} {task_row[8] or ''}".strip(),
        "start_page": start_page,
        "end_page": end_page,
        "max_counts_limit": max_counts_limit,
        "collect_config_info": collect_config_info,
        "collect_round": task_row[24],
        "sort": "升序",
        "device_id": device_row[11],
        "task_row": task_row,
        }


def build_task_payload_from_api(task_dict, device_row):
    """从调度 API 返回的 task dict + DB 设备行构建 worker 线程可消费的 payload。"""
    username = device_row[4]    # 调度系统账号名（如 pdd_1）
    device_id = device_row[11] or username  # ADB 序列号，空时用 username 兜底
    search_key = (
        f"{task_dict.get('product_brand', '')}"
        f"{task_dict.get('product_name', '')} "
        f"{task_dict.get('product_specs', '') or ''}"
    ).strip()
    spec_raw = task_dict.get('product_specs', '') or ''
    spec_list = [s.strip() for s in re.split(r'[|、，,\n\r]+', spec_raw) if s.strip()]

    collect_config_info = json.dumps({
        "sampling_cycle": task_dict.get('sampling_cycle', ''),
        "sampling_start_time": task_dict.get('sampling_start_time', 0),
        "sampling_end_time": task_dict.get('sampling_end_time', 0),
    }, ensure_ascii=False)

    return {
        "task_id": task_dict.get('id'),
        "search_key": search_key,
        "title_key": task_dict.get('product_name'),
        "spec_list": spec_list,
        "brand": task_dict.get('product_brand', ''),
        "save_search_key": search_key,
        "start_page": task_dict.get('current_page', 0),
        "end_page": None,
        "max_counts_limit": DEFAULT_MAX_COUNTS_LIMIT,
        "collect_config_info": collect_config_info,
        "platform": PLATFORM_PDD,
        "enterprise_id": task_dict.get('company_id'),
        "collect_round": task_dict.get('collect_round'),
        "collect_equipment_account_id": task_dict.get('collect_equipment_account_id', 1),
        "collect_region_id": task_dict.get('collect_region_id', 1),
        "sort": "升序",
        "username": username,   # 调度 API 账号名
        "device_id": device_id,  # ADB 序列号
        "device_row": device_row,
    }


def fetch_runnable_task_payloads():
    """通过调度 API 拉取任务，结合 DB 设备空闲状态，生成可执行的 payload 列表。"""
    if USE_MANUAL_TASKS:
        logging.info("手动任务模式启用，跳过 API 任务分配")
        return fetch_manual_task_payloads()

    api_tasks = fetch_pending_tasks()
    if not api_tasks:
        logging.info("当前没有待执行任务")
        return []

    payloads = []
    for task_dict in api_tasks:
        device_row = task_dict.get('_device_row')
        device_id = task_dict.get('_device_id')
        task_id = task_dict.get('id')

        if not device_row or not device_id:
            continue

        with dispatch_lock:
            if device_id in running_device_ids:
                logging.info(f"设备 {device_id} 已在本进程执行任务，跳过任务 {task_id}")
                continue
            running_device_ids.add(device_id)

        payloads.append(build_task_payload_from_api(task_dict, device_row))

    return payloads

def cleanup_finished_workers():
    # 功能：同步 worker_threads / running_device_ids / running_task_ids，
    # 只保留仍然存活的线程引用，避免失效线程长期占位导致设备和任务被永久跳过。
    with dispatch_lock:
        dead_device_ids = [
            device_id
            for device_id, thread in worker_threads.items()
            if not thread.is_alive()
        ]
        for device_id in dead_device_ids:
            worker_threads.pop(device_id, None)
            running_device_ids.discard(device_id)
        orphan_device_ids = [
            device_id
            for device_id in running_device_ids
            if device_id not in worker_threads
        ]
        for device_id in orphan_device_ids:
            running_device_ids.discard(device_id)
        if not worker_threads and running_task_ids:
            logging.info(
                f"cleanup: 无存活 worker，清空残留 running_task_ids: {running_task_ids}"
            )
            running_task_ids.clear()

def wait_for_active_workers(check_interval=1):
    while True:
        cleanup_finished_workers()
        with dispatch_lock:
            alive_threads = [thread for thread in worker_threads.values() if thread.is_alive()]
        if not alive_threads:
            return
        time.sleep(check_interval)


def has_active_workers():
    cleanup_finished_workers()
    with dispatch_lock:
        return any(thread.is_alive() for thread in worker_threads.values())


def all_manual_tasks_dispatched():
    expected_task_ids = {
        parse_optional_int(task.get("task_id"), 900000 + idx + 1)
        for idx, task in enumerate(MANUAL_TASKS)
    }
    return expected_task_ids.issubset(manual_dispatched_task_ids)



def run_task_worker(task_payload):
    """单个任务线程的主入口：通过 CrawlerScheduler 维持心跳，驱动 PDD 采集并上报。"""
    task_id = task_payload["task_id"]
    device_id = task_payload["device_id"]      # ADB 序列号（u2.connect_usb 用）
    username = task_payload.get("username", device_id)  # 调度系统账号名
    search_key = str(task_payload.get("search_key", "")).strip()
    pdd = None
    should_send_finish_notice = False

    # 每个设备线程拥有自己的 CrawlerScheduler 实例，负责心跳 + 上报
    scheduler = CrawlerScheduler(username, str(PLATFORM_PDD))
    scheduler.start()  # 启动后台心跳线程

    try:
        logging.info(f"[任务 {task_id}] 开始执行，设备: {device_id}")
        print(task_payload)
        pdd = PDD(
            task_payload["search_key"],
            device_id,
            title_key=task_payload.get("title_key"),
            spec_list=task_payload.get("spec_list"),
            brand=task_payload.get("brand", ""),
            save_search_key=task_payload.get("save_search_key"),
            start_page=task_payload.get("start_page"),
            end_page=task_payload.get("end_page"),
            max_counts_limit=task_payload.get("max_counts_limit"),
            collect_config_info=task_payload.get("collect_config_info", ""),
            direct_shop_lookup=task_payload.get("direct_shop_lookup", False),
            sort=task_payload.get("sort"),
            platform=task_payload.get("platform"),
            task_id=task_payload.get("task_id"),
            enterprise_id=task_payload.get("enterprise_id"),
            collect_round=task_payload.get("collect_round"),
            collect_equipment_account_id=task_payload.get("collect_equipment_account_id", 1),
            collect_region_id=task_payload.get("collect_region_id", 1),
            username=task_payload.get("username"),
        )
        completed_normally = pdd.main(device_id, 1, 0)
        if completed_normally:
            logging.info(f"[任务 {task_id}] 执行完成，设备: {device_id}")
        else:
            logging.info(f"[任务 {task_id}] 已结束，设备: {device_id}")
        should_send_finish_notice = True
    except Exception as e:
        end_page = task_payload.get("start_page")
        err_msg = str(e)
        if pdd is not None:
            end_page = getattr(pdd, "page", end_page)
            pdd.finish_task_abnormally(end_page, f"任务执行异常: {err_msg}")
        else:
            report_api(task_id, platform=PLATFORM_PDD, username="unknown",
                       page=end_page, is_finished=0)
        logging.exception(f"[任务 {task_id}] 执行异常，设备: {device_id}，错误: {err_msg}")
        send_error_card(
            task_name=f"PDD任务(task_id={task_id}, device={device_id}, key={search_key})",
            err_msg=err_msg,
            mention_all=False,
        )
        should_send_finish_notice = True
    finally:
        scheduler.stop()  # 停止心跳
        # 释放内存占用标记
        with dispatch_lock:
            running_device_ids.discard(device_id)
            worker_threads.pop(device_id, None)
        if pdd is not None:
            try:
                pdd.ensure_home_before_task_end(max_rounds=6)
            except Exception as cleanup_error:
                logging.info(f"[任务 {task_id}] 收尾回主页面失败: {cleanup_error}")
        if should_send_finish_notice:
            scraped_count = getattr(pdd, "max_counts", 0) if pdd is not None else 0
            notify_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            platform_name = get_platform_name(task_payload.get("platform"))
            drug_name = str(task_payload.get("search_key", "")).strip()
            send_text(
                f"{notify_time} 通知:\n"
                f"平台: {platform_name}, 药品: {drug_name}, 爬取数据: {scraped_count}条"
            )


def dispatch_pending_tasks():
    # 功能：执行一轮派单，把每个可运行任务绑定到对应设备线程。
    # 阶段 1：先清理已经结束的线程引用，避免占用状态过期。
    cleanup_finished_workers()
    # 阶段 2：重新计算本轮真正可运行的任务集合。
# 手动改任务处
    task_payloads = fetch_runnable_task_payloads()
#     task_payloads = [
#         {
#             "task_id": 1,
#             "equipment_id": 1,
#             "enterprise_id": 1,
#             "platform": 3,
#             "title_key": '依马打正红花油',
#             "spec_list": '25ml',
#             "brand": '金活',
#             "search_key": '金活 依马打正红花油 25ml',
#             "save_search_key": '依马打正红花油',
#             "start_page": 0,
#             "end_page": 6,
#             "max_counts_limit": 30,
#             "collect_config_info": [],
#             'collect_equipment_account_id': 1,
#             'collect_region_id': 1,
#             'collect_round': 1,
#             "sort": 1,
#             "device_id": 'XOYPOZDADU79VGVG',
#             "task_row": None,
#             "direct_shop_lookup": 1,
#         },
#     {
#             "task_id": 1,
#             "equipment_id": 1,
#             "enterprise_id": 1,
#             "platform": 3,
#             "title_key": '依马打正红花油',
#             "spec_list": '25ml',
#             "brand": '金活',
#             "search_key": '金活 依马打正红花油 25ml',
#             "save_search_key": '依马打正红花油',
#             "start_page": 0,
#             "end_page": 6,
#             "max_counts_limit": 30,
#             "collect_config_info": [],
#             'collect_equipment_account_id': 1,
#             'collect_region_id': 1,
#             'collect_round': 1,
#             "sort": 1,
#             "device_id": 'U47HZDRG8XJBBURW',
#             "task_row": None,
#             "direct_shop_lookup": 1,
#         },
#     {
#             "task_id": 1,
#             "equipment_id": 1,
#             "enterprise_id": 1,
#             "platform": 3,
#             "title_key": '依马打正红花油',
#             "spec_list": '25ml',
#             "brand": '金活',
#             "search_key": '金活 依马打正红花油 25ml',
#             "save_search_key": '依马打正红花油',
#             "start_page": 0,
#             "end_page": 6,
#             "max_counts_limit": 30,
#             "collect_config_info": [],
#             'collect_equipment_account_id': 1,
#             'collect_region_id': 1,
#             'collect_round': 1,
#             "sort": 1,
#             "device_id": 'U8ONIJJJS4CELVD6',
#             "task_row": None,
#             "direct_shop_lookup": 1,
#         }
#     ]
    if not task_payloads:
        return

    for task_payload in task_payloads:
        device_id = task_payload["device_id"]
        try:
            # 阶段 3：为每个任务创建独立线程，让不同设备可以并发执行采集。
            thread = threading.Thread(
                target=run_task_worker,
                args=(task_payload,),
                daemon=True,
                name=f"pdd-{device_id}",
            )
            with dispatch_lock:
                worker_threads[device_id] = thread
            thread.start()
            logging.info(f"[任务 {task_payload['task_id']}] 已分发到设备 {device_id}")
        except Exception:
            with dispatch_lock:
                # 线程创建失败时同步回滚占用状态，避免任务被"卡住"。
                running_task_ids.discard(task_payload["task_id"])
                running_device_ids.discard(device_id)
                worker_threads.pop(device_id, None)
            raise

def schedule_dispatch(delay_seconds=SCHEDULER_INTERVAL_SECONDS, job_func=None):
    # 功能：安排下一次轮询触发时间。
    # 功能：注册下一轮调度定时器，让调度器按固定间隔持续轮询。
    global scheduler_timer
    if scheduler_stop_event.is_set():
        return
    if job_func is None:
        job_func = scheduled_dispatch_job
    # 采用递归定时器而不是死循环，便于后续统一停机。
    scheduler_timer = threading.Timer(delay_seconds, job_func)
    scheduler_timer.daemon = False
    scheduler_timer.name = "pdd-scheduler"
    scheduler_timer.start()

def scheduled_dispatch_job():
    # 功能：执行一次定时派单，并保证下一轮轮询仍会继续注册。
    # 功能：包装一次定时调度执行，确保本轮出错也不会中断后续轮询。
    try:
        dispatch_pending_tasks()
    except Exception as e:
        logging.exception(f"PDD 定时调度异常: {e}")
    finally:
        schedule_dispatch(SCHEDULER_INTERVAL_SECONDS)

def manual_scheduled_dispatch_job():
    try:
        dispatch_pending_tasks()
    except Exception as e:
        logging.exception(f"PDD 手动任务定时调度异常: {e}")
    finally:
        if all_manual_tasks_dispatched() and not has_active_workers():
            logging.info("手动任务模式：全部任务已完成，停止轮询并退出")
            scheduler_stop_event.set()
            return
        schedule_dispatch(MANUAL_SCHEDULER_INTERVAL_SECONDS, manual_scheduled_dispatch_job)

# 调度上报配置（与 YBM 共用同一套 API）
TASK_REPORT_URL = "http://120.24.26.108:8082/api/collect_task/report"
TASK_HEARTBEAT_URL = "http://120.24.26.108:8082/api/collect_task/heartbeat"
TASK_API_HEADERS = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}


def report_api(task_id, platform, username, page=0, is_finished=0,
               crawled_count=0, total_pages=0):
    """向调度中心上报任务状态（完全对齐 YBM _send_report 格式）。"""
    if USE_MANUAL_TASKS:
        logging.info(f"手动任务模式：跳过接口上报 task_id={task_id}")
        return

    params = {
        "task_id": task_id,
        "platform": str(platform),
        "username": str(username) if username else "",
        "current_page": page,
        "total_pages": total_pages,
        "crawled_count": crawled_count,
        "is_finished": is_finished,
    }
    print(params)
    try:
        res = requests.post(TASK_REPORT_URL, json=params, headers=TASK_API_HEADERS, timeout=20)
        print(res.text)
    except Exception as e:
        logging.exception(f"上报接口异常 task_id={task_id}: {e}")


def report_heartbeat(task_id, page=None, device_id=None):
    """上报任务心跳"""
    if USE_MANUAL_TASKS:
        return

    params = {
        "collect_task_allocate_id": task_id,
        "heartbeat_time": int(time.time()),
        "current_page": page if page is not None else 0,
        "device_id": device_id if device_id else "",
    }

    try:
        requests.post(TASK_HEARTBEAT_URL, json=params, headers=TASK_API_HEADERS, timeout=10)
        logging.debug(f"心跳上报 task_id={task_id}, page={page}")
    except Exception as e:
        logging.warning(f"心跳上报失败: {e}")
# 获取滑块验证中滑块需要移动的距离
def slide_verify(img_path):
    # 功能：把本地截图交给第三方打码服务，换回滑块缺口位移。
    # 返回：识别成功时返回滑块位移结果，失败时返回服务端给出的空结果。
    with open(img_path, 'rb') as f:
        b = base64.b64encode(f.read()).decode()  ## 图片二进制流base64字符串
    url = "http://api.jfbym.com/api/YmServer/customApi"
    data = {
        ## 关于参数,一般来说有3个;不同类型id可能有不同的参数个数和参数名,找客服获取
        "token": "1nDVocTE2mJ0yLEYb2sZJ5uUY2VIEoGTkIpW44X7Kgk",
        "type": "22222",
        "image": b,
    }
    _headers = {
        "Content-Type": "application/json"
    }
    # 识别结果来自第三方服务，当前逻辑只读取它约定的 msg/data 字段。
    response = requests.request("POST", url, headers=_headers, json=data).json()
    print(response)

    if response.get("msg") == "识别成功":
        # 获取 data 中的 data 字段
        result = response.get("data", {}).get("data")
        if result:
            print(result)  # 输出结果
        else:
            print("无法获取数据")
    else:
        print("识别未成功")

    return result

class PDD:
    # 功能：这个类负责维护单次拼多多采集任务的运行状态，并封装设备连接、
    # 页面导航、数据提取、说明书解析、去重校验和写库等操作。
    # 边界：这个类负责"如何跑完整个采集流程"，但不负责外部调度器的轮询策略，
    # 也不负责 OCR/盒数提取等外部依赖的底层实现。
    def __init__(
            self,
            search_key,
            device_id,
            title_key=None,
            spec_list=None,
            brand="",
            save_search_key=None,
            start_page=0,
            end_page=None,
            max_counts_limit=None,
            collect_config_info="",
            direct_shop_lookup=False,
            sort=None,
            platform = None,
            task_id = None,
            enterprise_id=None,
            collect_round=None,
            collect_equipment_account_id=None,
            collect_region_id=None,
            username=None,

    ):
        # 阶段 1：初始化与 App、OCR、日志、OSS 相关的基础依赖。
        self.username = username  # 调度系统账号名
        self.package_name = 'com.xunmeng.pinduoduo'
        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.table_name = "retrieve_scrape_data" # "pdd_drug"
        self.shop_table_name = "retrieve_scrape_shop_info"  # "pdd_shop_info"

        self.loggerPdd = logging.getLogger()

        self.clipboard = ""  # 初始化剪切板的内容为空

        # 阶段 2：固化本次任务的筛选条件、页码边界和搜索参数。
        self.enterprise_id = enterprise_id
        self.task_id = task_id
        self.platform = platform
        self.sort = sort
        self.sort_key = 0
        self.search_key = search_key  # 参苓健脾胃颗粒 香砂平胃颗粒 舒肝颗粒 清肺化痰丸
        # title_key 支持把"搜索词"和"标题过滤词"拆开；未单独传入时回退到搜索词。
        self.title_key = title_key if title_key is not None else search_key
        # 规格统一整理成列表，后续匹配逻辑就不用再分辨单值、列表和空值三种输入。
        self.spec_list = self._normalize_rule_list(spec_list)
        # 统一做一次清洗，避免 None/空白写入数据库。
        self.brand = str(brand or "").strip()
        self.save_search_key = save_search_key or search_key
        # 起止页在入口阶段先做边界修正，主循环只消费规范化后的值。
        self.start_page = max(parse_optional_int(start_page, 0), 0)
        self.end_page = parse_optional_int(end_page, None)  # None = 不限页数
        self.max_counts_limit = max_counts_limit
        self.collect_config_info = collect_config_info or ""
        self.direct_shop_lookup = direct_shop_lookup
        # 任务轮次（数据库字段 collect_round）
        self.collect_round = collect_round
        self.collect_equipment_account_id = collect_equipment_account_id
        self.collect_region_id = collect_region_id

        self.unrelated_data = 0  # 无关数据数量
        self.device_id = device_id
        self.page = self.start_page
        self._city_lookup_cache = None
        self._city_lookup_path = ""


        # 阶段 3：初始化运行时统计状态，这些状态会在主循环中持续更新。
        # 统计售罄数量
        self.sold_out_counts = 0

        # 程序启动时间
        self.program_start_time = self.app_start_time()

        # 统计商品数量
        # 最大量数据阈值
        self.max_counts = 0

        # 统计点击商品的次数
        self.click_counts = 0

        # 商品在列表的位置
        self.search_key_loc = 0
        # finish_reported 用来保证无论走哪个退出分支，只向调度系统上报一次结束状态。
        self.finish_reported = False

        # oss配置
        self.oss_config = {
            "access_key_id": Config.access_key_id,
            "access_key_secret": Config.access_key_secret,
            "endpoint": Config.endpoint,  # 例: oss-cn-beijing.aliyuncs.com
            "bucket_name": Config.bucket_name,
            "oss_prefix": Config.oss_prefix  # OSS中存放截图的前缀(虚拟文件夹)
        }
        self.address_region_index = self.build_address_region_index(_DEFAULT_PATH)

    # 异常处理
    def wr_re(self, mod, device_id, sort=None, page=None):
        # 功能：读写或删除本地进度文件，给断点续跑保留入口。
        file_path = f'./ycwj/{device_id}_{self.title_key}.txt'
        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 == "读":
            try:
                if not os.path.exists(file_path):
                    return None
                with open(file_path, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    print(self.sort)
                    # if self.sort and self.sort_key == 0:
                    #     self.li_or_lo(self.sort)
                    if data['page'] != '':
                        progress_page = int(data['page'])
                        self.page = max(progress_page, self.start_page)
                        self.scroll_to_target_page(self.page)
                    else:
                        return None
                return data
            except Exception as e:
                print(f"读取进度失败", e)
                return None
        elif mod == "删":
            try:
                if os.path.exists(file_path):
                    os.remove(file_path)
                    print(f"进度文件已删除:{file_path}")
            except Exception as e:
                print(f"删除进度文件失败:{e}")
        return None

    def clear_progress_file(self):
        # 功能：任务正常完成后清理断点进度文件。
        self.wr_re("删", self.device_id, self.sort)

    def is_max_count_reached(self):
        # 功能：判断当前采集数量是否达到任务设定上限。
        return bool(self.max_counts_limit and self.max_counts >= self.max_counts_limit)

    def scroll_to_target_page(self, target_page):
        # 功能：按目标页数执行固定次数滑动，用于恢复上次采集的大致位置。
        target_page = int(target_page or 0)
        if target_page <= 0:
            return
        # 这里按"页数约等于滑动次数"的经验规则恢复列表位置，宁可保守，不做复杂校准。
        for _ in range(target_page):
            screen_w, screen_h = self.get_screen_size()
            end_y = int(screen_h * 0.18)
            self.d.swipe(screen_w // 2, int(screen_h * 0.85), screen_w // 2, end_y, 0.4)
            time.sleep(self.get_sleep_time())

    def finish_task_normally(self, end_page, reason):
        if not self.finish_reported:
            report_api(
                self.task_id,
                platform=self.platform,
                username=getattr(self, 'username', ''),
                page=end_page,
                is_finished=1,
                crawled_count=getattr(self, 'max_counts', 0),
                total_pages=end_page,
            )
            self.finish_reported = True
        print(reason)
        return True

    def finish_task_abnormally(self, end_page, reason, finish_status=0):
        if not self.finish_reported:
            report_api(
                self.task_id,
                platform=self.platform,
                username=getattr(self, 'username', ''),
                page=end_page,
                is_finished=0,
                crawled_count=getattr(self, 'max_counts', 0),
                total_pages=end_page,
            )
            self.finish_reported = True
        print(reason)
        return False

    def finish_task_with_max_count(self, end_page):
        # 功能：达到采集上限时复用正常结束逻辑，只替换结束原因。
        return self.finish_task_normally(
            end_page,
            f"达到最大采集数量 {self.max_counts_limit}，当前已采集 {self.max_counts} 条，停止任务"
        )

    # 排序
    def li_or_lo(self, key):
        # 功能：进入搜索结果页后切换价格排序方式。

        if key == "升序":
            self.sort_key += 1
            self.d.xpath('//*[@text="价格"]').click()
            n = self.d.xpath('//*[@text="总价低到高"]')
            if n.exists:
                n.click()
            time.sleep(self.get_sleep_time())

        if key == "降序":
            self.sort_key += 1
            self.d.xpath('//*[@text="价格"]').click()
            n = self.d.xpath('//*[@text="单粒价格低到高"]')
            if n:
                n.click()
            else:
                self.d.xpath('//*[@text="价格"]').click()

    # 返回列表页
    def back_to_list_page(self):
        # 功能：通过多次尝试返回键，尽量把页面恢复到商品列表页。
        try:
            for i in range(10):
                if self.distinct_target():
                    return True
                if self.d(className='android.widget.EditText').exists and not self.d.xpath('//*[@text="筛选"]').exists:
                    print('当前已进入搜索输入页，停止继续返回，交给上层重新进入搜索结果页')
                    return False
                print(f'第{i}次尝试退回到列表页')
                self.swipe_back(1)
        except:
            if self.is_pdd_home_page():
                print('当前已回到首页，停止继续返回，交给上层重新进入搜索结果页')
                return False
            else:
                print('页面出错，没有退回到列表页')
        return False

    def get_drug_lis(self, idx):
        # 功能：根据当前页布局拿到可点击的商品卡片列表。
        # 这里区分 idx==0 和后续页面，是因为首屏与翻页后的 RecyclerView 层级不完全一致。
        drug_lis = []
        if idx == 0:
            drug_lis = self.d.xpath(
                '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.FrameLayout[2]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout').all()
        else:
            for i in range(1, 6):
                drug_lis = self.d.xpath(
                    f'/hierarchy/android.widget.FrameLayout[{i}]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.support.v7.widget.RecyclerView[1]/android.widget.FrameLayout').all()
                if drug_lis:
                    break
        if drug_lis:
            return drug_lis

        fallback_xpaths = [
            '//android.support.v7.widget.RecyclerView/android.widget.FrameLayout',
            '//androidx.recyclerview.widget.RecyclerView/android.widget.FrameLayout',
            '//*[@class="android.support.v7.widget.RecyclerView"]/android.widget.FrameLayout',
            '//*[@class="androidx.recyclerview.widget.RecyclerView"]/android.widget.FrameLayout',
        ]
        for fallback_xpath in fallback_xpaths:
            drug_lis = self.d.xpath(fallback_xpath).all()
            if drug_lis:
                logging.info(f"get_drug_lis: fallback xpath 命中 {fallback_xpath}，数量={len(drug_lis)}")
                break
        return drug_lis

    # 【新增】图片拼接辅助函数（垂直拼接两张截图）
    def merge_two_images(self, img1_path, img2_path, output_path):
        """
        垂直拼接两张截图（统一宽度，高度累加）
        :param img1_path: 第一张图路径
        :param img2_path: 第二张图路径
        :param output_path: 拼接后输出路径
        """
        try:
            # 打开两张图片
            img1 = Image.open(img1_path)
            img2 = Image.open(img2_path)

            # 统一宽度（以较宽的图为准）
            max_width = max(img1.width, img2.width)
            img1 = img1.resize((max_width, int(img1.height * max_width / img1.width)), Image.Resampling.LANCZOS)
            img2 = img2.resize((max_width, int(img2.height * max_width / img2.width)), Image.Resampling.LANCZOS)

            # 创建拼接后的画布（垂直拼接）
            merge_height = img1.height + img2.height
            merge_img = Image.new('RGB', (max_width, merge_height))
            merge_img.paste(img1, (0, 0))
            merge_img.paste(img2, (0, img1.height))

            # 保存拼接后的图片
            merge_img.save(output_path, quality=95)
            self.loggerPdd.info(f"✅ 图片拼接完成：{output_path}")
            return True
        except Exception as e:
            self.loggerPdd.error(f"❌ 图片拼接失败：{str(e)}")
            return False

    # 【新增】图片压缩辅助函数（基于cv2，控制文件大小）
    def compress_image(self, input_path, output_path, target_size_kb=100, max_quality=80, min_quality=20):
        """
        压缩图片到指定大小（默认100KB以内），先缩尺寸再调质量
        :param input_path: 原始图片路径
        :param output_path: 压缩后输出路径
        :param target_size_kb: 目标大小（KB），默认100
        :param max_quality: 初始最大质量（1-100）
        :param min_quality: 最低质量（避免过度模糊）
        :return: bool: 压缩是否成功
        """
        try:
            # 1. 读取原始图片
            img = cv2.imread(input_path)
            if img is None:
                self.loggerPdd.error(f"❌ 读取图片失败：{input_path}")
                return False

            # 2. 先按比例缩小尺寸（核心：降低分辨率）
            height, width = img.shape[:2]
            # 先缩放到原尺寸的70%（可根据需要调整，比如60%/50%）
            scale = 0.7  # 缩放比例，越小文件越小（0.7=70%尺寸）
            new_width = int(width * scale)
            new_height = int(height * scale)
            img_resized = cv2.resize(
                img, (new_width, new_height),
                interpolation=cv2.INTER_AREA  # 缩小用INTER_AREA更清晰
            )

            # 3. 动态调整质量，直到文件≤target_size_kb或达到最低质量
            quality = max_quality
            ext = os.path.splitext(output_path)[1].lower()
            encode_param = [cv2.IMWRITE_JPEG_QUALITY, quality]

            while quality >= min_quality:
                # 保存图片
                cv2.imwrite(output_path, img_resized, encode_param)

                # 检测文件大小
                file_size = os.path.getsize(output_path) / 1024  # 转KB
                self.loggerPdd.info(f"🔍 质量{quality}，文件大小：{file_size:.2f}KB")

                # 达标则退出循环
                if file_size <= target_size_kb:
                    self.loggerPdd.info(f"✅ 图片压缩完成：{output_path}（最终质量：{quality}，大小：{file_size:.2f}KB）")
                    return True

                # 未达标则降低质量（每次降5，可调整步长）
                quality -= 5
                encode_param[1] = quality

            # 若到最低质量仍超标，再缩小尺寸（兜底）
            self.loggerPdd.warning(f"⚠️ 最低质量{min_quality}仍超标，再次缩小尺寸到50%")
            new_width_2 = int(new_width * 0.5)
            new_height_2 = int(new_height * 0.5)
            img_resized_2 = cv2.resize(img_resized, (new_width_2, new_height_2), cv2.INTER_AREA)
            encode_param[1] = min_quality
            cv2.imwrite(output_path, img_resized_2, encode_param)
            final_size = os.path.getsize(output_path) / 1024
            self.loggerPdd.info(f"✅ 兜底压缩完成：{output_path}（质量：{min_quality}，大小：{final_size:.2f}KB）")
            return True

        except Exception as e:
            self.loggerPdd.error(f"❌ 图片压缩失败：{str(e)}")
            return False
    # 从商品顶部直接慢速向下滚动截长图 → 堆叠拼接 → 压缩 → 上传OSS
    def screenshot_and_upload_oss(self, screenshot_desc='shop_name_screenhshot'):
        """
        1. 从当前位置慢速向下滚动截屏 → 2. 堆叠拼接 → 3. 压缩 → 4. 上传OSS → 5. 清理本地文件
        :param screenshot_desc: 截图描述（用于OSS文件名）
        :return: oss_url: 上传后的OSS地址 | None: 失败
        """
        timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
        unique_id = uuid.uuid4().hex[:8]
        base_name = f"{screenshot_desc}_{self.device_id}_{timestamp}_{unique_id}"

        # 定义临时文件路径
        scroll_screenshot_path = f"./{base_name}_scroll.png"  # 滚动长图
        compress_path = f"./{base_name}_compress.jpg"  # 压缩后的图

        # 初始化OSS客户端（若未初始化）
        if not hasattr(self, 'oss_bucket'):
            self.oss_auth = oss2.Auth(
                self.oss_config["access_key_id"],
                self.oss_config["access_key_secret"]
            )
            self.oss_bucket = oss2.Bucket(
                self.oss_auth,
                self.oss_config["endpoint"],
                self.oss_config["bucket_name"]
            )
        else:
            print("oss初始化完毕")

        try:
            # ===================== 步骤1：滚动截屏 =====================
            self.loggerPdd.info("📸 开始滚动截屏...")
            try:
                screenshot_image = self._scroll_screenshot()
                screenshot_image.save(scroll_screenshot_path)
            except Exception as e:
                self.loggerPdd.error(f"滚动截图失败：{str(e)}")
                return None

            self.loggerPdd.info(f"✅ 滚动截图完成：{scroll_screenshot_path}")

            # ===================== 步骤2：压缩图片 =====================
            if not self.compress_image(scroll_screenshot_path, compress_path, target_size_kb=100):
                raise Exception("图片压缩失败")

            # ===================== 步骤3：上传压缩后的图片到OSS =====================
            oss_file_path = f"{self.oss_config['oss_prefix']}/{os.path.basename(compress_path)}"
            # 上传文件到OSS
            self.oss_bucket.put_object_from_file(oss_file_path, compress_path)
            # 生成OSS访问URL
            oss_url = f"https://{self.oss_config['bucket_name']}.{self.oss_config['endpoint']}/{oss_file_path}"
            self.loggerPdd.info(f"🚀 OSS上传成功：{oss_url}")

            return oss_url

        except Exception as e:
            self.loggerPdd.error(f"❌ 截图/压缩/上传OSS失败：{str(e)}")
            return None

        finally:
            # ===================== 步骤4：清理所有本地临时文件 =====================
            for file_path in [scroll_screenshot_path, compress_path]:
                if os.path.exists(file_path):
                    try:
                        os.remove(file_path)
                        self.loggerPdd.info(f"🗑️ 清理临时文件：{file_path}")
                    except Exception as e:
                        self.loggerPdd.warning(f"⚠️ 清理临时文件失败 {file_path}：{str(e)}")

    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

        self.loggerPdd.info(f"✅ 拼接完成，尺寸: {total_width} x {total_height}")
        return merged_img

    def _scroll_screenshot(self, scroll_times=1):
        """
        1. 截第一张（当前位置）
        2. 如果「进店」已经在首屏 → 直接返回，不滚动
        3. 否则最多滚动 scroll_times 次（默认 1 次），到「进店」即停止
        4. 堆叠拼接
        """
        self.loggerPdd.info(f"📸 滚动截图开始（最多 {scroll_times} 次滚动）...")

        w, h = self.d.window_size()

        # 第1张：当前位置
        screen_list = [self.d.screenshot()]
        self.loggerPdd.info(f"  截取第 1 张，尺寸: {screen_list[0].width}x{screen_list[0].height}")

        # 首屏已有「进店」→ 直接返回，不用滚
        if self.d(textContains='进店').exists or self.d(textStartsWith='进店').exists:
            self.loggerPdd.info(f"✅ 「进店」已在首屏，不滚动")
            return self._merge_screenshots(screen_list)

        for i in range(scroll_times):
            self.d.swipe(w // 2, int(h * 0.85), w // 2, int(h * 0.15),
                         duration=random.uniform(0.8, 1.5))
            time.sleep(random.uniform(2.0, 4.0))

            screen_list.append(self.d.screenshot())
            self.loggerPdd.info(f"  截取第 {i + 2} 张")

            if self.d(textContains='进店').exists or self.d(textStartsWith='进店').exists:
                self.loggerPdd.info(f"✅ 已到达「进店」位置，停止滚动")
                break

        return self._merge_screenshots(screen_list)

    # 代码运行那时候的时间
    def app_current_time(self):
        # 功能：返回当前时刻的格式化时间字符串，主要用于日志打印。
        return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    def slide_link(self):
        # 功能：在分享面板中横向滑动，把"复制链接"附近的目标入口滑到可见区域。
        value_tag = None
        if self.d.xpath('//*[@text="微信"]').exists:
            value_tag = self.d.xpath('//*[@text="微信"]').info['bounds']
            start_x = value_tag['right'] + 20
            end_x = max(start_x - 280, 30)
            self.d.swipe(start_x, value_tag['top'], end_x, value_tag['top'], 0.3)
            return
        if self.d.xpath('//*[@text="朋友圈"]').exists:
            value_tag = self.d.xpath('//*[@text="朋友圈"]').info['bounds']
            start_x = value_tag['right'] + 20
            end_x = max(start_x - 280, 30)
            self.d.swipe(start_x, value_tag['top'], end_x, value_tag['top'], 0.3)
            return
        if self.d.xpath('//*[@text="QQ好友"]').exists:
            value_tag = self.d.xpath('//*[@text="QQ好友"]').info['bounds']
            start_x = value_tag['right'] + 20
            end_x = max(start_x - 280, 30)
            self.d.swipe(start_x, value_tag['top'], end_x, value_tag['top'], 0.3)
            return

    def app_start_time(self):
        """
        获取app启动时间
        :return:
        """
        return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    def stop_app(self):
        # 功能：停止拼多多 App，并等待设备状态稳定下来。
        self.d.app_stop(self.package_name)
        time.sleep(5)

    def start_app(self):
        # 功能：启动拼多多 App，并预留加载时间。
        self.d.app_start(self.package_name)
        time.sleep(5)

    def restart_app(self):
        """
        重启app
        :return:
        """
        self.stop_app()
        # 这里先调用 stop_app()，是为了清掉上一次运行残留的页面状态；
        # 后续 start_app() 依赖应用已经被完全拉起，搜索流程才能从稳定起点开始。
        self.start_app()

    def is_pdd_package_foreground(self):
        """只按包名判断是否在拼多多前台，不要求页面锚点可见。"""
        try:
            current_app = self.d.app_current() or {}
            current_package = current_app.get("package", "")
            return current_package == self.package_name
        except Exception:
            return False

    def is_pdd_home_page(self):
        """判断是否在拼多多主页面（底部主导航可见）。"""
        try:
            home_anchors = [
                '//*[@text="首页"]',
                '//*[@text="分类"]',
                '//*[@text="多多视频"]',
                '//*[@text="聊天"]',
                '//*[@text="我的"]',
                '//*[@text="个人中心"]',
            ]
            hit_count = 0
            for anchor_xpath in home_anchors:
                if self.d.xpath(anchor_xpath).exists:
                    hit_count += 1
            return hit_count >= 3
        except Exception:
            return False

    def back_to_pdd_home_page(self, max_back_times=2):
        """
        尝试通过返回键回到拼多多主页面。
        按你的要求最多执行两次返回。
        """
        if not self.is_pdd_package_foreground():
            return False

        for _ in range(max_back_times):
            if self.is_pdd_home_page():
                break
            self.d.press('back')
            time.sleep(self.get_sleep_time())

        return self.is_pdd_home_page()

    def ensure_home_before_task_end(self, max_rounds=6):
        """
        任务结束前做收尾：持续回退直到回到拼多多主页面。
        每轮最多回退两次，总轮数可配置，避免无限循环。
        """
        try:
            if not self.is_pdd_package_foreground():
                logging.info("ensure_home_before_task_end: 拼多多不在前台，先拉起APP")
                self.start_app()
                time.sleep(self.get_sleep_time())
            #有bug，回到拼多多主页后不会继续执行
            for round_idx in range(max_rounds):
                if self.back_to_pdd_home_page(max_back_times=2):
                    logging.info(
                        f"ensure_home_before_task_end: 已回到拼多多主页面（轮次 {round_idx + 1}/{max_rounds}）"
                    )
                    return True
                logging.info(
                    f"ensure_home_before_task_end: 第 {round_idx + 1}/{max_rounds} 轮未到主页面，继续回退"
                )

            home_ready = self.is_pdd_home_page()
            logging.info(f"ensure_home_before_task_end: 收尾结束，主页面状态={home_ready}")
            return home_ready
        except Exception as e:
            logging.info(f"ensure_home_before_task_end: 执行异常: {e}")
            return False

    def focus_search_box(self, max_attempts=6):
        """把搜索框点出来，为 enter_target_page 提供稳定起点。"""
        search_entry_candidates = [
            '//*[@resource-id="android:id/content"]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]'
        ]

        for attempt in range(max_attempts):
            if self.d(className='android.widget.EditText').wait(timeout=1):
                return True

            clicked = False
            for entry_xpath in search_entry_candidates:
                if self.d.xpath(entry_xpath).exists:
                    self.d.xpath(entry_xpath).click()
                    clicked = True
                    time.sleep(self.get_sleep_time())
                    if self.d(className='android.widget.EditText').wait(timeout=1):
                        return True

            if not clicked:
                # 兜底点顶部搜索区域，兼容节点层级变化。
                try:
                    screen_width = self.d.info.get('displayWidth', 1080)
                    screen_height = self.d.info.get('displayHeight', 2400)
                    self.d.click(screen_width // 2, int(screen_height * 0.08))
                    time.sleep(self.get_sleep_time())
                    if self.d(className='android.widget.EditText').wait(timeout=1):
                        return True
                except Exception as click_error:
                    logging.info(f"focus_search_box: 顶部点击失败: {click_error}")

            if not self.is_pdd_package_foreground():
                logging.info("focus_search_box: 检测到拼多多不在前台，重新拉起APP")
                self.start_app()
                time.sleep(self.get_sleep_time())

            logging.info(f"focus_search_box: 第 {attempt + 1}/{max_attempts} 次未定位到搜索框")

        return False

    def focus_search_box_after_restart(self, max_attempts=6):
        """兼容旧调用，统一复用 focus_search_box。"""
        return self.focus_search_box(max_attempts=max_attempts)

    def prepare_entry_before_enter_target(self, force_restart=False):
        """
        统一入口准备：
        1) 按需重启；
        2) 回退最多两次到主页面；
        3) 点击出搜索框；
        4) 后续交给 enter_target_page 执行输入和搜索。
        """
        if force_restart:
            self.restart_app()
            time.sleep(self.get_sleep_time())
        elif not self.is_pdd_package_foreground():
            logging.info("prepare_entry: 当前拼多多不在前台，改为先重启")
            self.restart_app()
            time.sleep(self.get_sleep_time())

        home_ready = self.back_to_pdd_home_page(max_back_times=2)
        logging.info(f"prepare_entry: 回到主页面结果={home_ready}")

        search_ready = self.focus_search_box(max_attempts=6)
        logging.info(f"prepare_entry: 搜索框定位结果={search_ready}")
        return search_ready

    def is_pdd_foreground(self):
        """仅当"前台包名是拼多多 + 页面可见"时返回 True。"""
        try:
            current_app = self.d.app_current() or {}
            current_package = current_app.get("package", "")
            current_activity = current_app.get("activity", "")
            device_info = self.d.info or {}
            screen_on = bool(device_info.get("screenOn", True))

            print(
                f"foreground check -> package={current_package}, "
                f"activity={current_activity}, screen_on={screen_on}"
            )

            if not screen_on:
                return False

            if current_package != self.package_name:
                return False

            # 仅包名命中仍不够：要求当前页面能看到拼多多关键锚点，避免"后台残留包名"误判。
            visible_anchor_xpaths = [
                '//*[contains(@text, "搜索")]',
                '//*[@text="首页"]',
                '//*[@text="多多视频"]',
                '//*[@text="聊天"]',
                '//*[@text="我的"]',
                '//*[@text="个人中心"]',
                '//*[@text="进店"]',
                '//*[starts-with(@text, "商品参数")]',
            ]
            for anchor_xpath in visible_anchor_xpaths:
                if self.d.xpath(anchor_xpath).exists:
                    return True

            # 商品详情页有时锚点不稳定，放一个输入框兜底判断。
            if self.d(className='android.widget.EditText').exists:
                return True

            print("package 命中拼多多，但页面锚点不可见，按后台处理并触发重启")
            return False
        except Exception as e:
            print(f"failed to read foreground package/page state: {e}")
            return False

    def prepare_search_in_current_app(self):
        """Reuse current PDD page: swipe first, then clear and search without app restart."""
        print("pdd is in foreground, reuse app without restart")

        # Strong swipe-up first, as requested.
        for _ in range(2):
            self.d.swipe_ext("up", scale=0.45)
            time.sleep(0.8)

        search_entry_candidates = [
            '//*[contains(@text, "\u641c\u7d22")]',
            '//*[contains(@content-desc, "\u641c\u7d22")]',
            '//*[@resource-id="android:id/content"]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]'
            '/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]'
            '/android.widget.LinearLayout[1]',
        ]

        # Try to get search input directly; if not found, click possible search entries.
        if not self.d(className='android.widget.EditText').wait(timeout=2):
            for entry_xpath in search_entry_candidates:
                if self.d.xpath(entry_xpath).exists:
                    self.d.xpath(entry_xpath).click()
                    time.sleep(self.get_sleep_time())
                    if self.d(className='android.widget.EditText').wait(timeout=2):
                        break

        # Some list pages need one back press before search input appears.
        if not self.d(className='android.widget.EditText').wait(timeout=2):
            self.d.press("back")
            time.sleep(self.get_sleep_time())

        # Last fallback: use existing enter_target_page flow.
        if not self.d(className='android.widget.EditText').wait(timeout=2):
            print("search input not found after swipe, fallback to enter_target_page")
            self.enter_target_page()
            return

        self.d(className='android.widget.EditText').click()
        time.sleep(self.get_sleep_time())

        clear_xpath_candidates = [
            '//*[@text="\u6e05\u9664"]',
            '//*[@text="\u6e05\u7a7a"]',
            '//*[@content-desc="\u6e05\u9664"]',
            '//*[@content-desc="\u6e05\u7a7a"]',
        ]
        for clear_xpath in clear_xpath_candidates:
            if self.d.xpath(clear_xpath).exists:
                self.d.xpath(clear_xpath).click()
                time.sleep(0.5)
                print("clicked clear button")
                break

        self.d.send_keys(self.search_key, clear=True)
        time.sleep(self.get_sleep_time())

        if self.d.xpath('//*[@text="\u641c\u7d22"]').exists:
            self.d.xpath('//*[@text="\u641c\u7d22"]').click()
        else:
            self.d.press("enter")
        time.sleep(self.get_sleep_time())

    @staticmethod
    def get_sleep_time():
        # 功能：生成短随机等待时间，减少固定节奏操作带来的页面未加载或风控风险。
        return random.uniform(0.5, 1.0)
        # return random.randint(5, 8)

    def get_screen_size(self):
        """返回 (width, height) 屏幕尺寸，首次调用后缓存，兼容不同分辨率设备"""
        if not hasattr(self, '_cached_screen_w'):
            info = self.d.info
            self._cached_screen_w = info.get('displayWidth', 720)
            self._cached_screen_h = info.get('displayHeight', 1640)
        return self._cached_screen_w, self._cached_screen_h

    @staticmethod
    def get_current_date():
        # 功能：返回当前采集日期，作为去重和落库字段使用。
        return datetime.datetime.now().strftime('%Y/%m/%d')

    @staticmethod
    def _normalize_rule_list(value):
        # 功能：把单值或集合统一归一成非空字符串列表，供过滤逻辑直接复用。
        if value is None:
            return []
        if isinstance(value, (list, tuple, set)):
            raw_values = value
        else:
            raw_values = [value]
        result = []
        for item in raw_values:
            item_str = str(item).strip()
            if item_str:
                result.append(item_str)
        return result

    @staticmethod
    def _normalize_match_text(value):
        # 功能：把待匹配文本做去空白和小写归一，减少页面文案格式差异带来的误判。
        return re.sub(r'\s+', '', str(value or '')).lower()

    @staticmethod
    def _normalize_region_name(name, level):
        text = re.sub(r"\s+", "", str(name or ""))
        text = text.replace("　", "")
        if not text:
            return ""

        if level == "province":
            province_alias = {
                "内蒙古自治区": "内蒙古",
                "广西壮族自治区": "广西",
                "宁夏回族自治区": "宁夏",
                "新疆维吾尔自治区": "新疆",
                "西藏自治区": "西藏",
                "香港特别行政区": "香港",
                "澳门特别行政区": "澳门",
            }
            if text in province_alias:
                return province_alias[text]

            for suffix in ("特别行政区", "维吾尔自治区", "回族自治区", "壮族自治区", "自治区", "省", "市"):
                if text.endswith(suffix) and len(text) > len(suffix):
                    return text[:-len(suffix)]
            return text

        city_alias = {
            "北京市": "北京",
            "天津市": "天津",
            "上海市": "上海",
            "重庆市": "重庆",
        }
        if text in city_alias:
            return city_alias[text]

        for suffix in ("自治州", "地区", "盟", "州", "市"):
            if text.endswith(suffix) and len(text) > len(suffix):
                return text[:-len(suffix)]
        return text

    def _load_city_lookup(self):
        if self._city_lookup_cache is not None:
            return self._city_lookup_cache

        base_dir = os.path.dirname(os.path.abspath(__file__))
        candidate_paths = [
            os.path.join(base_dir, "city.json"),
            os.path.join(base_dir, "process_shop", "city.json"),
        ]

        city_data = []
        loaded_path = ""
        for path in candidate_paths:
            if not os.path.exists(path):
                continue
            try:
                with open(path, "r", encoding="utf-8") as f:
                    candidate_data = json.load(f)
                if isinstance(candidate_data, list):
                    city_data = candidate_data
                    loaded_path = path
                    break
            except Exception as exc:
                logging.warning(f"加载 city.json 失败: {path}, error={exc}")

        province_to_id = {}
        province_city_to_id = {}
        global_city_matches = {}

        for province in city_data:
            province_norm = self._normalize_region_name(province.get("name", ""), "province")
            province_id = parse_optional_int(province.get("id"), 0) or 0

            if province_norm and province_id and province_norm not in province_to_id:
                province_to_id[province_norm] = province_id

            city_map = {}
            for city in province.get("sons", []) or []:
                city_norm = self._normalize_region_name(city.get("name", ""), "city")
                city_id = parse_optional_int(city.get("id"), 0) or 0
                if city_norm and city_id and city_norm not in city_map:
                    city_map[city_norm] = city_id
                if city_norm and city_id and province_id:
                    global_city_matches.setdefault(city_norm, []).append((province_id, city_id, province_norm))

            if province_norm and province_norm not in province_city_to_id:
                province_city_to_id[province_norm] = city_map

        if loaded_path:
            logging.info(f"已加载 city.json: {loaded_path}")
        else:
            logging.warning("未找到可用 city.json，省市ID将回填为 0")

        self._city_lookup_path = loaded_path
        self._city_lookup_cache = {
            "province_to_id": province_to_id,
            "province_city_to_id": province_city_to_id,
            "global_city_matches": global_city_matches,
        }
        return self._city_lookup_cache

    def _resolve_region_ids(self, province_name, city_name):
        lookup = self._load_city_lookup()
        province_to_id = lookup["province_to_id"]
        province_city_to_id = lookup["province_city_to_id"]
        global_city_matches = lookup["global_city_matches"]

        province_norm = self._normalize_region_name(province_name, "province")
        city_norm = self._normalize_region_name(city_name, "city")

        province_id = 0
        city_id = 0

        if province_norm:
            province_id = province_to_id.get(province_norm, 0)
            if not province_id:
                for key, value in province_to_id.items():
                    if province_norm in key or key in province_norm:
                        province_id = value
                        province_norm = key
                        break

        if province_id and city_norm:
            city_map = province_city_to_id.get(province_norm, {})
            city_id = city_map.get(city_norm, 0)
            if not city_id:
                for key, value in city_map.items():
                    if city_norm in key or key in city_norm:
                        city_id = value
                        break

        if city_norm and (not province_id or not city_id):
            matches = global_city_matches.get(city_norm, [])
            if len(matches) == 1:
                matched_province_id, matched_city_id, _ = matches[0]
                province_id = province_id or matched_province_id
                city_id = city_id or matched_city_id
            elif province_id and not city_id:
                for matched_province_id, matched_city_id, _ in matches:
                    if matched_province_id == province_id:
                        city_id = matched_city_id
                        break

        return province_id or 0, city_id or 0

    def _match_any_keyword(self, text, keywords):
        # 功能：判断目标文本是否命中任一过滤词；过滤词为空时直接放行。
        keyword_list = self._normalize_rule_list(keywords)
        if not keyword_list:
            # 没有配置过滤词时默认放行，让调用方只在需要时开启细筛。
            return True
        normalized_text = self._normalize_match_text(text)
        return any(self._normalize_match_text(keyword) in normalized_text for keyword in keyword_list)

    def _get_brand_match_keywords(self):
        brand_keywords = self._normalize_rule_list(self.brand)
        for brand in list(brand_keywords):
            brand_keywords.extend(self._normalize_rule_list(BRAND_ALIAS_MAP.get(brand, [])))
        return list(dict.fromkeys(brand_keywords))

    def is_link_spec_useful(self, product_title, specifications=''):
        # 功能：判断标题或说明书规格里是否包含目标品规。
        if not self.spec_list:
            return True

        title_text = self._normalize_match_text(product_title)
        spec_text = self._normalize_match_text(specifications)

        for spec in self.spec_list:
            normalized_spec = self._normalize_match_text(spec)
            if normalized_spec in title_text or normalized_spec in spec_text:
                return True
        return False

    def is_link_useful(self, product_title, specifications=''):
        # 功能：统一做标题、品牌、规格三层过滤，尽量在早期就排除无关商品。
        if not self._match_any_keyword(product_title, self.title_key):
            print(f"当前商品名称:{product_title} 不包含{self.title_key}关键字")
            return False
        brand_keywords = self._get_brand_match_keywords()
        if not self._match_any_keyword(product_title, brand_keywords):
            print(f"当前商品名称:{product_title} 不包含{brand_keywords}品牌")
            return False
        if not self.is_link_spec_useful(product_title, specifications):
            print(f"当前商品名称:{product_title} 不包含{self.spec_list}品规")
            return False
        return True

    def remove_watermark(self, img_path):
        # 功能：弱化截图中的水印或遮罩，提升后续 OCR 识别成功率。
        """
        图片去水印(将水印部分变成白色背景)并将数据转化为二进制数据
        :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_license_info(self):
        img_path = f"./shop_info_screenshot.png"  # 截图地址
        has_jindian = self.d.xpath('//*[@text="进店"]').exists

        if has_jindian:
            # 旧版：滚动直到"进店"接近顶部
            for i in range(15):
                if not self.d.xpath('//*[@text="进店"]').exists:
                    break
                shop_bounds = self.d.xpath('//*[@text="进店"]').info['bounds']
                element_top = shop_bounds['top']
                print(f"进店到顶部的距离：{element_top}")
                _, screen_h = self.get_screen_size()
                if element_top <= int(screen_h * 0.30):
                    self.loggerPdd.info(f"🔍 第{i + 1}次上滑，「进店」上滑到最顶部")
                    break
                self.d.swipe_ext("up", scale=0.2)
                time.sleep(self.get_sleep_time())
        else:
            # 新版：点击底部"店铺"tab进入店铺页
            self.loggerPdd.info('无进店按钮，通过底部店铺tab进入店铺页')
            shop_tab = self.d.xpath('//*[@content-desc="店铺"]')
            if not shop_tab.exists:
                shop_tab = self.d.xpath('//*[@text="店铺"]')
            if shop_tab.exists:
                shop_tab.click()
                time.sleep(2)

        license_info = {}
        # 新版 + 旧版：搜索资质证照图标（兼容 support.v7 和 androidx）
        shop_license_xpaths = [
            '(//android.support.v7.widget.RecyclerView[@resource-id="com.xunmeng.pinduoduo:id/pdd"]/android.widget.FrameLayout/android.view.ViewGroup/android.widget.ImageView[@resource-id="com.xunmeng.pinduoduo:id/pdd"])[2]',
            '(//androidx.recyclerview.widget.RecyclerView[@resource-id="com.xunmeng.pinduoduo:id/pdd"]/android.widget.FrameLayout/android.view.ViewGroup/android.widget.ImageView[@resource-id="com.xunmeng.pinduoduo:id/pdd"])[2]',
        ]
        found_license = False
        for xp in shop_license_xpaths:
            if self.d.xpath(xp).exists:
                print(f"存在资质图标: {xp[:60]}...")
                self.d.xpath(xp).click()
                found_license = True
                break

        if not found_license:
            # 兜底：搜索 contentDescription 含"资质"或"证照"的 ImageView
            for n in self.d.xpath('//android.widget.ImageView').all():
                desc = (n.info.get('contentDescription') or '').strip()
                if '资质' in desc or '证照' in desc:
                    n.click()
                    found_license = True
                    break

        if found_license:
            print(f"点击资质图标")
            time.sleep(3)
            self.d.screenshot(img_path)
            self.loggerPdd.info(f"截图资质成功：{img_path}")
            ocr_res = self.get_baidu_ocr_jgh_res(img_path)
            if ocr_res:
                for key in ocr_res:
                    if '企业名称' in key:
                        license_info['business_license_company'] = ocr_res[key]
                    if '经营地址' in key:
                        license_info['business_license_address'] = ocr_res[key]
                    if '注册地址' in key:
                        license_info['business_license_address'] = ocr_res[key]
                    if '信用代码' in key:
                        license_info['qualification_number'] = ocr_res[key]
                if license_info.get('business_license_address'):
                    license_info['province'], license_info['city'] = self.infer_region_from_address(license_info['business_license_address'])
            # 从资质页返回
            if self.d.xpath('//*[@content-desc="返回"]').exists:
                self.d.xpath('//*[@content-desc="返回"]').click()
            elif not has_jindian:
                # 新版：从店铺页返回详情页
                self.d.press('back')
        else:
            print(f"不存在shop资质")

        # 新版：从店铺页返回
        if not has_jindian and not found_license:
            self.d.press('back')
            time.sleep(self.get_sleep_time())
        return license_info

    @staticmethod
    def _norm_region_name(name):
        if name is None:
            return ""
        text = str(name).strip()
        if not text:
            return ""
        return re.sub(r'(省|市|自治区|特别行政区)$', '', text)

    def build_address_region_index(self, file_path):
        pairs = []
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            for province in data:
                p_name = str(province.get("name") or "").strip()
                if not p_name:
                    continue
                p_candidates = {p_name, self._norm_region_name(p_name)}
                cities = province.get("sons") or []
                if not cities:
                    pairs.append({
                        "province": p_name,
                        "city": "",
                        "province_candidates": p_candidates,
                        "city_candidates": set()
                    })
                    continue
                for city in cities:
                    c_name = str(city.get("name") or "").strip()
                    c_candidates = {c_name, self._norm_region_name(c_name)} if c_name else set()
                    pairs.append({
                        "province": p_name,
                        "city": c_name,
                        "province_candidates": p_candidates,
                        "city_candidates": c_candidates
                    })
        except Exception as e:
            print(f"构建地址省市索引失败: {e}")
        return pairs

    def infer_region_from_address(self, address):
        text = str(address or "").strip()
        if not text:
            return "", ""
        for item in self.address_region_index:
            if any(k and k in text for k in item["province_candidates"]):
                city = item["city"]
                if city and any(k and k in text for k in item["city_candidates"]):
                    return item["province"], city
        for item in self.address_region_index:
            if any(k and k in text for k in item["province_candidates"]):
                return item["province"], item["city"]
        return "", ""

    def get_shop_name(self):
        # 功能：优先从当前详情页直接提取店铺名，失败时再进入店铺页兜底提取。
        """
        获取店铺名
        兼容多版本：进店锚点 → 文案搜索 → 底部店铺tab → 滚回顶部重试
        :return:
        """
        try:
            shop_keywords = ['大药房', '旗舰店', '专卖店', '药房', '药店', '官方店', '专营店', '健康']
            xpath_v1 = '//*[@text="进店"]/preceding-sibling::android.widget.LinearLayout/android.widget.TextView'
            xpath_v2 = '//*[@text="进店"]/preceding-sibling::android.view.ViewGroup/android.widget.LinearLayout/android.widget.TextView'

            def _try_direct():
                """尝试直接从详情页读取店铺名"""
                if self.d.xpath(xpath_v1).exists:
                    return self.d.xpath(xpath_v1).text, 'v1进店锚点'
                if self.d.xpath(xpath_v2).exists:
                    return self.d.xpath(xpath_v2).text, 'v2进店锚点'
                for kw in shop_keywords:
                    for n in self.d.xpath(f'//android.widget.TextView[contains(@text, "{kw}")]').all():
                        txt = (n.info.get('text') or '').strip()
                        b = n.info.get('bounds', {})
                        if len(txt) < 25 and b.get('top', 0) > 300:
                            return txt, f'文案搜索({kw})'
                return None, None

            # 第一次尝试
            shop_name, method = _try_direct()
            if shop_name:
                self.loggerPdd.info(f'获取到店铺名({method}): {shop_name}')
                return shop_name

            # 滚回顶部再试（截图等操作可能把页面滚走了）
            self.loggerPdd.info('首轮未获取到店铺名，滚回顶部重试')
            for _ in range(3):
                self.d.swipe_ext("down", scale=0.5)
                time.sleep(0.5)
            shop_name, method = _try_direct()
            if shop_name:
                self.loggerPdd.info(f'获取到店铺名(滚回后-{method}): {shop_name}')
                return shop_name

            # 方式3：点击底部"店铺"tab进入店铺页
            shop_tab = self.d.xpath('//*[@content-desc="店铺"]')
            if not shop_tab.exists:
                shop_tab = self.d.xpath('//*[@text="店铺"]')
            if shop_tab.exists:
                shop_tab.click()
                time.sleep(2)

                for kw in shop_keywords:
                    for n in self.d.xpath(f'//android.widget.TextView[contains(@text, "{kw}")]').all():
                        txt = (n.info.get('text') or '').strip()
                        b = n.info.get('bounds', {})
                        if len(txt) < 25 and b.get('top', 0) < 600:
                            self.loggerPdd.info(f'获取到店铺名(店铺页): {txt}')
                            self.swipe_back(1)
                            return txt

                self.swipe_back(1)

            self.loggerPdd.info('所有方式均未获取到店铺名')
            return ''
        except Exception as e:
            print(f'获取店铺名出错:{e}')
            self.loggerPdd.error(f'获取店铺名出错:{e}')
            return None

    def save_to_shop_database(self, data):
        # 功能：把当前商品采集结果落库；只有 commit 成功后才计入采集数量。
        print(f'保存店铺数据到店铺数据库:{data}')
        shop = str(data.get('shop') or '').strip()
        if not shop:
            print("保存店铺数据失败：shop 为空")
            return False

        # 数据库部分字段为 NOT NULL，统一做非空兜底，避免写入 None 触发 1048。
        contact_address = str(data.get('contact_address') or '').strip()
        qualification_number = str(data.get('qualification_number') or '').strip()
        business_license_company = str(data.get('business_license_company') or '').strip()
        business_license_address = str(data.get('business_license_address') or '').strip()
        shop_url = str(data.get('store_url') or '').strip()
        scrape_date = str(data.get('scrape_date') or self.get_current_date()).strip()
        platform = str(data.get('platform') or '3').strip()
        province = str(data.get('province_name') or data.get('province') or '').strip()
        city = str(data.get('city_name') or data.get('city') or '').strip()
        create_time = data.get('create_time') or datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        update_time = data.get('update_time') or datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        max_retries = 5
        # 数据库偶发抖动时允许短重试，但只有 commit 成功后才算真正采集到一条数据。
        for attempt in range(max_retries):
            conn = None
            try:
                conn = get_mysql()
                with conn.cursor() as cur:
                    add_sql = """
                                INSERT INTO retrieve_scrape_shop_info (
                                    shop, contact_address, qualification_number, business_license_company,
                                    business_license_address, shop_url, scrape_date, platform,
                                    province,city, create_time, update_time
                                ) VALUES (
                                    %s, %s, %s, %s, %s,
                                    %s, %s, %s, %s, %s,
                                    %s, %s
                                )
                            """
                    cur.execute(add_sql, (
                        shop,
                        contact_address,
                        qualification_number,
                        business_license_company,
                        business_license_address,
                        shop_url,
                        scrape_date,
                        platform,
                        province,
                        city,
                        create_time,
                        update_time,
                    ))
                conn.commit()

                print(f"存入数据库成功")
                return True
            except Exception as e:
                print(f'保存数据库异常 (尝试 {attempt + 1}/{max_retries}): {e}')
                if conn:
                    conn.rollback()
                    conn.close()
                if attempt == max_retries - 1:
                    print("达到最大重试次数，保存失败")
                    return False
                time.sleep(2)




    def save_to_database(self, data):
        # 功能：把当前商品采集结果落库；只有 commit 成功后才计入采集数量。
        print(f'保存数据到数据库:{data}')
        max_retries = 5
        # 数据库偶发抖动时允许短重试，但只有 commit 成功后才算真正采集到一条数据。
        for attempt in range(max_retries):
            conn = None
            try:
                conn = get_mysql()
                with conn.cursor() as cur:
                    add_sql = """
                        INSERT INTO retrieve_scrape_data (
                            enterprise_id, platform_id, platform_item_id, province_id, city_id,
                            province_name, city_name, area_info, product_brand, product_name, product_specs, search_name,
                            collect_config_info, 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,collect_equipment_account_id,insert_time,update_time,collect_round,collect_region_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
                        )
                    """
                    cur.execute(add_sql, (
                        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_brand'],
                        data['product_name'],
                        data['product_specs'],
                        data['search_name'],
                        data['collect_config_info'],
                        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'],
                        data['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.get('collect_equipment_account_id',0),
                        data['insert_time'],
                        data['update_time'],
                        data['collect_round'],
                        data['collect_region_id'],
                    ))
                conn.commit()
                self.max_counts += 1
                print(f"存入数据库成功，当前已采集 {self.max_counts} 条")

                return True
            except Exception as e:
                print(f'保存数据库异常 (尝试 {attempt + 1}/{max_retries}): {e}')
                if conn:
                    conn.rollback()
                    conn.close()
                if attempt == max_retries - 1:
                    print("达到最大重试次数，保存失败")
                    return False
                time.sleep(2)

    def click_target_product_by_search_key(self, fuzzy_match=False, timeout=10):
        # 功能：在列表页重新定位当前搜索词对应的商品，常用于异常恢复后的重新对焦。
        """
        动态匹配self.search_key对应的商品并点击
        :param fuzzy_match: 是否模糊匹配（应对商品名带额外后缀/前缀的情况） 不模糊匹配
        :param timeout: 等待元素出现的超时时间（秒）
        :return: 点击是否成功（bool）
        """
        try:
            # 1. 定义定位条件（动态使用self.search_key）
            # 异常恢复后需要重新找到"当前任务真正想点的那一个商品"，
            # 这里支持精确和模糊两种定位策略。
            if fuzzy_match:
                # 模糊匹配：包含search_key即可（推荐，适配搜索结果商品名略有差异）
                locator = self.d(textContains=self.search_key)
                print(f"🔍 模糊匹配商品：包含「{self.search_key}」的元素")
            else:
                # 精确匹配：商品名与search_key完全一致
                locator = self.d(text=self.search_key)
                print(f"🔍 精确匹配商品：「{self.search_key}」")

            # 2. 等待元素出现（核心：避免元素未加载就点击）
            if locator.wait(timeout=timeout):
                print(f"✅ 找到匹配的商品，准备点击")
                # 执行点击（优先点击可点击的元素）
                locator.click()
                print(f"✅ 成功点击「{self.search_key}」对应的商品")
                # 点击后等待页面加载
                time.sleep(self.get_sleep_time())
                return True
            else:

                print(f"❌ 滑动后仍未找到「{self.search_key}」对应的商品")
                return False
        except Exception as e:
            print(f"❌ 点击「{self.search_key}」对应商品时异常：{e}")
            return False

    def swipe_down(self):
        # 功能：执行带随机性的向下滑动，兼顾页面恢复、回找搜索框和设备适配。
        """
        下滑（模拟真人操作，抗风控+设备适配+容错）
        核心：起点在屏幕上方，终点在屏幕下方（和上滑相反）
        :return: None
        """
        try:
            # 1. 获取屏幕尺寸（兼容不同设备，给默认值避免获取失败）
            screen_width = self.d.info.get('displayWidth', 1080)  # 默认1080px宽度
            screen_height = self.d.info.get('displayHeight', 2400)  # 默认2400px高度

            # 2. 随机滑动时长（0.1~0.3秒，避免固定值被风控，且不设0秒）
            duration_rate = random.uniform(0.1, 0.3)

            # 3. 计算滑动坐标（用屏幕比例，适配所有设备）
            start_x = screen_width // 2  # 水平居中（和上滑一致，符合真人操作习惯）
            start_y = screen_height * 0.2  # 起点：屏幕20%高度（上方偏下）
            end_y = screen_height * 0.8  # 终点：屏幕80%高度（下方偏上）
            # 强制确保起点y < 终点y（必为向下滑，避免逻辑错误）
            start_y, end_y = min(start_y, end_y - 10), max(end_y, start_y + 10)

            # 4. 核心向下滑动操作
            self.d.swipe(start_x, start_y, start_x, end_y, duration=duration_rate)
            # 滑动后全局等待（确保页面加载，避免元素定位失败）
            time.sleep(self.get_sleep_time())


        except Exception as e:
            # 异常捕获：避免设备断开/滑动失败导致程序崩溃
            print(f"向下滑动失败：{e}")
            # 兜底方案：用屏幕比例坐标重试
            sw, sh = self.get_screen_size()
            self.d.swipe(sw // 2, int(sh * 0.2), sw // 2, int(sh * 0.8), duration=0.2)
            time.sleep(self.get_sleep_time())

    def swipe_up(self):
        # 功能：执行向上滑动，用于翻页或继续向下浏览详情。
        """
        上滑
        :return:
        """
        screen_width = self.d.info['displayWidth']
        screen_height = self.d.info['displayHeight']
        duration_rate = random.uniform(0, 0.3)
        self.d.swipe(screen_width // 2, screen_height - 100, screen_width // 2, 100, duration=duration_rate)
        no = random.uniform(0, 1)
        if no > 0.85:
            # 有的时候卡着 再稍微往上滑一点点
            self.d.swipe_ext("up", 0.1)
            time.sleep(self.get_sleep_time())

    def swipe_back(self, no):
        # 功能：按指定次数执行返回，但只有当前不在列表页时才真正后退。
        """
        返回
        :param no: 回退次数
        :return:
        """
        for idx in range(no):
            if self.distinct_target():  # 已经在列表页
                print(f'已在列表页，停止返回')
                return True
            self.d.press('back')
            time.sleep(self.get_sleep_time())

    def drug_price(self):
        # 功能：直接从详情页读取价格，作为规格弹窗取价失败时的兜底方案。
        """
        获取药品价格
        :return:
        """
        try:
            xpath = '//*[@text="¥"]/following-sibling::android.widget.TextView[1]'
            price_str = self.d.xpath(xpath).text
            price = float(re.search(r'[\d\.]+', price_str).group())
            print(f'获取到价格:{price}')
            return float(price)
        except Exception as e:
            print(f'提取价格出错-->{e}')
            return None

    def drug_price_ex(self):
        # 功能：优先从规格选择弹窗里同时提取价格和"已选规格"文本。

        price_str = ''  # 价格初始化
        ext = ''  # 初始化已选择的信息
        price = ''

        # 阶段 1：先尝试打开规格/品规弹窗，因为后续价格和规格文本都依赖这个弹窗内容。
        # 这是点击进入品规的按钮
        button_xpath_1 = '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.LinearLayout[1]/android.widget.LinearLayout[2]/android.widget.LinearLayout[1]/android.widget.LinearLayout[1]/android.view.ViewGroup[last()]'

        button_xpath_2 = '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.LinearLayout[1]/android.widget.LinearLayout[2]/android.widget.LinearLayout[1]/android.widget.LinearLayout[1]/android.widget.LinearLayout[1]/android.view.ViewGroup[last()]'

        button_xpath_3 = '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.LinearLayout/android.widget.LinearLayout/android.widget.LinearLayout[2]/android.widget.LinearLayout/android.view.ViewGroup'

        # 调试
        # test_button = self.d.xpath(button_xpath_1).exists
        # print(test_button)

        # test_button_2 = self.d.xpath(button_xpath_2).exists
        # print(test_button_2)
        # time.sleep(1000)
        # if self.d.xpath('//*[@text="发起拼单"]').exists:
        #     self.d.xpath('//*[@text="发起拼单"]').click()
        # elif self.d.xpath('//*[@text="去复诊开药"]').exists:
        #     self.d.xpath('//*[@text="去复诊开药"]').click()

        if self.d.xpath(button_xpath_1).exists:
            self.d.xpath(button_xpath_1).click()
        elif self.d.xpath(button_xpath_2).exists:
            self.d.xpath(button_xpath_2).click()
        elif self.d.xpath(button_xpath_3).exists:
            self.d.xpath(button_xpath_3).click()
        else:
            print("button1 and button_2 and button_3 all not exist")
            return price, ext

        # 阶段 2：根据不同弹窗布局选择对应的 XPath 解析策略。
        select_xpath_1 = '//*[@resource-id="android:id/content"]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.TextView[last()]'
        select_xpath_2 = '//*[@resource-id="android:id/content"]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.TextView[last()]'
        select_xpath_3 = '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.view.ViewGroup[2]/android.widget.LinearLayout[1]/android.view.ViewGroup[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.view.ViewGroup[1]/android.widget.TextView[last()]'
        select_xpath_3_2 = '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.view.ViewGroup[2]/android.widget.LinearLayout[1]/android.view.ViewGroup[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.view.ViewGroup[1]/android.widget.TextView[last()-1]'

        price_xpath_1 = '//*[@resource-id="android:id/content"]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.TextView[1]'
        price_xpath_2 = '//*[@resource-id="android:id/content"]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.TextView[1]'
        price_xpath_3 = '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.view.ViewGroup[2]/android.widget.LinearLayout[1]/android.view.ViewGroup[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.view.ViewGroup[1]//android.widget.TextView[1]'

        # 新版规格选项路径（androidx RecyclerView，兼容新版拼多多弹窗）
        scroll_xpath_2 = '//*[@resource-id="android:id/content"]//android.widget.ScrollView[1]/android.widget.LinearLayout/androidx.recyclerview.widget.RecyclerView/android.widget.LinearLayout/android.view.ViewGroup[1]'
        recycler_view_xpath_2 = '//*[@resource-id="android:id/content"]//androidx.recyclerview.widget.RecyclerView[1]/android.widget.LinearLayout/android.view.ViewGroup[1]'

        if self.d.xpath(select_xpath_1).exists:
            text1 = self.d.xpath(select_xpath_1).text

            print(f"select_xpath_1--text1={text1}")

            # 这里先判断是否已经有默认规格，是为了减少额外点击；
            # 如果已经存在"已选"文本，后续可以直接读取价格和规格。
            if '已选' in text1:
                if self.d.xpath(price_xpath_1).exists:
                    price_str = self.d.xpath(price_xpath_1).text
                    print(f"select_xpath_1--price_str-1={price_str}")
                else:
                    print("select_xpath_1--price_xpath_1-1 not exist")

                ext = text1
            elif '请选择' in text1:
                # 调用 click() 的目的是补齐一次规格选择动作，
                # 调用后价格文本和"已选规格"文案才会稳定刷新出来。
                # 需要再下面点击选择
                scroll_xpath_1 = '//*[@resource-id="android:id/content"]//android.widget.ScrollView[1]/android.widget.LinearLayout[1]/android.support.v7.widget.RecyclerView[1]/android.widget.LinearLayout[last()]/android.view.ViewGroup[1]/android.view.ViewGroup[last()]'
                scroll_xpath_2 = ''
                if self.d.xpath(scroll_xpath_1).exists:
                    self.d.xpath(scroll_xpath_1).click()
                    time.sleep(2)  # 延时2秒钟，选择了之后价格会刷新
                    if self.d.xpath(select_xpath_1).exists:
                        text2 = self.d.xpath(select_xpath_1).text
                        if '已选' in text2:
                            print(f"select_xpath_1--已选择2:text2={text2}")
                            if self.d.xpath(price_xpath_1).exists:
                                price_str = self.d.xpath(price_xpath_1).text
                                print(f"select_xpath_1--price_str-2={price_str}")
                            else:
                                print("select_xpath_1--price_xpath_1-2 not exist")

                            ext = text2
                else:
                    print("select_xpath_1--scroll_xpath_1 not exist")
        elif self.d.xpath(select_xpath_2).exists:
            text1 = self.d.xpath(select_xpath_2).text
            print(f"xpath2--text1={text1}")
            if '已选' in text1:
                ext = text1
                if self.d.xpath(price_xpath_2).exists:
                    price_str = self.d.xpath(price_xpath_2).text
                    print(f"select_xpath_2--price_str-2={price_str}")
                else:
                    print("select_xpath_2--price_xpath_2-1 not exist")
            elif '请选择' in text1:
                # 当前布局下如果不先选择一个规格，后续既拿不到准确价格，也无法计算盒数。
                print('come in here')
                # 需要再下面点击选择
                scroll_xpath_1 = '//*[@resource-id="android:id/content"]//android.widget.ScrollView[1]/android.widget.LinearLayout[1]/android.support.v7.widget.RecyclerView[1]/android.widget.LinearLayout[last()]/android.view.ViewGroup[1]/android.view.ViewGroup[1]'

                hit_scroll = False
                if self.d.xpath(scroll_xpath_1).exists:
                    print("scroll_xpath_1 exists")
                    self.d.xpath(scroll_xpath_1).click()
                    hit_scroll = True
                elif self.d.xpath(scroll_xpath_2).exists:
                    print("scroll_xpath_2 exists")
                    self.d.xpath(scroll_xpath_2).click()
                    hit_scroll = True

                if hit_scroll:
                    time.sleep(2)  # 延时2秒钟，选择了之后价格可能会刷新
                    if self.d.xpath(select_xpath_2).exists:
                        text2 = self.d.xpath(select_xpath_2).text
                        if '已选' in text2:
                            ext = text2
                            print(f"select_xpath_2--已选择2:text2={text2}")
                            if self.d.xpath(price_xpath_2).exists:
                                price_str = self.d.xpath(price_xpath_2).text
                                print(f"select_xpath_2--price_str-2={price_str}")
                            else:
                                print("select_xpath_2--price_xpath_2-2 not exist")
                else:
                    print("scroll_xpath_1 and scroll_xpath_2 not exists")
            else:
                print("not exist 请选择 or 已选")
        elif self.d.xpath(select_xpath_3).exists:
            text1 = self.d.xpath(select_xpath_3).text
            print(f"xpath3--text1-1={text1}")

            if ('请选择' not in text1) and ('已选' not in text1):
                text1 = self.d.xpath(select_xpath_3_2).text
                print(f"xpath3--text1-2={text1}")

            if '已选' in text1:
                ext = text1
                if self.d.xpath(price_xpath_3).exists:
                    price_str = self.d.xpath(price_xpath_3).text
                    print(f"select_xpath_3--price_str-3-3-1={price_str}")
                else:
                    print("select_xpath_3--price_xpath_3-3-1 not exist")
            elif '请选择' in text1:
                # 这一支兼容另一类规格弹窗结构，核心目标仍然是先拿到"已选"文本。
                print('come in here')
                # 需要再下面点击选择
                scroll_xpath_1 = '//*[@resource-id="android:id/content"]//android.widget.ScrollView[1]/android.widget.LinearLayout[1]/android.support.v7.widget.RecyclerView[1]/android.widget.LinearLayout[last()]/android.view.ViewGroup[1]/android.view.ViewGroup[1]'
                recycler_view_xpath = '//*[@resource-id="android:id/content"]//android.support.v7.widget.RecyclerView[1]/android.widget.LinearLayout[1]/android.widget.LinearLayout[last()]/android.view.ViewGroup[1]/android.view.ViewGroup[1]'

                hit_scroll = False
                if self.d.xpath(scroll_xpath_1).exists:
                    print("scroll_xpath_1 exists")
                    self.d.xpath(scroll_xpath_1).click()
                    hit_scroll = True
                elif self.d.xpath(recycler_view_xpath).exists:
                    self.d.xpath(recycler_view_xpath).click()
                    hit_scroll = True
                elif self.d.xpath(scroll_xpath_2).exists:
                    print("scroll_xpath_2 exists")
                    self.d.xpath(scroll_xpath_2).click()
                    hit_scroll = True
                elif self.d.xpath(recycler_view_xpath_2).exists:
                    print("recycler_view_xpath_2 exists")
                    self.d.xpath(recycler_view_xpath_2).click()
                    hit_scroll = True

                if hit_scroll:
                    time.sleep(2)  # 延时2秒钟，选择了之后价格可能会刷新
                    if self.d.xpath(select_xpath_3).exists:
                        text2 = self.d.xpath(select_xpath_3).text
                        if '已选' in text2:
                            ext = text2
                            print(f"select_xpath_3--已选择2:text2={text2}")
                            if self.d.xpath(price_xpath_3).exists:
                                price_str = self.d.xpath(price_xpath_3).text
                                print(f"select_xpath_3--price_str-3-2={price_str}")
                            else:
                                print("select_xpath_3--price_xpath_3-3-2 not exist")

                else:
                    print("scroll_xpath_1, recycler_view_xpath, scroll_xpath_2, recycler_view_xpath_2 all not exists")
            else:
                print(f"xpath3--text1-不包含请选择和已选择")

        else:
            print("select_xpath_1 and select_xpath_2 and select_xpath_3 all not exist")

        # 阶段 3：从界面文案中抽取纯价格值，供后续去重和单盒价格计算。
        if price_str:
            # price = float(re.search('[\d\.]+', price_str).group())
            match = re.search(r'¥([\d\.]+)', price_str)
            if match:
                price = float(match.group(1))
            else:
                price = ''

                # price = float(re.search(r'¥([\d\.]+)', price_str).group(1))
            print(f'获取到价格:{price}')

        print(f"ext={ext}")

        # 调用 swipe_back() 的目的是把页面从规格弹窗恢复回商品详情页，
        # 后续提取店铺名、链接和说明书都依赖当前仍停留在详情页。
        self.swipe_back(1)  #

        return price, ext

    def restart_uiautomator_services(self, device_id):
        # 功能：重启设备上的 atx-agent/uiautomator 服务，恢复自动化控制能力。
        """
        重启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 connect_devices(self, device_id):
        # 功能：建立 USB 设备连接，并把自动化服务重置到可用状态。
        """
        连接设备
        :return:
        """
        try:
            self.d = u2.connect_usb(device_id)
            # 设置隐形等待时间
            # self.d.implicitly_wait(5)
            # 连上设备后主动重启 atx-agent，减少长时间运行后的控件失效问题。
            self.restart_uiautomator_services(device_id)
            print(f'[{self.program_start_time}]连接到设备:{device_id}')
        except Exception as e:
            print(f'{device_id} 连接错误: {e}')
            raise Exception(e)

    # 验证码 WebView XPath — 所有验证码相关页面都有这个元素
    CAPTCHA_WEBVIEW = '//android.webkit.WebView[@text="安全验证"]'

    def check_and_solve_captcha(self):
        """检测验证码 WebView 并调用 test_captcha.solve 解决，成功返回 True"""
        if not self.d.xpath(self.CAPTCHA_WEBVIEW).exists:
            return False
        logging.info("检测到验证码页面（WebView 安全验证），开始解决...")
        result = captcha_solve(self.d)
        if result:
            logging.info("验证码已解决，休息30分钟...")
            time.sleep(1800)
        return result

    def get_baidu_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=" + get_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 get_baidu_ocr_jgh_res(self, img):
        try:
            # img地址
            print(f'开始识别图片:{img}')
            request_url = "https://aip.baidubce.com/rest/2.0/ocr/v1/smart_struct"
            f = open(img, 'rb')
            img = base64.b64encode(f.read())
            params = {"image": img}
            request_url = request_url + "?access_token=" + get_access_token()
            headers = {'content-type': 'application/x-www-form-urlencoded'}
            response = requests.post(request_url, data=params, headers=headers)
            if response:
                res = response.json()
                print(f"获取店铺资质信息ocr返回：{res}")
                new_dic = dict()
                data = res['words_result']['struct_info']['group']
                for ite in data:
                    new_dic[ite['key'][0]['word']] = ite['value'][0]['word']
                print(f'ocr资质数据信息:{new_dic}')
                return new_dic
            else:
                return None
        except:
            return None

    def get_ocr_res(self, img):
        # 功能：对截图做去水印后调用百度 OCR，返回识别出的文字结果列表。
        try:
            image = self.remove_watermark(img)
            res_image = self.client.basicGeneral(image)
            data = res_image.get('words_result', '')
            print(f'百度api返回结果:{data}')

            return data
        except:
            return None

    def get_title(self):
        # 功能：从商品详情页提取当前标题，作为第一层匹配和落库名称来源。
        try:
            print('开始提取标题')
            time.sleep(self.get_sleep_time())
            title_xpath = '//*[@resource-id="com.xunmeng.pinduoduo:id/tv_title"]'
            if self.d.xpath(title_xpath).exists:
                title = self.d.xpath(title_xpath).info['contentDescription'].strip()
            else:
                # 标题没找到 → 检测是否是验证码页面
                self.check_and_solve_captcha()
                return None
            # title = self.d.xpath('//*[@resource-id="com.xunmeng.pinduoduo:id/tv_title"]').info['contentDescription'].strip()
            print(f'提取到标题：{title}')
            return title
        except Exception as e:
            print(f'获取标题出错：{e}')
            return None
        # 从里面匹配出药品名和规格
        # drugs_name
        # specifications
        # match = re.search(r'([^\d]+)([\d\D]+)', title)

        # match = re.search(r'(\[[^\]]+\])(.+?)(\d+.*)', title)
        # if match:
        #     drugs_name = match.group(1).strip() + match.group(2).strip()
        #     specifications = match.group(3).strip()
        #     print("药品名:", drugs_name)
        #     print("规格:", specifications)
        #     print('完整药名:', drugs_name + specifications)
        #     return drugs_name, specifications
        # else:
        #     print("没有匹配到预期格式")

    def enter_shop(self):
        # 功能：进入店铺页，供后续读取店铺或资质信息时使用。
        """
        进店，方便提取资质环境
        :return:
        """
        # self.d.xpath('//*[@text="进店"]').click()
        self.d.xpath('//*[@text="店铺"]').click()
        time.sleep(self.get_sleep_time())

    #店铺去重
    def shop_is_exists(self, data):
        # 功能：按店铺去重校验，避免同类数据重复入库。
        # 1. 验证必要字段
        # 先校验去重所需字段是否齐全，避免把不完整的数据带到 SQL 条件里。
        required_keys = ['shop']
        if not all(key in data for key in required_keys):
            missing = [key for key in required_keys if key not in data]

            print(f"缺少必要字段: {', '.join(missing)}")
            return None

        shop_value = data.get('shop')
        if not shop_value or not str(shop_value).strip():
            print("shop 字段为空，无法执行去重查询")
            return False

        conn = None
        try:
            conn = get_mysql()
            with conn.cursor() as cur:
                query_sql = """
                    SELECT * FROM {}
                    WHERE shop = %s
                    LIMIT 1
                """.format(self.shop_table_name)

                cur.execute(query_sql, (
                    data['shop']
                ))

                result = cur.fetchone()

            return bool(result)  # 如果存在返回True，否则False
        except Exception as e:
            print(f"MySQL 错误: {str(e)}")
        finally:
            if conn:
                conn.close()

    def get_province_city(self,data):
        """
        从 retrieve_scrape_shop_info 表中查询已存在的 province 和 city，
        并赋值给 data['province_name'] 和 data['city_name']
        """
        print("获取店铺营业公司对应的省份和城市")
        shop_name = data.get('shop')
        if not shop_name:
            print("shop 字段为空，无法执行查询")
            return
        conn = None
        try:
            conn = get_mysql()
            with conn.cursor() as cur:
                # 查询 shop_info_middle 表，获取 province 和 city
                sql = "SELECT province, city, business_license_company FROM retrieve_scrape_shop_info WHERE shop = %s AND platform = 3  LIMIT 1"
                cur.execute(sql, (shop_name,))
                result = cur.fetchone()

                if result:
                    province, city, company = result
                    data['province_name'] = province if province else ''
                    data['city_name'] = city if city else ''
                    data['company_name'] = company if company else ''
                    print(f"店铺 {shop_name} 对应的省份和城市为: {province}, {city}, 公司: {company}")
                else:
                    print(f"未在 shop_info_middle 表中找到店铺：{shop_name}")
                    # 可根据业务需求设置默认值或保持原样
                    data['province_name'] = ''
                    data['city_name'] = ''
                    data['company_name'] = ''
        except Exception as e:
            print(f"查询省市信息失败: {str(e)}")
            # 异常时也可设置默认空值，避免后续代码因缺少键而报错
            data['province_name'] = ''
            data['city_name'] = ''
        finally:
            if conn:
                conn.close()




    def data_is_exists(self, data):
        # 功能：按价格、店铺、日期、平台,账户，轮次做去重校验，避免同类数据重复入库。
        # 1. 验证必要字段
        # 先校验去重所需字段是否齐全，避免把不完整的数据带到 SQL 条件里。
        required_keys = ['min_price', 'shop', 'scrape_date', 'platform']
        if not all(key in data for key in required_keys):
            missing = [key for key in required_keys if key not in data]

            print(f"缺少必要字段: {', '.join(missing)}")
            return None

        conn = None
        try:
            conn = get_mysql()
            with conn.cursor() as cur:
                query_sql = """
                    SELECT * FROM {}
                    WHERE min_price = %s
                    AND store_name = %s
                    AND scrape_date = %s
                    AND platform_id = %s AND collect_equipment_account_id = %s
                AND collect_round = %s
                """.format(self.table_name)

                cur.execute(query_sql, (
                    data['min_price'],
                    data['shop'],
                    data['scrape_date'],
                    data['platform'], data.get('collect_equipment_account_id', 0),
                data.get('collect_round', 0),
                ))

                result = cur.fetchone()

            return bool(result)  # 如果存在返回True，否则False
        except Exception as e:
            print(f"MySQL 错误: {str(e)}")
        finally:
            if conn:
                conn.close()

    def get_instructions_data(self):
        # 功能：在详情页中提取说明书/商品参数区域的关键字段，整理成统一字典。
        """
        确定有详情页之后之后，提取所有的详情页数据
        :return:
        """
        # 先把页面滚到说明书/参数区域附近，再开始解析键值对。
        for i in range(8):

            if self.d.xpath('//*[@text="品牌"]').exists or self.d.xpath('//*[@text="药品通用名"]').exists:
                self.d.swipe_ext("up", scale=0.1)
                print('开始采集详情数据')
                break
            self.d.swipe_ext("up", scale=0.5)
            time.sleep(self.get_sleep_time())

        # 阶段 2：进入"查看全部"区域，把折叠的参数信息完整展开。
        # 点击查看全部
        if self.d.xpath('//*[@text="品牌"]').exists:
            self.d.xpath('//*[@text="品牌"]').click()
        else:
            self.d.xpath('//*[@text="药品通用名"]').click()
        time.sleep(self.get_sleep_time())
        attr = dict()
        # 阶段 3：批量解析键值对文本，构造说明书字段字典。
        # # 获取详情页信息
        xpath = '//*[starts-with(@text,"商品参数")]/parent::*/parent::*/following-sibling::*/*/*/android.view.ViewGroup//android.widget.TextView'
        ddd = self.d.xpath(xpath).all()
        for i in range(0, len(ddd), 2):
            group = ddd[i:i + 2]
            attr[group[0].text] = group[1].text
        # 截图获取未获取到的数据
        # if not all(i in ['有效期', '生产企业', '批准文号', '药品规格', '产品规格'] for i in attr.keys()):
        if not all(i in ['有效期', '生产企业', '批准文号', '药品规格'] for i in attr.keys()):
            # 首轮解析拿不到关键字段时再补一次较短滑动，兼容参数区未完整展示的情况。
            self.d.swipe_ext("up", 0.4)
            time.sleep(self.get_sleep_time())
            xpath = '//*[starts-with(@text,"商品参数")]/parent::*/parent::*/following-sibling::*/*/*/android.view.ViewGroup//android.widget.TextView'
            ddd = self.d.xpath(xpath).all()
            for i in range(0, len(ddd), 2):
                group = ddd[i:i + 2]
                attr[group[0].text] = group[1].text
        print(f'当前说明书规格参数:{attr}')
        res_data = {
            # "有效期": attr.get('有效期',''),
            # "生产单位": attr['生产企业'],
            # "批准文号": attr['批准文号'],
            # "产品规格": attr.get('药品规格') if attr.get('药品规格', '') else attr.get('药品规格')
            "有效期": attr.get('有效期', ''),
            "生产单位": attr.get('生产企业', ''),
            "批准文号": attr.get('批准文号', ''),
            "产品规格": attr.get('药品规格', ''),
            "发货地": attr.get('发货地', '')
        }
        print(f'当前规格参数字典数据:{res_data}')
        return res_data

    def has_instructions(self):
        # 功能：判断当前详情页能否找到说明书/商品详情区域。
        """
        是否有详情页
        :return:如果有详情页返回True，否则返回False
        """
        # 没有说明书的无法采集具体数据
        max_attempts = 12  # 最大尝试次数
        attempt = 0  # 当前尝试次数

        while attempt < max_attempts:
            time.sleep(0.5)
            xpath = '//*[@text="商品详情"]'
            is_has_instructions = self.d.xpath(xpath).exists
            if is_has_instructions:
                return True  # 如果找到"商品详情"，则返回True
            self.d.swipe_ext("up", 0.3)
            attempt += 1

        return False  # 如果尝试次数达到最大次数，则返回False

    def distinct_target(self):
        # 功能：判断当前页面是否已经回到商品列表页。
        # 这里同时检查多个锚点，是为了兼容拼多多不同活动页和不同 UI 版本。
        if self.is_pdd_home_page():
            return False

        search_camera = self.d.xpath('//*[@content-desc="拍照搜索"]').exists
        filter_bar = self.d.xpath('//*[@text="筛选"]').exists
        repeat_buy = self.d.xpath('//*[@text="618大促"]').exists
        promo_banner = self.d.xpath('//*[@text="回头客常拼"]').exists
        subsidy = self.d.xpath('//*[@text="百亿补贴"]').exists
        list_page_xpaths = [
            '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[1]/android.widget.FrameLayout[2]/android.view.ViewGroup[1]/android.widget.LinearLayout[1]//android.support.v7.widget.RecyclerView[1]',
            '//android.support.v7.widget.RecyclerView',
            '//androidx.recyclerview.widget.RecyclerView',
        ]
        list_container = any(self.d.xpath(xpath).exists for xpath in list_page_xpaths)

        result = False
        if filter_bar and (list_container or repeat_buy or promo_banner or search_camera):
            result = True
        elif list_container and (filter_bar or repeat_buy or promo_banner or subsidy):
            result = True

        print(
            "distinct_target -> "
            f"filter_bar={filter_bar}, repeat_buy={repeat_buy}, promo_banner={promo_banner}, "
            f"search_camera={search_camera}, list_container={list_container}, result={result}"
        )
        return result

    def  enter_target_page(self):
        # 功能：进入搜索页、输入关键字并恢复排序/页位，为主循环建立起始页面。
        # 阶段 1：进入搜索框并提交当前任务的搜索词。
        # 注意：back_to_pdd_home_page + focus_search_box 已由 prepare_entry_before_enter_target 完成，这里不再重复。
        search_entry_candidates = [
            '//*[contains(@text, "搜索")]',
            '//*[contains(@content-desc, "搜索")]',
            '//*[@resource-id="android:id/content"]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[1]'
            '/android.widget.FrameLayout[1]/android.widget.LinearLayout[1]/android.widget.FrameLayout[1]'
            '/android.widget.LinearLayout[1]',
        ]
        # 多轮重试：避免"按 back 误退到桌面"导致搜索框始终找不到。
        for attempt in range(4):
            if self.d(className='android.widget.EditText').wait(timeout=2):
                break

            if not self.is_pdd_foreground():
                logging.info("enter_target_page: pdd不在前台，尝试重新拉起APP")
                self.start_app()
                time.sleep(self.get_sleep_time())

            clicked_search_entry = False
            for entry_xpath in search_entry_candidates:
                if self.d.xpath(entry_xpath).exists:
                    self.d.xpath(entry_xpath).click()
                    clicked_search_entry = True
                    time.sleep(self.get_sleep_time())
                    if self.d(className='android.widget.EditText').wait(timeout=2):
                        break

            if self.d(className='android.widget.EditText').wait(timeout=1):
                break

            # 兜底：点击顶部中间区域，兼容节点层级变化但视觉位置不变的机型。
            if not clicked_search_entry:
                try:
                    screen_width = self.d.info.get('displayWidth', 1080)
                    screen_height = self.d.info.get('displayHeight', 2400)
                    self.d.click(screen_width // 2, int(screen_height * 0.08))
                    time.sleep(self.get_sleep_time())
                except Exception as click_error:
                    logging.info(f"enter_target_page: 顶部兜底点击失败: {click_error}")

            if self.d(className='android.widget.EditText').wait(timeout=1):
                break

            logging.info(f"enter_target_page: 第 {attempt + 1}/4 次仍未找到搜索输入框，继续重试")

        if not self.d(className='android.widget.EditText').wait(timeout=2):
            current_package = (self.d.app_current() or {}).get("package", "")
            raise RuntimeError(f"enter_target_page: search input not found, current_package={current_package}")

        self.d(className='android.widget.EditText').click()
        time.sleep(self.get_sleep_time())
        self.d.send_keys(self.search_key, clear=True)
        time.sleep(self.get_sleep_time())

        if self.d.xpath('//*[@text="搜索"]').exists:
            self.d.xpath('//*[@text="搜索"]').click()
        else:
            self.d.press("enter")
        time.sleep(self.get_sleep_time())
        # 阶段 2：如果任务要求排序，则在首次进入结果页后先切到目标排序方式。
        # 排序只在进入列表后的第一次执行，避免恢复进度时重复切换排序方向。
        # if self.sort and self.sort_key == 0:
        #     self.li_or_lo(self.sort)
        # progress = self.wr_re("读", self.device_id)
        progress = None
        # 阶段 3：如有历史页码，则把列表大致恢复到目标位置。
        # 进度恢复逻辑目前停用，但保留按页滑动的入口，便于后续重新启用断点续跑。
        # if not progress and self.page > 0:
        #     self.scroll_to_target_page(self.page)

    def get_clipboard(self):
        # 功能：读取设备剪贴板内容，并去掉空值和首尾空白。
        self.loggerPdd.info(f"Clipboard content:{self.d.clipboard}")  # 打印调试信息
        clipboard_content = self.d.clipboard
        if clipboard_content is None:
            return ''
        return clipboard_content.strip()

    def get_product_link(self):
        # 功能：通过商品详情页的分享入口复制商品链接。
        product_link = ''
        print('开始获取商品链接')

        content_frame = self.d.xpath('//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]').exists
        print(content_frame)

        relative_layout = self.d.xpath(
            '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]').exists
        print(relative_layout)

        relative_layout2 = self.d.xpath(
            '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]').exists
        print(relative_layout2)

        Frame_Layout = self.d.xpath(
            '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[2]').exists
        print(Frame_Layout)
        ImageView = self.d.xpath(
            '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[2]/android.view.View[1]').exists
        print(ImageView)

        ImageView2 = self.d.xpath(
            '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[3]/android.view.View[1]').exists
        print(ImageView2)
        # 多种可能的"分享"按钮
        # 分享入口在不同商品页布局里位置不稳定，因此保留多套候选 XPath。
        dots_xpaths = [
            # '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[2]/android.view.View[1]',
            '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[1]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[last()]/android.view.View[1]',
            # '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.RelativeLayout[1]/android.widget.FrameLayout[2]/android.view.View[1]',
            # '//*[@resource-id="android:id/content"]/android.widget.FrameLayout[1]/android.widget.RelativeLayout[2]/android.widget.RelativeLayout[2]/android.widget.FrameLayout[3]/android.widget.ImageView[1]',
        ]

        # 阶段 1：遍历候选分享入口，找到当前布局下可点击的"更多/分享"按钮。
        max_retry = 5  # 最多尝试次数
        # 分享面板偶尔会因为动画或按钮未露出而失败，因此允许多次重试。
        for idx in range(1, max_retry + 1):
            if product_link:  # 已经拿到则退出
                break

            for xp in dots_xpaths:
                if self.d.xpath(xp).exists:
                    # print(f'{idx}-进入分享点点点')
                    self.loggerPdd.info(f'{idx}-进入分享点点点')

                    # 调用 click() 的目的是打开分享面板；
                    # 后续 slide_link() 和"复制链接"点击都依赖分享面板已经展开。
                    self.d.xpath(xp).click()
                    time.sleep(1)
                    self.loggerPdd.info('开始滑动')
                    # 这里先调用 slide_link()，是为了把"复制链接"按钮滑到当前可见区域。
                    self.slide_link()
                    time.sleep(0.2)
                    # 调用 click_exists() 的目的是直接触发系统复制动作，
                    # 调用后 get_clipboard() 才有机会读到最新商品链接。
                    self.d.xpath('//*[@text="复制链接"]').click_exists()
                    time.sleep(1)
                    product_link = self.get_clipboard()
                    time.sleep(0.5)

                    self.loggerPdd.info(f'{idx}-商品链接:{product_link}')
                    break  # 找到并执行后跳出内层循环

        if not product_link and idx < max_retry:
            time.sleep(0.5)  # 最后一次不需要再等待

        # time.sleep(100000)

        return product_link

    def integrate_data_v2(self):
        # 功能：在单个商品详情页内完成价格、链接、店铺、说明书、去重和落库的完整聚合流程。
        """
        基于入口配置统一校验标题、品牌和品规，替代内部大量硬编码分支。
        """
        # 阶段 1：先拿价格和标题，并在最早阶段过滤无关商品。
        # 价格优先走规格弹窗，因为这里还能顺便拿到已选规格文本。
        min_price, ext = self.drug_price_ex()
        title_info = self.get_title()

        if not title_info:
            print('标题获取为空')
            return
            # ========== 盒数提取（放在标题判断之后） ==========
        from box_script import extract_quantity_and_unit
        quantity, unit = extract_quantity_and_unit(ext)
        if quantity:
            print(f"✅ 提取到数量: {quantity}{unit}")
            one_box_price = min_price / quantity if min_price else 0
        else:
            print(f"⚠️ 未提取到数量，ext={ext}")
            one_box_price = 0
        # =============================================
        # 先只按标题/品牌做一次粗过滤，尽早淘汰无关商品。
        if not self.is_link_useful(title_info):
            self.unrelated_data += 1
            return
        # 规格弹窗提价失败时，再回退到详情页直接取价。
        if not min_price:
            min_price = self.drug_price()
            if not min_price:
                print('提取价格出错，回退到列表页')
                self.unrelated_data += 1
                return
        # 阶段 2：补齐商品链接和店铺信息，这两类字段是后续落库和去重的关键上下文。
        product_link = self.get_product_link()
        time.sleep(2)
        oss_url = ''
        try:
            result = self.screenshot_and_upload_oss()
            if result:
                oss_url = result
                print(f"OSS快照上传成功: {oss_url}")
            else:
                print("OSS快照上传失败（返回空）")
        except Exception as e:
            print(f"OSS快照异常: {e}")
            oss_url = ''
        # 有的页面店铺信息不在首屏，这里按配置决定是否直接读取还是先滑动到店铺区域。
        if self.direct_shop_lookup:
            shop = self.get_shop_name()
        else:
            # 先检查首屏是否已有店铺信息（进店按钮 或 店铺名关键词 或 底部店铺tab）
            has_shop_anchor = (
                self.d(textStartsWith="进店").exists
                or self.d.xpath('//*[@content-desc="店铺"]').exists
                or any(
                    self.d.xpath(f'//android.widget.TextView[contains(@text, "{kw}")]').exists
                    for kw in ['大药房', '旗舰店', '专卖店', '药房']
                )
            )
            if not has_shop_anchor:
                for _ in range(3):
                    self.d.swipe_ext("up", scale=0.3)
                    time.sleep(self.get_sleep_time())
                    if (
                        self.d(textStartsWith="进店").exists
                        or self.d.xpath('//*[@content-desc="店铺"]').exists
                        or any(
                            self.d.xpath(f'//android.widget.TextView[contains(@text, "{kw}")]').exists
                            for kw in ['大药房', '旗舰店']
                        )
                    ):
                        print('可以开始获取店铺名')
                        break
            shop = self.get_shop_name()

        if not shop:
            print('当前店铺名称为空')
            self.unrelated_data += 1
            return

        scrape_date = self.get_current_date()
        dup_data = {
            'min_price': min_price,
            'shop': shop,
            'scrape_date': scrape_date,
            'platform': '3'
        }
        # 同一天同店铺同价格的数据视为重复，避免重复入库。
        if self.data_is_exists(dup_data):
            print('存在相同数据不入库')
            self.back_to_list_page()
            return

        shop_data = {
            'shop': shop,
            'store_url': product_link,
            'scrape_date': scrape_date,
            'platform': 3,
            'create_time': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            'update_time': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }


        province_name = ''
        city_name = ''
        company_name = ''

        # 插入店铺数据
        if self.shop_is_exists(shop_data):
            print("店铺数据已存在,进行省市回填")
            self.get_province_city(shop_data)
            province_name = shop_data['province_name']
            city_name = shop_data['city_name']
            company_name = shop_data.get('company_name', '')
        else:
            # 获取产品详情页的店铺资质
            shop_info = self.get_license_info()
            print(f"店铺数据不存在，获取店铺资质信息：{shop_info}")
            print(f"店铺数据不存在，插入{self.shop_table_name}店铺表")
            if shop_info:
                shop_data.update(shop_info)
                print(f"组合店铺数据:{shop_data}")
            self.save_to_shop_database(shop_data)
            # 从合并后的 shop_data 提取字段，供后续 save_data 写入 retrieve_scrape_data
            province_name = shop_data.get('province', '')
            city_name = shop_data.get('city', '')
            company_name = shop_data.get('business_license_company', '')



        # 阶段 3：确认是否存在说明书页，并在有说明书时补提取规格、生产单位和批准文号。
        is_has_instructions = self.has_instructions()
        self.loggerPdd.info(f'是否有说明书:{is_has_instructions}')
        manufacture_date = ''
        shipmentProvinceName = ''
        credit_code = ext
        # 说明书页不是每个商品都有；没有时允许继续落库，只是相关字段留空。
        province_id, city_id = self._resolve_region_ids(province_name, city_name)
        if is_has_instructions:
            try:
                instructions_info = self.get_instructions_data()
                expiry_date = instructions_info['有效期'].strip('。')
                manufacturer = instructions_info['生产单位'].strip('。')
                approval_number = instructions_info['批准文号'].strip('。')
                specifications = instructions_info['产品规格'].strip('。')
                shipmentProvinceName = instructions_info.get('发货地', '').strip('。')
            except Exception as e:
                print(f'获取详情页规格参数出错：{e}')
                return
        else:
            expiry_date = ''
            manufacturer = ''
            approval_number = ''
            specifications = ''

        # 二次校验把说明书里的规格也纳入判断，避免标题模糊匹配带来误采。
        if not self.is_link_useful(title_info, specifications):
            self.unrelated_data += 1
            return

        self.unrelated_data = 0
        # 不再依赖盒数处理脚本，单盒价格默认回退为 0。
        one_box_price = 0

        # 阶段 4：把当前详情页提取结果整理成统一落库结构。
        save_data = {
            'enterprise_id': self.enterprise_id,
            'platform_id': self.platform,
            'platform_item_id': '',
            'province_id': province_id,
            'city_id': city_id,
            'province_name': province_name,
            'city_name': city_name,
            'area_info': "",
            'product_brand': self.brand,
            'product_name': title_info,
            'product_specs': specifications,
            'search_name': self.search_key,
            'collect_config_info': self.collect_config_info,
            'one_box_price': one_box_price,
            'manufacture_date': manufacture_date,
            'expiry_date': expiry_date,
            'manufacturer': manufacturer,
            'approval_number': approval_number,
            'is_sold_out': 0,
            'online_posting_count': 1,
            'continuous_listing_count': 1,
            'link_url': product_link,
            'store_name': shop,
            'store_url': '',
            'shipment_province_id': 0,
            'shipment_province_name': shipmentProvinceName,
            'shipment_city_id': 0,
            'shipment_city_name': "",
            'company_name': company_name,
            'qualification_number': "",
            'scrape_date': scrape_date,
            'min_price': min_price,
            'number': 1,
            'sales': "",
            'inventory': "",
            'snapshot_url': oss_url or '',
            'insert_time': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            'update_time': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            'collect_equipment_account_id': self.collect_equipment_account_id,
            'collect_region_id': self.collect_region_id,
            'collect_round': self.collect_round,
        }
        # 调用 save_to_database() 的目的是把当前已经校验通过的数据立即持久化，
        # 避免后续页面跳转、返回或异常中断导致采集结果丢失。
        self.save_to_database(save_data)
        return True
    def main(self, device_id, search_key_length, keyword_idx):
        # 功能：执行单设备的完整采集主循环，直到达到结束页、采集上限或异常退出条件。
        completed_normally = False
        stop_by_max_count = False
        spider_no = 0
        current_page = self.page
        consecutive_empty_pages = 0
        # 阶段 1：建立设备连接，准备进入搜索页面。
        self.connect_devices(device_id)
        time.sleep(self.get_sleep_time())

        # 统一入口准备：按需重启，然后回到主页面并点出搜索框。
        if self.is_pdd_foreground():
            logging.info("当前在拼多多可操作页面，执行统一入口准备（回主页面+点搜索框）")
            self.prepare_entry_before_enter_target(force_restart=False)
        else:
            logging.info("当前不在拼多多可操作页面，先重启，再执行统一入口准备")
            self.prepare_entry_before_enter_target(force_restart=True)
        self.enter_target_page()

        # 阶段 2：向调度系统上报"执行中"，然后开始按页扫描商品列表。
        #  上报状态
        # 进入主循环前先上报"执行中"，让调度系统能看到设备已经开始跑任务。
        report_api(self.task_id, platform=self.platform, username=getattr(self, 'username', ''),
                   page=self.page, is_finished=0)
        # ========== 新增：启动心跳上报线程 ==========
        self._heartbeat_running = True

        def heartbeat_worker():
            while self._heartbeat_running:
                time.sleep(HEARTBEAT_INTERVAL_SECONDS)
                if self._heartbeat_running:
                    report_heartbeat(self.task_id, getattr(self, 'page', 0), device_id=self.device_id)

        heartbeat_thread = threading.Thread(target=heartbeat_worker, daemon=True)
        heartbeat_thread.start()
        # ==========================================
        try:
            for idx in range(300):
                print(f'第{current_page}页')
                # self.wr_re("写", self.device_id, self.sort, current_page)
                if spider_no > 30:
                    time.sleep(300)
                    spider_no = 0
                # 连续命中太多无关商品时，认为当前搜索结果已偏离目标，主动收尾。
                if self.unrelated_data > 15:
                    print(f'[{self.program_start_time}]----{self.search_key}----连续超过15个不达标的数据则停止采集')
                    print(
                        f"[程序启动时间：{self.program_start_time}-----程序结束时间：{self.app_current_time()}]----搜索关键词：{self.search_key}----点击了{self.click_counts}个商品")
                    completed_normally = self.finish_task_normally(
                        current_page,
                        '连续超过15个不达标的数据，结束采集'
                    )
                    break
                # 达到采集上限后不再继续翻页，直接走正常结束分支。
                if self.is_max_count_reached():
                    completed_normally = self.finish_task_with_max_count(current_page)
                    # 向下滑
                    self.swipe_down()
                    time.sleep(self.get_sleep_time())
                    # 点击搜索框
                    click_success = self.click_target_product_by_search_key(fuzzy_match=False)
                    if not click_success:
                        print(f"关键词「{self.search_key}」商品点击失败")
                        break
                    print("点击搜索框")
                    self.d(className='android.widget.EditText').click()
                    time.sleep(self.get_sleep_time())
                    break

                # 售罄次数大于4基本就是号废了但是如果下次点击不会出现这种情况就要重置为0
                # 连续多次命中售罄商品时，认为当前账号/结果页已失去采集价值，提前退出。
                if self.sold_out_counts > 4:
                    self.finish_task_abnormally(
                        current_page,
                        "====商品已售罄4次，结束采集（号不能用）",
                        finish_status=1
                    )
                    print(
                        f"[程序启动时间：{self.program_start_time}-----程序结束时间：{self.app_current_time()}]----搜索关键词：{self.search_key}----点击了{self.click_counts}个商品")
                    break
                # 阶段 3：获取当前页可见商品卡片，并逐个点击进入详情页采集。
                # 检测"官方旗舰店"区域：如果"进店逛逛"和店铺icon同时存在，
                # 说明列表顶部有旗舰店入口，先下滑跳过该区域再取商品列表。
                if self.d.xpath('//*[@text="进店逛逛"]').exists or \
                   self.d.xpath('//*[@text="进店"]').exists:
                    print('检测到官方旗舰店区域，向下滑动半屏跳过')
                    screen_w = self.d.info.get('displayWidth', 1080)
                    screen_h = self.d.info.get('displayHeight', 2400)
                    self.d.swipe(
                        screen_w // 2,
                        int(screen_h * 0.7),
                        screen_w // 2,
                        int(screen_h * 0.2),
                        duration=0.4
                    )
                    time.sleep(self.get_sleep_time())
                drug_lis = self.get_drug_lis(idx)
                print('数量', len(drug_lis))
                if not drug_lis:
                    consecutive_empty_pages += 1
                    logging.warning(
                        f"main: 第 {current_page} 页未识别到商品，连续空页次数={consecutive_empty_pages}"
                    )

                    if not self.distinct_target():
                        logging.warning("main: 当前疑似不在商品列表页，尝试回退恢复列表页")
                        recovered = self.back_to_list_page()
                        if not recovered and (
                            self.is_pdd_home_page()
                            or (self.d(className='android.widget.EditText').exists and not self.d.xpath('//*[@text="筛选"]').exists)
                        ):
                            logging.warning("main: 已落到首页/搜索输入页，直接重新进入搜索结果页")
                            self.enter_target_page()
                            current_page = self.page
                            consecutive_empty_pages = 0
                            continue
                        time.sleep(self.get_sleep_time())
                        drug_lis = self.get_drug_lis(idx)
                        print('恢复后数量', len(drug_lis))
                    else:
                        logging.warning("main: 仍在列表页但商品列表为空，先做一次轻微下滑恢复")
                        self.swipe_down()
                        time.sleep(self.get_sleep_time())
                        drug_lis = self.get_drug_lis(idx)
                        print('下滑恢复后数量', len(drug_lis))

                    if drug_lis:
                        consecutive_empty_pages = 0
                    elif consecutive_empty_pages >= 3:
                        logging.warning("main: 连续空页达到阈值，重新进入搜索结果页")
                        self.enter_target_page()
                        current_page = self.page
                        consecutive_empty_pages = 0
                        continue
                for idd, drug_one in enumerate(drug_lis):
                    print(idd + 1, drug_one.info)
                    time.sleep(self.get_sleep_time())
                    top = drug_one.info['bounds']['top']
                    bottom = drug_one.info['bounds']['bottom']
                    _, screen_h = self.get_screen_size()
                    if bottom <= int(screen_h * 0.93) and top >= int(screen_h * 0.16):
                        drug_one.click()
                        self.click_counts += 1
                        time.sleep(self.get_sleep_time())
                        # 先判断是否售罄次数是否大于4
                        if self.sold_out_counts >= 4:
                            print(
                                f"[程序启动时间：{self.program_start_time}-----程序结束时间：{self.app_current_time()}]----搜索关键词：{self.search_key}----点击了{self.click_counts}个商品")
                            self.finish_task_abnormally(
                                current_page,
                                "====这是在第一页有两个，商品已售罄4次，结束采集（号不能用）====",
                                finish_status=1
                            )
                            time.sleep(self.get_sleep_time())
                            self.d.press('home')
                            stop_by_max_count = True
                            break
                        # 这里先判断"商品已售罄"，是为了尽早放弃无效详情页；
                        # 如果不先做这一步，后续详情采集会浪费时间且可能干扰账号状态判断。
                        if self.d.xpath('//*[contains(@text, "商品已售罄")]').wait(timeout=5):
                            print("======商品已售罄======")
                            self.sold_out_counts += 1
                            if self.back_to_list_page():
                                continue
                        # 采集药品信息
                        # 进入详情页后的采集与回退是最容易卡死的阶段，需要单独兜底。
                        try:
                            # 重置商品售罄次数
                            self.sold_out_counts = 0

                            saved = self.integrate_data_v2()
                            # 新加判断逻辑（增加采集速度），如果保存成功数据返回true，然后执行点商品参数的×，否则就执行一次返回
                            if saved:
                                x_btn = self.d.xpath('//*[@text=""]')
                                if x_btn.exists:
                                    x_btn.click()
                                    time.sleep(random.uniform(0.5, 1))
                                self.d.press("back")
                                time.sleep(random.uniform(0.5, 1))
                            else:
                                self.d.press("back")
                                time.sleep(random.uniform(0.5, 0.8))

                            # 检测下是否回退到列表页
                            if self.back_to_list_page():
                                print('回退到列表页', True)
                            else:
                                print(f'[{self.app_current_time()}] 回退到列表页失败')
                                print(
                                    f"[程序启动时间：{self.program_start_time}-----结束时间：{self.app_current_time()}]----搜索关键词：{self.search_key}----点击了{self.click_counts}个商品")
                                self.finish_task_abnormally(current_page, "回退到列表页失败，结束采集")
                                stop_by_max_count = True
                                break
                            time.sleep(self.get_sleep_time())
                            spider_no += 1
                            if self.is_max_count_reached():
                                completed_normally = self.finish_task_with_max_count(current_page)
                                stop_by_max_count = True
                                break
                        except Exception as e:
                            self.loggerPdd.error(f'采集药品详情数据出错:{e}')
                            if not self.back_to_list_page():
                                self.finish_task_abnormally(current_page, '采集药品详情数据出错且无法回到列表页，结束采集')
                                stop_by_max_count = True
                                break
                            else:
                                continue
                if drug_lis:
                    consecutive_empty_pages = 0
                # 阶段 4：处理翻页前的收尾条件，包括采集上限、结束页和列表到底。
                if stop_by_max_count:
                    break
                # 配置了结束页时，以调用方传入的页边界作为最高优先级退出条件。
                if self.end_page is not None and current_page >= self.end_page:
                    completed_normally = self.finish_task_normally(
                        current_page,
                        f"已采集到结束页 {self.end_page}，结束任务"
                    )
                    break
                if self.d(textStartsWith="抱歉，没有更多商品啦~").exists:
                    completed_normally = self.finish_task_normally(current_page, '已经到达列表页最底部')
                    break
                # 阶段 5：当前页还没触发任何结束条件时，继续滑到下一页。
                print('开始滑入下一页')
                # 修复滑动抖动：使用固定的滑动参数
                screen_width = self.d.info.get('displayWidth', 1080)
                screen_height = self.d.info.get('displayHeight', 2400)

                # 固定滑动参数，不随页数变化
                start_x = screen_width // 2  # 屏幕中心
                start_y = int(screen_height * 0.8)  # 从屏幕80%高度开始
                end_y = int(screen_height * 0.2)  # 滑动到屏幕20%高度

                self.d.swipe(start_x, start_y, start_x, end_y, duration=random.uniform(0.2, 0.4))
                time.sleep(self.get_sleep_time())
                current_page += 1
                self.page = current_page
        finally:
            # ========== 新增：停止心跳上报线程 ==========
            self._heartbeat_running = False
            # ==========================================

        # 阶段 6：根据最终状态做统一收尾，保证任务一定会走到正常或异常结束分支之一。
        if completed_normally:
            self.clear_progress_file()
        elif not self.finish_reported:
            self.finish_task_abnormally(current_page, "采集流程异常结束")
        return completed_normally
# pdd
def main():
    # 功能：启动调度器入口，先立即执行一轮派单，再注册后续轮询。
    interval_seconds = MANUAL_SCHEDULER_INTERVAL_SECONDS if USE_MANUAL_TASKS else SCHEDULER_INTERVAL_SECONDS
    logging.info(f"PDD 调度器启动，轮询间隔 {interval_seconds} 秒")
    dispatch_pending_tasks()
    if USE_MANUAL_TASKS:
        logging.info(f"手动任务模式：启动定时轮询，间隔 {MANUAL_SCHEDULER_INTERVAL_SECONDS} 秒")
        if all_manual_tasks_dispatched() and not has_active_workers():
            logging.info("手动任务模式：全部任务已完成，进程退出")
            return
        schedule_dispatch(MANUAL_SCHEDULER_INTERVAL_SECONDS, manual_scheduled_dispatch_job)
        scheduler_stop_event.wait()
        return
    schedule_dispatch(SCHEDULER_INTERVAL_SECONDS)
    scheduler_stop_event.wait()
if __name__ == '__main__':
   main()