first commit

.idea/.gitignore

@@ -0,0 +1,8 @@
+# 默认忽略的文件
+/shelf/
+/workspace.xml
+# 基于编辑器的 HTTP 客户端请求
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

.idea/inspectionProfiles/profiles_settings.xml

@@ -0,0 +1,6 @@
+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>

.idea/misc.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Black">
+    <option name="sdkName" value="Python 3.13" />
+  </component>
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.13" project-jdk-type="Python SDK" />
+</project>

.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/taobao_detail.iml" filepath="$PROJECT_DIR$/.idea/taobao_detail.iml" />
+    </modules>
+  </component>
+</project>

.idea/taobao_detail.iml

@@ -0,0 +1,12 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+  <component name="PyDocumentationSettings">
+    <option name="format" value="PLAIN" />
+    <option name="myDocStringFormat" value="Plain" />
+  </component>
+</module>

.idea/vcs.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>

README.md

@@ -0,0 +1,39 @@
+# 淘宝商品详情采集
+
+淘宝药品/商品数据采集系统，支持列表搜索和详情页 SKU 级抓取。
+
+
+
+
+## 依赖
+
+- Python 3.8+
+- DrissionPage（浏览器自动化）
+- curl_cffi（API 请求）
+- lxml
+- MySQL
+
+```bash
+pip install DrissionPage curl_cffi lxml pymysql
+```
+
+需要本地安装 Chrome 浏览器，路径默认 `C:\Program Files\Google\Chrome\Application\chrome.exe`。
+
+## detail.py 说明
+
+从 `retrieve_scrape_data` 表读取带 `+` 的商品（一条链接含多个 SKU），打开详情页逐个点击规格，拆解出每条 SKU 的独立记录写入数据库。
+
+**流程：**
+1. 查询当天 `platform_item_id` 含 `+` 的记录，按 `pre_id` 去重
+2. 浏览器打开商品链接
+3. 点击 SKU 列表逐一抓取：标题、价格、链接
+4. 删除旧聚合记录，逐条写入拆分后的 SKU 记录
+
+**运行：**
+```bash
+python detail.py
+```
+
+## 注意事项
+
+- `detail.py` 删写操作在事务内执行，失败自动回滚。

commons/Logger.py

@@ -0,0 +1,51 @@
+import logging
+from logging.handlers import TimedRotatingFileHandler
+import os
+from datetime import datetime
+
+LOG_FORMAT = "%(asctime)s [%(levelname)s] %(message)s"
+
+
+def _build_logger(name: str, log_file: str) -> logging.Logger:
+    logger_obj = logging.getLogger(name)
+    if logger_obj.handlers:
+        return logger_obj
+
+    logger_obj.setLevel(logging.DEBUG)
+    logger_obj.propagate = False
+
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(logging.INFO)
+
+    file_handler = logging.FileHandler(log_file, encoding="utf-8")
+    file_handler.setLevel(logging.DEBUG)
+
+    formatter = logging.Formatter(LOG_FORMAT)
+    console_handler.setFormatter(formatter)
+    file_handler.setFormatter(formatter)
+
+    logger_obj.addHandler(console_handler)
+    logger_obj.addHandler(file_handler)
+    return logger_obj
+
+
+def get_spider_logger(spider_name: str) -> logging.Logger:
+    base_dir = os.path.join("logs", spider_name)
+    os.makedirs(base_dir, exist_ok=True)
+    today_str = datetime.now().strftime("%Y-%m-%d")
+    log_file = os.path.join(base_dir, f"taobao_{today_str}.log")
+    return _build_logger(f"spider.{spider_name}", log_file)
+
+# 获取今天日期字符串
+today_str = datetime.now().strftime("%Y-%m-%d")
+legacy_dir = "logs"
+os.makedirs(legacy_dir, exist_ok=True)
+legacy_file = os.path.join(legacy_dir, f"taobao_{today_str}.log")
+logger = _build_logger("MyLogger", legacy_file)
+
+# 测试
+# logger.debug("调试信息")
+# logger.info("普通提示信息")
+# logger.warning("警告信息")
+# logger.error("错误信息")
+# logger.critical("严重错误信息")

commons/collect_schedule_runner.py

@@ -0,0 +1,126 @@
+"""定时从调度库拉取任务、执行爬虫并上报状态的通用入口。"""
+import random
+import time
+import json
+import requests
+
+from commons.Logger import logger
+from commons.conn_mysql import MySQLPoolOnline
+from commons.feishu_webhook import send_text
+
+RESULT_REPORT_URL = "http://scheduleapi.findit.ltd/api/collect_equipment_execute/result_report"
+
+_TASK_SELECT = (
+    "select `id`,`collect_task_id`,`company_id`,`product_name`,`product_specs`,"
+    "`product_keyword`,`product_brand`,`sampling_cycle`,`sampling_start_time`,`sampling_end_time`,`collect_equipment_account_id`,`collect_region_id`,`collect_equipment_id`,`collect_round` "
+    "from `retrieve_collect_task_allocate` where `platform`=%s and `status`=1 limit 1"
+)
+
+
+class CollectScheduleRunner:
+    """按平台 ID 拉取一条待执行任务，运行爬虫并上报结果。"""
+
+    def __init__(self, platform_name, platform_id, spider_cls):
+        self.platform_name = platform_name
+        self.platform_id = platform_id
+        self.spider_cls = spider_cls
+        self.db_online = MySQLPoolOnline()
+        self.crawl_count = ""
+        self.task_id = ""
+        self.task_dict = None
+
+    def _report_status(self, status):
+        if status not in (2, 3, 4):
+            logger.warning("未知状态值: %s, 跳过状态上报", status)
+            return
+        now = int(time.time())
+        if status == 2:
+            params = {
+                "collect_task_allocate_id": self.task_id,
+                "status": status,
+                "finish_status": 0,
+                "start_time": now,
+            }
+        elif status == 3:
+            params = {
+                "collect_task_allocate_id": self.task_id,
+                "status": status,
+                "finish_status": 1,
+                "real_count": self.crawl_count,
+                "end_time": now,
+            }
+        else:
+            params = {
+                "collect_task_allocate_id": self.task_id,
+                "status": status,
+                "finish_status": 0,
+                "end_time": now,
+            }
+
+        try:
+            res = requests.get(RESULT_REPORT_URL, params=params, timeout=20)
+            res.raise_for_status()
+            logger.debug("状态上报: %s", res.text[:500])
+        except Exception as e:
+            logger.warning("状态上报失败: %s", e)
+
+    def get_task(self):
+        task_list = self.db_online.select_data(_TASK_SELECT, (self.platform_id,))
+        if not task_list:
+            return {}
+        task_dict = task_list[0]
+        self.task_id = task_dict["id"]
+        logger.info(json.dumps(task_dict))
+        return task_dict
+
+    def heartbeat_task(self):
+        url = "https://scheduleapi.findit.ltd/api/collect_equipment_execute/heartbeat"
+        params = {
+            "collect_task_allocate_id": self.task_id,
+        }
+        try:
+            res = requests.get(url, params=params, timeout=20)
+            logger.info("心跳任务上报成功")
+        except Exception as e:
+            logger.info("心跳任务上报失败")
+
+    def run(self):
+        self.task_dict = self.get_task()
+        if not self.task_dict:
+            logger.info("%s暂无任务", self.platform_name)
+            return
+        self._report_status(2)
+        self.crawl_count, is_success = self.spider_cls(self.task_dict).run()
+
+        self.heartbeat_task()
+        send_text(
+            f"{time.strftime('%Y-%m-%d %H:%M:%S')} 通知:\n"
+            f"平台: {self.platform_name}, 药品: {self.task_dict.get('product_name')}, "
+            f"爬取数据: {self.crawl_count}条"
+        )
+        self._report_status(3 if is_success else 4)
+
+
+def run_scheduled_loop(
+    platform_name,
+    platform_id,
+    spider_cls,
+    *,
+    interval_minutes=5,
+    sleep_seconds=3,
+):
+    """循环拉取任务并执行爬虫；每轮结束后休眠 interval_minutes 分钟。"""
+    idle_seconds = random.randint(180,300)
+    logger.info(
+        "循环任务已启动，平台=%s，每轮间隔 %s 秒",
+        platform_name,
+        idle_seconds,
+    )
+    while True:
+        try:
+            logger.info("开始执行%s爬虫任务", platform_name)
+            CollectScheduleRunner(platform_name, platform_id, spider_cls).run()
+            logger.info("%s爬虫任务执行完成", platform_name)
+        except Exception as e:
+            logger.error("%s爬虫任务执行失败: %s", platform_name, e, exc_info=True)
+        time.sleep(idle_seconds)

commons/config.py

@@ -0,0 +1 @@
+

commons/conn_mysql.py

@@ -0,0 +1,176 @@
+import logging
+import os
+
+import pymysql
+from dbutils.pooled_db import PooledDB
+from pymysql.cursors import DictCursor
+
+logger = logging.getLogger(__name__)
+
+# 连接池公共参数（可自行调大 maxconnections 等）
+_POOL_COMMON = {
+    "maxconnections": 10,
+    "mincached": 2,
+    "maxcached": 5,
+    "blocking": True,
+}
+
+
+def _create_pool(
+        host,
+        port,
+        user,
+        password,
+        database,
+):
+    return PooledDB(
+        creator=pymysql,
+        cursorclass=DictCursor,
+        charset="utf8mb4",
+        host=host,
+        port=port,
+        user=user,
+        password=password,
+        database=database,
+        **_POOL_COMMON,
+    )
+
+
+class _MySQLPoolBase:
+    """共用的取连接、写库、读库逻辑，避免两套类复制粘贴。"""
+
+    def __init__(self, pool):
+        self.pool = pool
+
+    def get_conn(self):
+        return self.pool.connection()
+
+    def _close(self, conn, cursor) -> None:
+        try:
+            if cursor:
+                cursor.close()
+        finally:
+            if conn:
+                conn.close()
+
+    def _write(self, fn, *args, **kwargs):
+        conn = self.get_conn()
+        cursor = conn.cursor()
+        try:
+            result = fn(cursor, *args, **kwargs)
+            conn.commit()
+            return result
+        except Exception as e:
+            logger.error("SQL 执行错误: %s", e)
+            conn.rollback()
+            return None
+        finally:
+            self._close(conn, cursor)
+
+    def execute_many(self, sql, data):
+        rows = list(data)
+        if not rows:
+            return
+
+        def _do(c, s, d):
+            c.executemany(s, d)
+
+        self._write(_do, sql, rows)
+
+    def execute_one(self, sql, data):
+        def _do(c, s, d):
+            c.execute(s, d)
+
+        self._write(_do, sql, data)
+
+    def update_data(self, sql, data):
+        self.execute_one(sql, data)
+
+    def select_data(self, sql, data=None):
+        """只读查询，不应 commit。"""
+        conn = self.get_conn()
+        cursor = conn.cursor()
+        try:
+            if data is not None:
+                cursor.execute(sql, data)
+                return cursor.fetchall()
+            else:
+                cursor.execute(sql)
+                return cursor.fetchall()
+        except Exception as e:
+            logger.error("SQL 查询错误: %s", e)
+            return []
+        finally:
+            self._close(conn, cursor)
+
+    def execute(self, sql, data=None):
+        """通用写操作，返回受影响行数；失败返回 0。"""
+
+        def _do(c, s, d):
+            if d is not None:
+                return c.execute(s, d)
+            return c.execute(s)
+
+        result = self._write(_do, sql, data)
+        return result if isinstance(result, int) else 0
+
+
+class MySQLPool(_MySQLPoolBase):
+    """test2 库（可通过环境变量覆盖，便于部署；未设置则沿用原默认）。"""
+
+    def __init__(self):
+        pool = _create_pool(
+            host=os.environ.get("MYSQL_TEST2_HOST", "47.119.164.65"),
+            port=int(os.environ.get("MYSQL_TEST2_PORT", "3306")),
+            user=os.environ.get("MYSQL_TEST2_USER", "test_c"),
+            password=os.environ.get("MYSQL_TEST2_PASSWORD", "Dfwy@2025"),
+            database=os.environ.get("MYSQL_TEST2_DB", "test2"),
+        )
+        super().__init__(pool)
+
+
+class MySQLPool39(_MySQLPoolBase):
+    """drug_retrieve 库。"""
+
+    def __init__(self):
+        pool = _create_pool(
+            host=os.environ.get("MYSQL_DRUG_HOST", "39.108.116.125"),
+            port=int(os.environ.get("MYSQL_DRUG_PORT", "3306")),
+            user=os.environ.get("MYSQL_DRUG_USER", "drug_retrieve"),
+            password=os.environ.get("MYSQL_DRUG_PASSWORD", "Pem287cwM58jNpe2"),
+            database=os.environ.get("MYSQL_DRUG_DB", "drug_retrieve"),
+        )
+        super().__init__(pool)
+
+
+class MySQLPoolOnline(_MySQLPoolBase):
+    """drug_retrieve 库。"""
+
+    def __init__(self):
+        pool = _create_pool(
+            host=os.environ.get("MYSQL_DRUG_HOST", "120.25.48.236"),
+            port=int(os.environ.get("MYSQL_DRUG_PORT", "3306")),
+            user=os.environ.get("MYSQL_DRUG_USER", "drug_retrieve"),
+            password=os.environ.get("MYSQL_DRUG_PASSWORD", "ksCt3xm6chzdkafj"),
+            database=os.environ.get("MYSQL_DRUG_DB", "drug_retrieve"),
+        )
+        super().__init__(pool)
+
+class MySQLPoolOn2(_MySQLPoolBase):
+    """drug_retrieve 库。"""
+
+    def __init__(self):
+        pool = _create_pool(
+            host=os.environ.get("MYSQL_DRUG_HOST", "120.24.26.108"),
+            port=int(os.environ.get("MYSQL_DRUG_PORT", "3307")),
+            user=os.environ.get("MYSQL_DRUG_USER", "root"),
+            password=os.environ.get("MYSQL_DRUG_PASSWORD", "zhijiayun123456"),
+            database=os.environ.get("MYSQL_DRUG_DB", "drug_retrieve_test"),
+        )
+        super().__init__(pool)
+
+
+if __name__ == "__main__":
+    pool = MySQLPool()
+    rows = pool.select_data("SELECT 1 AS one")
+    print(rows)

commons/feishu_webhook.py

@@ -0,0 +1,69 @@
+import requests
+import json
+import time
+
+WEBHOOK_URL = "https://open.feishu.cn/open-apis/bot/v2/hook/f8a70e6b-6b75-43c2-aaf2-9ba8f79f40f6"
+
+
+def send_text(text):
+    data = {
+        "msg_type": "text",
+        "content": {"text": text},
+    }
+    headers = {"Content-Type": "application/json"}
+    try:
+        response = requests.post(
+            WEBHOOK_URL,
+            headers=headers,
+            data=json.dumps(data),
+            timeout=5
+        )
+    except Exception as e:
+        pass
+
+
+def send_error_card(task_name, err_msg, mention_all=False):
+    """发送异常红色卡片"""
+    content = (
+        f"**任务**：{task_name}\n"
+        f"**状态**：❌ 失败\n"
+        f"**错误**：{err_msg}\n"
+        f"**时间**：{time.strftime('%Y-%m-%d %H:%M:%S')}"
+    )
+    if mention_all:
+        content = f"<at id=all></at>\n{content}"
+
+    data = {
+        "msg_type": "interactive",
+        "card": {
+            "config": {"wide_screen_mode": True},
+            "header": {
+                "template": "red",
+                "title": {"tag": "plain_text", "content": "异常告警"},
+            },
+            "elements": [
+                {
+                    "tag": "div",
+                    "text": {"tag": "lark_md", "content": content},
+                }
+            ],
+        },
+    }
+    headers = {"Content-Type": "application/json"}
+    response = requests.post(
+        WEBHOOK_URL,
+        headers=headers,
+        data=json.dumps(data),
+        timeout=5,
+    )
+    print(response.json())
+
+
+# 使用示例
+if __name__ == "__main__":
+    account1 = "aaaaa"
+    plat_form = "京东"
+    drug = "999小儿感冒颗粒24粒"
+    drug_count = 128
+    text = f"**重要通知** {str(time.strftime("%Y-%m-%d %H:%M:%S"))}\n 账号{account1}, {plat_form} 采集 {drug} 数据 {drug_count} 条"
+    send_text(text)

commons/scheduler.py

@@ -0,0 +1,121 @@
+import time
+import requests
+from commons.Logger import get_spider_logger
+import threading
+
+logger = get_spider_logger('scheduler')
+
+
+class CrawlerScheduler:
+    """爬虫任务调度器"""
+
+    def __init__(self, DEVICE_ID, platform, heartbeat_interval=30):
+        """
+        初始化调度器
+
+        Args:
+            platform: 平台名称
+            heartbeat_url: 心跳上报URL
+            heartbeat_interval: 心跳间隔时间（秒）
+        """
+
+        self.username = DEVICE_ID
+        self.platform = platform
+
+        self._lock = threading.Lock()
+
+        self.heartbeat_url = 'http://pricesys2.kailin.com.cn:8082/api/collect_task/heartbeat'
+        self.heartbeat_interval = heartbeat_interval
+        self.end = False
+        self.heartbeat_thread = None
+
+    def _heartbeat_reporter(self):
+        """守护线程：只负责上报心跳"""
+        headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
+        while True:
+            try:
+
+                response = requests.post(
+                    self.heartbeat_url,
+                    json={
+                        "platform": self.platform,
+                        "username": self.username
+                    },
+                    headers=headers,
+                    timeout=3
+                )
+
+                print(f"[心跳] 发送成功: {response.status_code}")
+                result = response.json()
+
+                if self.end or result.get('code') != 'success':
+                    logger.error(f'心跳回传：{result}')
+                    self.set_flag(True)
+                    # 写日志
+                    break
+
+                logger.info(f'心跳回传：{result}')
+                self.set_flag(False)
+
+                time.sleep(self.heartbeat_interval)
+
+            except Exception as e:
+                print(e)
+                logger.error(e)
+                time.sleep(5)
+
+    def start(self):
+        """启动调度器"""
+        self.set_flag(False)
+        self.heartbeat_thread = threading.Thread(
+            target=self._heartbeat_reporter,
+            daemon=True
+        )
+        self.heartbeat_thread.start()
+
+    def stop(self):
+        """停止调度器"""
+        self.set_flag(True)
+        logger.info('心跳停止')
+
+    def get_task(self):
+        try:
+            task_api = "http://pricesys2.kailin.com.cn:8082/api/collect_task/pull"
+            headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
+            params = {
+                'platform': self.platform,
+                'username': self.username
+            }
+            response = requests.get(task_api, params=params, headers=headers, timeout=5)
+            result = response.json()
+            logger.info(f'拉取任务：{result}')
+            if result.get('code') == 'success':
+                return result.get('data').get('task')
+
+            print('拉取任务返回', result)
+
+        except Exception as e:
+            logger.error('获取任务报错', e)
+
+    def post_report(self, data):
+        try:
+            url = "http://pricesys2.kailin.com.cn:8082/api/collect_task/report"
+            print('传给返回接口的数据', data)
+            logger.info(f'report上传数据：{data}')
+            headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
+            response = requests.post(url, json=data, headers=headers, timeout=5)
+            # 记录日志
+            result = response.json()
+
+            logger.info(result)
+
+            if (result.get('code') != 'success'):
+                logger.error(f'翻页回传结果不成功：{result}')
+                self.stop()
+            print(f'任务进度上传 {result}')
+        except Exception as e:
+            logger.error(e)
+
+    def set_flag(self, value):
+        with self._lock:
+            self.end = value

commons/sql_data.py

@@ -0,0 +1,70 @@
+"""
+retrieve_scrape_data 入库语句。
+列顺序须与 DrugPipeline.storge_data 中构建的 row 字段一致。
+"""
+
+# 与 pipelines/drug_pipelines.py 里 row 的键顺序一致（便于核对与扩展）
+RETRIEVE_SCRAPE_INSERT_COLUMNS = (
+    "platform_id",
+    "platform_item_id",
+    "enterprise_id",
+    "product_name",
+    "product_specs",
+    "product_brand",
+    "one_box_price",
+    "link_url",
+    "store_name",
+    "store_url",
+    "shipment_province_id",
+    "shipment_province_name",
+    "shipment_city_id",
+    "shipment_city_name",
+    "manufacturer",
+    "company_name",
+    "scrape_date",
+    "is_sold_out",
+    "min_price",
+    "sales",
+    "inventory",
+    "snapshot_url",
+    "approval_number",
+    "expiry_date",
+    "update_time",
+    "insert_time",
+    "number",
+    "task_id",
+    "anonymous_store_name",
+    "search_name",
+    "collect_config_info",
+    "area_info",
+    "city_name",
+    "city_id",
+    "province_name",
+    "province_id",
+    "collect_equipment_account_id",
+    "collect_region_id",
+    "collect_round",
+)
+
+
+def _retrieve_scrape_insert_if_absent_sql() -> str:
+    cols = RETRIEVE_SCRAPE_INSERT_COLUMNS
+    column_list = ", ".join(f"`{c}`" for c in cols)
+    inner_select = ",\n            ".join(f"%s AS `{c}`" for c in cols)
+    return f"""INSERT INTO `retrieve_scrape_data` ({column_list})
+                SELECT * FROM (
+                    SELECT
+                            {inner_select}
+                ) AS `t`
+                WHERE NOT EXISTS (
+                    SELECT 1 FROM `retrieve_scrape_data` AS `p`
+                    WHERE `p`.`platform_id` <=> %s
+                      AND `p`.`scrape_date` <=> %s
+                      AND `p`.`platform_item_id` <=> %s
+                      AND `p`.`collect_equipment_account_id` <=> %s
+                      AND `p`.`collect_round` <=> %s
+                );"""
+sql_map = {
+    # 仅插入：同 platform_id, scrape_date, platform_item_id, collect_equipment_account_id, collect_round 已存在则跳过
+    "retrieve_scrape_insert_if_absent_sql": _retrieve_scrape_insert_if_absent_sql(),
+}

detail.py

@@ -0,0 +1,77 @@
+import json
+import time
+import copy
+from commons.Logger import logger
+from commons.conn_mysql import MySQLPoolOnline
+from spiders.taobao.taobao_login import (TaobaoAutoCrawl)
+
+
+db_online = MySQLPoolOnline()
+
+sql = "SELECT *,SUBSTRING_INDEX(platform_item_id, '+', 1) AS pre_id FROM retrieve_scrape_data WHERE scrape_date = '2026-06-13' and enterprise_id=11 and platform_item_id LIKE '%+%' group  by pre_id ;"
+
+data_list =db_online.select_data(sql,)
+print(data_list)
+
+taobao_auto = TaobaoAutoCrawl('1', {}, 'n')
+driver = taobao_auto.run()
+for i in data_list:
+
+    url = i.get('link_url')
+    driver.get(url)
+    try:
+        target = driver.eles('xpath://*[@class="content--DIGuLqdf"]')[-1]
+
+        title_list = target.eles('xpath://span')
+        title_list[-1].click()
+        time.sleep(1)
+        new_data = []
+        for j in title_list:
+            j.click()
+            new_price = (driver.eles('xpath://*[@class="block2--MLcO9YdF"]//*[@class="text--LP7Wf49z"]'))[0].text
+            new_url = driver.url
+            new_title = j.text
+
+            data = copy.copy(i)
+            data.update({'product_name': new_title, 'min_price': new_price, 'link_url': new_url})
+            new_data.append(data)
+            time.sleep(2)
+    except Exception as e:
+        logger.exception("页面操作失败, url=%s: %s", url, e)
+        continue
+
+    if not new_data:
+        continue
+
+    pre_id = data.get('pre_id')
+
+    db_online.execute("START TRANSACTION")
+    try:
+        if not pre_id:
+            logger.warning('pre_id 为空，跳过删除')
+        else:
+            delete_sql = "DELETE FROM retrieve_scrape_data WHERE platform_item_id LIKE %s AND scrape_date = %s"
+            db_online.execute(delete_sql, (f'{pre_id}%', '2026-06-13'))
+
+        table_name = "retrieve_scrape_data"
+
+        for product in new_data:
+            del product['pre_id']
+            del product['id']
+            cols = list(product.keys())
+            placeholders = ["%s"] * len(cols)
+
+            sql = f"""
+            INSERT INTO `{table_name}` ({','.join([f'`{c}`' for c in cols])})
+            VALUES ({','.join(placeholders)})
+            """
+            db_online.execute(sql, list(product.values()))
+            logger.info("%s", json.dumps(product, ensure_ascii=False, default=str))
+
+        db_online.execute("COMMIT")
+        logger.info("写入 %d 条, pre_id=%s", len(new_data), pre_id)
+    except Exception as e:
+        db_online.execute("ROLLBACK")
+        logger.exception("事务回滚, pre_id=%s: %s", pre_id, e)
+driver.quit()
+

spiders/taobao/snapshot_taobao_crawl.py

@@ -0,0 +1,627 @@
+import hashlib
+import json
+import random
+import re
+import time
+from decimal import Decimal, InvalidOperation
+from curl_cffi import requests
+from lxml import etree
+from commons.Logger import get_spider_logger
+from commons.conn_mysql import MySQLPoolOnline
+from pipelines.drug_pipelines import DrugPipeline
+from spiders.taobao.snapshot_taobao_login import (TaobaoAutoCrawl)
+from area_info.city_name_to_id import get_city
+from oss_upload.oss_upload import AliyunOSSUploader
+from urllib.parse import quote, unquote
+
+
+logger = get_spider_logger("taobao")
+from urllib.parse import quote
+
+
+
+MTOP_APP_KEY = "12574478"
+MTOP_APP_ID = "34385"
+SEARCH_MAX_PAGE = 20
+REQUEST_RETRY_COUNT = 3
+COOKIE_MAX_AGE_SEC = 3600
+
+
+
+def build_taobao_search_url(keyword: str, page: int = 1) -> str:
+    """
+    构建淘宝搜索URL
+
+    参数:
+        keyword: 搜索关键词 (例如: "999 玉屏风口服液 10支")
+        page: 页码，从1开始
+
+    返回:
+        完整的淘宝搜索URL字符串
+    """
+    # 对关键词进行URL编码（空格转为%20）
+    encoded_keyword = quote(keyword, safe='').replace(' ', '%20')
+
+    # 固定参数
+    fixed_params = {
+        "_input_charset": "utf-8",
+        "commend": "all",
+        "ie": "utf8",
+        "preLoadOrigin": "https://www.taobao.com",
+        "search_type": "item",
+        "source": "suggest",
+        "sourceId": "tb.index",
+        "spm": "a21bo.jianhua/a.search_history.d1",
+        "ssid": "s5-e",
+        "tab": "all",
+        "suggest_query": "",
+    }
+
+    # 动态参数
+    dynamic_params = {
+        "q": encoded_keyword,
+        "page": str(page),
+    }
+
+    # 合并参数
+    all_params = {**fixed_params, **dynamic_params}
+
+    # 构建查询字符串并返回完整URL
+    query_string = "&".join([f"{k}={v}" for k, v in all_params.items()])
+    return f"https://s.taobao.com/search?{query_string}"
+
+def extract_item_data(item_element):
+    """
+    从商品元素中提取数据
+    """
+    result = {
+        "item_id": "",
+        "title": "",
+        "price": "",
+        "realSales": "",
+        "shopInfo": {"title": ""},
+        "procity": "",
+        "auctionURL": ""
+    }
+
+    # 1. 提取 item_id - 从 a 标签的 id 属性
+    a_elem = item_element.ele('xpath=.//a[contains(@id, "item_id_")]')
+    if a_elem:
+        item_id_full = a_elem.attr('id')
+        if item_id_full:
+            result["item_id"] = item_id_full.replace("item_id_", "")
+
+    # 2. 提取 title - 从 div 的 title 属性
+    title_elem = item_element.ele('xpath=.//div[contains(@class, "title--")]')
+    if title_elem:
+        title = title_elem.attr('title')
+        if not title:
+            # 如果没有 title 属性，取文本内容
+            title = title_elem.text
+        result["title"] = title
+
+    # 3. 提取 price - 整数部分 + 小数部分
+    price_int = item_element.ele('xpath=.//div[contains(@class, "priceInt--")]')
+    price_float = item_element.ele('xpath=.//div[contains(@class, "priceFloat--")]')
+    if price_int and price_float:
+        result["price"] = f"{price_int.text}.{price_float.text.replace('.', '')}"
+
+    # 4. 提取 realSales (销量)
+    sales_elem = item_element.ele('xpath=.//span[contains(@class, "realSales--")]')
+    if sales_elem:
+        result["realSales"] = sales_elem.text
+
+    # 5. 提取 shopInfo.title (店铺名称)
+    shop_elem = item_element.ele('xpath=.//span[contains(@class, "shopNameText--")]')
+    if shop_elem:
+        result["shopInfo"]["title"] = shop_elem.text
+
+    # 6. 提取 procity (发货地)
+    procity_elem = item_element.ele('xpath=.//div[contains(@class, "procity--")]/span')
+    if procity_elem:
+        result["procity"] = procity_elem.text
+
+    # 7. 提取 auctionURL
+    if a_elem:
+        href = a_elem.attr('href')
+        if href:
+            result["auctionURL"] = href
+        elif result["item_id"]:
+            result["auctionURL"] = f"https://item.taobao.com/item.htm?id={result['item_id']}"
+
+    return result
+
+
+headers = {
+    "accept": "*/*",
+    "accept-language": "zh-CN,zh;q=0.9",
+    "referer": "https://s.taobao.com/search?page=1&q=999%E6%84%9F%E5%86%92%E7%81%B5&spm=a21bo.jianhua%2Fa.201867-main.d4_first.42f72a89n1ITMs&tab=mall",
+    "sec-ch-ua": '"Not:A-Brand";v="99", "Google Chrome";v="145", "Chromium";v="145"',
+    "sec-ch-ua-mobile": "?0",
+    "sec-ch-ua-platform": '"Windows"',
+    "sec-fetch-dest": "script",
+    "sec-fetch-mode": "no-cors",
+    "sec-fetch-site": "same-site",
+    "user-agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36"
+    ),
+}
+
+MTOP_URL = (
+    "https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/"
+)
+
+
+
+
+
+class TaobaoCrawl:
+    def __init__(self, drug_dict=None):
+        self.cookies = None
+        self.db = MySQLPoolOnline()
+        self.pipeline = DrugPipeline("taobao")
+        self.session = None
+        self.proxies = None
+        self.account_name = None
+        self.ip = None
+        self.cookie_stamp = None
+        self.platform = 1
+        self.task_dict = drug_dict or {}
+        self.collect_task_id = None
+        self.success = True
+        if self.task_dict:
+            self.get_product_data()
+        self.is_no_product = 0
+        self.driver=''
+        self.ossuploader = AliyunOSSUploader()
+
+    def get_product_data(self):
+        self.task_id = self.task_dict["id"]
+        self.company_id = self.task_dict["company_id"]
+        self.product = self.task_dict["product_name"]
+        self.product_desc = self.task_dict.get("product_specs", "")
+        self.brand = self.task_dict.get("product_brand", "")
+        self.product_keyword = self.task_dict.get("product_keyword", "")
+        self.collect_task_id = self.task_dict.get("collect_task_id", "")
+        self.sampling_cycle = self.task_dict.get("sampling_cycle", "")
+        self.sampling_start_time = self.task_dict.get("sampling_start_time", "")
+        self.sampling_end_time = self.task_dict.get("sampling_end_time", "")
+        self.collect_equipment_id = self.task_dict.get("collect_equipment_id", "")
+        self.account_id = self.task_dict.get("collect_equipment_account_id", "")
+        self.collect_region_id = self.task_dict.get("collect_region_id", "")
+        self.collect_round = self.task_dict.get("collect_round", 1)
+
+    @staticmethod
+    def _normalize_url(url):
+        if not url:
+            return ""
+        url = str(url)
+        if url.startswith("//"):
+            return "https:" + url
+        return url
+
+    @staticmethod
+    def _extract_shop_id(shop_url):
+        if not shop_url:
+            return ""
+        shop_re = re.search(r"appUid=(\w+)", shop_url)
+        if shop_re:
+            return shop_re.group(1)
+        return hashlib.md5(shop_url.encode("utf-8")).hexdigest()
+
+    @staticmethod
+    def _sql_literal(value):
+        """避免拼接账号名时单引号打断 SQL（非完整防注入，仅兜底）。"""
+        if value is None:
+            return ""
+        return str(value).replace("'", "''").replace("\\", "\\\\")
+
+    def init_session(self):
+        self.session = requests.Session(impersonate="chrome124")
+        self.session.cookies.update(self.cookies or {})
+        self.session.headers.update(headers)
+        if self.proxies:
+            self.session.proxies.update(self.proxies)
+
+    @staticmethod
+    def _is_transport_error(err):
+        msg = str(err or "")
+        return ("curl: (16)" in msg) or ("Failed to perform" in msg)
+
+    def get_token(self, t, app_key, data_str):
+        _m_h5_tk = (self.cookies or {}).get("_m_h5_tk", "")
+        token = _m_h5_tk.split("_")[0] if _m_h5_tk else ""
+        text = f"{token}&{t}&{app_key}&{data_str}"
+        return hashlib.md5(text.encode()).hexdigest()
+
+    def get_html_content(self, res_html):
+        if not res_html:
+            return ""
+        ele_html = etree.HTML(res_html)
+        if ele_html is None:
+            return str(res_html)
+        text_list = ele_html.xpath(".//text()")
+        return "".join(text_list)
+
+    def _build_search_payload(self, keyword, page, page_size=50):
+        return {
+            "appId": MTOP_APP_ID,
+            "params": {
+                "device": "HMA-AL00",
+                "isBeta": "false",
+                "grayHair": "false",
+                "from": "nt_history",
+                "brand": "HUAWEI",
+                "info": "wifi",
+                "index": "4",
+                "rainbow": "",
+                "schemaType": "auction",
+                "elderHome": "false",
+                "isEnterSrpSearch": "true",
+                "newSearch": "false",
+                "network": "wifi",
+                "subtype": "",
+                "hasPreposeFilter": "false",
+                "prepositionVersion": "v2",
+                "client_os": "Android",
+                "gpsEnabled": "false",
+                "searchDoorFrom": "srp",
+                "debug_rerankNewOpenCard": "false",
+                "homePageVersion": "v7",
+                "searchElderHomeOpen": "false",
+                "search_action": "initiative",
+                "sugg": "_4_1",
+                "sversion": "13.6",
+                "style": "list",
+                "ttid": "600000@taobao_pc_10.7.0",
+                "needTabs": "true",
+                "areaCode": "CN",
+                "vm": "nw",
+                "countryNum": "156",
+                "m": "pc",
+                "page": page,
+                "n": 48,
+                "q": keyword,
+                "qSource": "url",
+                "pageSource": "",
+                "channelSrp": "",
+                "tab": "all",
+                "pageSize": str(page_size),
+                "sourceS": "2",
+                "ntoffset": "0",
+                "filterTag": "",
+                "service": "",
+                "prop": "",
+                "loc": "",
+                "categoryp": "",
+                "screenResolution": "1920x1080",
+                "viewResolution": "1092x4722",
+                "userAgent": headers["user-agent"],
+                "couponUnikey": "",
+                "subTabId": "",
+                "np": "",
+                "clientType": "h5",
+                "isNewDomainAb": "false",
+                "forceOldDomain": "false",
+            },
+        }
+
+    def _request_search_page(self, keyword, page):
+        t = str(int(time.time() * 1000))
+        data = self._build_search_payload(keyword, page)
+        data_str = json.dumps(data, separators=(",", ":"))
+        sign = self.get_token(t, MTOP_APP_KEY, data_str)
+        params = {
+            "jsv": "2.7.4",
+            "appKey": MTOP_APP_KEY,
+            "t": t,
+            "sign": sign,
+            "api": "mtop.relationrecommend.wirelessrecommend.recommend",
+            "v": "2.0",
+            "timeout": "10000",
+            "type": "jsonp",
+            "dataType": "jsonp",
+            "callback": "",
+            "data": data_str,
+        }
+        return self.session.get(MTOP_URL, params=params, timeout=30)
+
+    def _parse_jsonp_body(self, res_text):
+        res_text = (res_text or "").strip()
+        json_str = res_text
+        m = re.match(r"^[^(]*\((.*)\)\s*;?\s*$", res_text, re.DOTALL)
+        if m:
+            json_str = m.group(1)
+        return json.loads(json_str)
+
+    def get_search(self):
+        keyword = self.product
+        if self.brand:
+            keyword = (self.brand + " " + self.product).strip()
+
+        if self.product_desc:
+            keyword = (keyword + " " + self.product_desc).strip()
+        for page in range(1, SEARCH_MAX_PAGE + 1):
+            logger.info(f"正在爬取关键词：{keyword}，{page}页数据")
+
+            # input_box = self.driver.ele('xpath=//*[@id="q"]')
+            # input_box.input(keyword)
+            # time.sleep(1)
+            # button = self.driver.ele('xpath=//*[@id="J_TSearchForm"]/div[2]/button')
+            # button.click()
+            # time.sleep(2)
+            base_url = build_taobao_search_url(keyword)
+            tab = self.driver.latest_tab
+
+            tab.listen.start('https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/')  # 开始监听，指定获取包含该文本的数据包
+            if page==1:
+                tab.get(base_url)
+            else:
+                next_btn = tab.ele('xpath=//button[contains(@aria-label, "下一页")]')
+                if not next_btn.click():
+                    break
+            for _ in range(5):
+                response = tab.listen.wait()  # 等待并获取一个数据包
+                if len(response.url)>3000:
+                    res = response.response.raw_body
+                    break
+                else:
+                    continue
+            time.sleep(1.5)
+            try:
+                json_data = self._parse_jsonp_body(res)
+                item_array = json_data.get("data", {}).get("itemsArray", [])
+            except Exception as e:
+                logger.warning(
+                    "解析数据异常，%s 账号可能退出登录，尝试重新登录: %s",
+                    self.account_name,
+                    e,
+                )
+                break
+            s2 = 0
+            if not item_array:
+                logger.warning("关键词 %s 第 %s 页未获取到商品数据", keyword, page)
+                return
+            elems =  tab.eles('xpath=//*[@id="content_items_wrapper"]/div//div/a/span/../../../..')
+            title_eles = tab.eles('xpath=//*[@id="content_items_wrapper"]/div//div/a/span')
+            for s,raw in enumerate(item_array):
+                try:
+                    item_id = raw.get("item_id", "")
+
+                    if not item_id:
+                        continue
+                    while raw.get('shopInfo').get('title') != title_eles[s2].text:
+                        s2 += 1
+                    item_title = self.get_html_content(raw.get("title") or "")
+
+                    if self.brand not in item_title:
+                        self.is_no_product += 1
+                        continue
+                    if self.product not in item_title:
+                        self.is_no_product += 1
+                        continue
+                    if "+" in item_title:
+                        continue
+
+                    if self.product_desc:
+                        if self.product_desc in item_title:
+                            crawl_product_desc = self.product_desc
+                        else:
+                            crawl_product_desc = ""
+                    else:
+                        crawl_product_desc = ""
+                    self.is_no_product = 0
+                    status = 1
+                    if self.product_keyword:
+                        search_keyword_list = self.product_keyword.split(",")
+                        for search_keyword in search_keyword_list:
+                            if search_keyword.strip() not in item_title:
+                                status = 0
+                    if status == 0:
+                        continue
+
+                    item_price = raw.get("price")
+                    item_price_show = raw.get("priceShow", {}).get("price", 0)
+                    item_sales = raw.get("realSales") or ""
+
+                    sale_num = ""
+                    sales_m = re.search(r"(.*?)人付款", item_sales)
+                    if sales_m:
+                        sale_num = sales_m.group(1)
+
+                    item_url = self._normalize_url(raw.get("auctionURL"))
+
+                    match = re.search(r'skuId=([^&]*)', unquote(item_url))
+
+                    if match:
+                        if match.group(1) != 'null':
+                            item_id = item_id + '_' + match.group(1)
+
+
+
+                    shop_name = raw.get("shopInfo", {}).get("title", "")
+
+                    area_str = (raw.get("procity", "") or "").strip()
+                    city_id, province_id, city, province = get_city(area_str)
+
+                    shop_url = self._normalize_url(
+                        raw.get("shopInfo", {}).get("url", "")
+                    )
+
+                    structured_list = raw.get("structuredUSPInfo",{})
+                    for structured in structured_list:
+                        if structured.get("propertyName","") == "规格":
+                            crawl_product_desc = structured.get("propertyValueName","")
+
+                    pic_path = raw.get("pic_path", "")
+                    raw_price = item_price_show
+                    if raw_price in (None, ""):
+                        price = Decimal("0.00")
+                    else:
+                        try:
+                            price = Decimal(str(raw_price)).quantize(Decimal("0.00"))
+                        except (InvalidOperation, ValueError):
+                            price = Decimal("0.00")
+                    upload_key = hashlib.md5(item_url.encode("utf-8")).hexdigest()
+                    for i in range(3):
+                        try:
+                            jpg_bytes = elems[s2].get_screenshot(as_bytes="jpg")
+                            snapshot_url = self.ossuploader.upload_from_bytes(jpg_bytes, str(upload_key))
+                            break
+                        except Exception as e:
+                            snapshot_url=''
+                            print(e)
+                    s2+=1
+                    scrape_date = time.strftime("%Y-%m-%d")
+                    update_time = time.strftime("%Y-%m-%d %H:%M:%S")
+                    #snapshot_url = self._normalize_url(pic_path) if pic_path else ""
+
+                    # 字段与 yaofangwang_crawl 对齐；键顺序须与 commons.sql_data.RETRIEVE_SCRAPE_INSERT_COLUMNS 一致
+                    product = {
+                        "platform": self.platform,
+                        "item_id": item_id,
+                        "enterprise_id": self.company_id,
+                        "product_name": item_title,
+                        "spec": crawl_product_desc,
+                        "one_price": "",
+                        "detail_url": item_url,
+                        "shop_name": shop_name,
+                        "anonymous_store_name": "",
+                        "shop_url": shop_url,
+                        "city_name": "",
+                        "city_id": "",
+                        "province_name": "",
+                        "province_id": "",
+                        "shipment_city_name": city,
+                        "shipment_city_id": city_id,
+                        "shipment_province_name": province,
+                        "shipment_province_id": province_id,
+                        "area_info": area_str,
+                        "factory_name": "",
+                        "scrape_date": scrape_date,
+                        "price": price,
+                        "sales": sale_num,
+                        "stock_count": "",
+                        "snapshot_url": snapshot_url,
+                        "approval_num": "",
+                        "produced_time": "",
+                        "deadline": "",
+                        "update_time": update_time,
+                        "insert_time": update_time,
+                        "number": 1,
+                        "product_brand": self.brand or "",
+                        "collect_task_id": self.collect_task_id,
+                        "search_name": self.product,
+                        "company_name": "",
+                        "collect_config_info": json.dumps(
+                            {
+                                "sampling_cycle": self.sampling_cycle,
+                                "sampling_start_time": self.sampling_start_time,
+                                "sampling_end_time": self.sampling_end_time,
+                            }
+                        ),
+                        "account_id": self.account_id,
+                        "collect_region_id": self.collect_region_id,
+                        "collect_round": self.collect_round,
+                        "is_sold_out": 0
+
+                    }
+
+                    try:
+                        self.pipeline.storge_data(product)
+                        logger.info("%s", json.dumps(product, ensure_ascii=False, default=str))
+                    except Exception as e:
+                        logger.exception("写入数据库失败: %s", e)
+                except:
+                    continue
+
+
+            logger.info(
+                "关键词 %s 第 %s 页爬取完成",
+                keyword,
+                page,
+            )
+
+            total_page = (
+                (json_data or {}).get("data", {}).get("mainInfo", {}).get("totalPage")
+            )
+
+            try:
+                total_page_int = int(total_page) if total_page is not None else 50
+            except (TypeError, ValueError):
+                total_page_int = 50
+
+            if page >= total_page_int:
+                break
+            if self.is_no_product > 100:
+                break
+            sleep_second = random.uniform(30, 60)
+            logger.info("第 %s 页爬取完成，休息 %.1fs", page, sleep_second)
+            time.sleep(sleep_second)
+
+
+    def update_cookie(self):
+        taobao_auto = TaobaoAutoCrawl(self.account_name, self.ip, self.product)
+        self.driver=taobao_auto.run()
+        if not self.driver :
+            return False
+
+        return True
+
+    def get_account(self):
+
+        sql_account = """
+                      SELECT * FROM `retrieve_collect_equipment_account` WHERE `id` = %s and `status` = 0
+                      """
+
+        account_list = self.db.select_data(sql_account,self.account_id)
+        if not account_list:
+            return False
+
+        account_dict = account_list[0]
+        self.ip = account_dict.get("ip")
+        cookie_str = account_dict.get("cookie_str")
+
+        self.ip = account_dict.get("ip")
+        self.account_name = account_dict.get("username")
+        self.login_username = account_dict.get("phone", "")
+        self.login_password = account_dict.get("password", "")
+        self.cookie_stamp = account_dict.get("update_time")
+        if self.ip:
+            account_proxy = f"http://{self.ip}"
+            self.proxies = {"http": account_proxy, "https": account_proxy}
+        else:
+            self.proxies = None
+
+        need_refresh = (
+                not cookie_str
+                or int(time.time()) - int(self.cookie_stamp or 0) > COOKIE_MAX_AGE_SEC
+        )
+        if 1:
+            if not self.update_cookie():
+                return False
+
+
+        logger.info("获取到账号: %s, ip: %s", self.account_name, self.ip)
+        return True
+
+    def run(self):
+        if not self.get_account():
+            logger.info("==================当前无账号可用==================")
+            self.success = False
+            return self.pipeline.crawl_count, self.success
+
+        logger.info("获取到账号:%s,代理ip:%s", self.account_name, self.ip)
+        try:
+            self.get_search()
+            self.driver.quit()
+        except Exception as e:
+            print('数据获取错误',e)
+        logger.info(
+            "任务id:%s, 任务状态已更新, 产品名称:%s, 爬取数据:%s条",
+            self.task_id,
+            self.product,
+            self.pipeline.crawl_count,
+        )
+        return self.pipeline.crawl_count, self.success

spiders/taobao/snapshot_taobao_crawl2.py

@@ -0,0 +1,652 @@
+import hashlib
+import json
+import random
+import re
+import time
+from decimal import Decimal, InvalidOperation
+from curl_cffi import requests
+from lxml import etree
+from commons.Logger import get_spider_logger
+from commons.conn_mysql import MySQLPoolOn2
+from pipelines.drug_pipelines import DrugPipeline
+
+from spiders.taobao.snapshot_taobao_login import (TaobaoAutoCrawl)
+from area_info.city_name_to_id import get_city
+from oss_upload.oss_upload import AliyunOSSUploader
+from urllib.parse import quote, unquote
+from commons.config import TB_DEVICE_ID
+
+
+logger = get_spider_logger("taobao")
+from urllib.parse import quote
+
+
+
+MTOP_APP_KEY = "12574478"
+MTOP_APP_ID = "34385"
+SEARCH_MAX_PAGE = 20
+REQUEST_RETRY_COUNT = 3
+COOKIE_MAX_AGE_SEC = 3600
+
+
+
+def build_taobao_search_url(keyword: str, page: int = 1) -> str:
+    """
+    构建淘宝搜索URL
+
+    参数:
+        keyword: 搜索关键词 (例如: "999 玉屏风口服液 10支")
+        page: 页码，从1开始
+
+    返回:
+        完整的淘宝搜索URL字符串
+    """
+    # 对关键词进行URL编码（空格转为%20）
+    encoded_keyword = quote(keyword, safe='').replace(' ', '%20')
+
+    # 固定参数
+    fixed_params = {
+        "_input_charset": "utf-8",
+        "commend": "all",
+        "ie": "utf8",
+        "preLoadOrigin": "https://www.taobao.com",
+        "search_type": "item",
+        "source": "suggest",
+        "sourceId": "tb.index",
+        "spm": "a21bo.jianhua/a.search_history.d1",
+        "ssid": "s5-e",
+        "tab": "all",
+        "suggest_query": "",
+    }
+
+    # 动态参数
+    dynamic_params = {
+        "q": encoded_keyword,
+        "page": str(page),
+    }
+
+    # 合并参数
+    all_params = {**fixed_params, **dynamic_params}
+
+    # 构建查询字符串并返回完整URL
+    query_string = "&".join([f"{k}={v}" for k, v in all_params.items()])
+    return f"https://s.taobao.com/search?{query_string}"
+
+def extract_item_data(item_element):
+    """
+    从商品元素中提取数据
+    """
+    result = {
+        "item_id": "",
+        "title": "",
+        "price": "",
+        "realSales": "",
+        "shopInfo": {"title": ""},
+        "procity": "",
+        "auctionURL": ""
+    }
+
+    # 1. 提取 item_id - 从 a 标签的 id 属性
+    a_elem = item_element.ele('xpath=.//a[contains(@id, "item_id_")]')
+    if a_elem:
+        item_id_full = a_elem.attr('id')
+        if item_id_full:
+            result["item_id"] = item_id_full.replace("item_id_", "")
+
+    # 2. 提取 title - 从 div 的 title 属性
+    title_elem = item_element.ele('xpath=.//div[contains(@class, "title--")]')
+    if title_elem:
+        title = title_elem.attr('title')
+        if not title:
+            # 如果没有 title 属性，取文本内容
+            title = title_elem.text
+        result["title"] = title
+
+    # 3. 提取 price - 整数部分 + 小数部分
+    price_int = item_element.ele('xpath=.//div[contains(@class, "priceInt--")]')
+    price_float = item_element.ele('xpath=.//div[contains(@class, "priceFloat--")]')
+    if price_int and price_float:
+        result["price"] = f"{price_int.text}.{price_float.text.replace('.', '')}"
+
+    # 4. 提取 realSales (销量)
+    sales_elem = item_element.ele('xpath=.//span[contains(@class, "realSales--")]')
+    if sales_elem:
+        result["realSales"] = sales_elem.text
+
+    # 5. 提取 shopInfo.title (店铺名称)
+    shop_elem = item_element.ele('xpath=.//span[contains(@class, "shopNameText--")]')
+    if shop_elem:
+        result["shopInfo"]["title"] = shop_elem.text
+
+    # 6. 提取 procity (发货地)
+    procity_elem = item_element.ele('xpath=.//div[contains(@class, "procity--")]/span')
+    if procity_elem:
+        result["procity"] = procity_elem.text
+
+    # 7. 提取 auctionURL
+    if a_elem:
+        href = a_elem.attr('href')
+        if href:
+            result["auctionURL"] = href
+        elif result["item_id"]:
+            result["auctionURL"] = f"https://item.taobao.com/item.htm?id={result['item_id']}"
+
+    return result
+
+
+headers = {
+    "accept": "*/*",
+    "accept-language": "zh-CN,zh;q=0.9",
+    "referer": "https://s.taobao.com/search?page=1&q=999%E6%84%9F%E5%86%92%E7%81%B5&spm=a21bo.jianhua%2Fa.201867-main.d4_first.42f72a89n1ITMs&tab=mall",
+    "sec-ch-ua": '"Not:A-Brand";v="99", "Google Chrome";v="145", "Chromium";v="145"',
+    "sec-ch-ua-mobile": "?0",
+    "sec-ch-ua-platform": '"Windows"',
+    "sec-fetch-dest": "script",
+    "sec-fetch-mode": "no-cors",
+    "sec-fetch-site": "same-site",
+    "user-agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36"
+    ),
+}
+
+MTOP_URL = (
+    "https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/"
+)
+
+
+
+
+
+class TaobaoCrawl:
+    def __init__(self, drug_dict=None, scheduler=None):
+        self.cookies = None
+        self.db = MySQLPoolOn2()
+        self.pipeline = DrugPipeline("taobao")
+        self.session = None
+        self.proxies = None
+        self.account_name = None
+        self.ip = None
+        self.cookie_stamp = None
+        self.platform = 1
+        self.task_dict = drug_dict or {}
+        self.collect_task_id = None
+        self.success = True
+        self.scheduler = scheduler
+        self.report_data = {}
+        if self.task_dict:
+            self.get_product_data()
+        self.is_no_product = 0
+        self.driver=''
+        self.ossuploader = AliyunOSSUploader()
+
+    def get_product_data(self):
+        self.task_id = self.task_dict["id"]
+        self.company_id = self.task_dict["company_id"]
+        self.product = self.task_dict["product_name"]
+        self.product_desc = self.task_dict.get("product_specs", "")
+        self.brand = self.task_dict.get("product_brand", "")
+        self.product_keyword = self.task_dict.get("product_keyword", "")
+        self.collect_task_id = self.task_dict.get("collect_task_id", "")
+        self.sampling_cycle = self.task_dict.get("sampling_cycle", "")
+        self.sampling_start_time = self.task_dict.get("sampling_start_time", "")
+        self.sampling_end_time = self.task_dict.get("sampling_end_time", "")
+        self.collect_equipment_id = self.task_dict.get("collect_equipment_id", "")
+        self.account_id = self.task_dict.get("collect_equipment_account_id", "17")
+        self.collect_region_id = self.task_dict.get("collect_region_id", "")
+        self.collect_round = self.task_dict.get("collect_round", 1)
+        self.report_data = {'task_id': self.task_id, 'platform': self.platform, 'username': self.task_dict.get("username", TB_DEVICE_ID)}
+
+    def _error_report(self, data):
+        self.report_data.update(data)
+        if self.scheduler:
+            self.scheduler.stop()
+            self.scheduler.post_report(self.report_data)
+
+    def _success_report(self, data):
+        self.report_data.update(data)
+        if self.scheduler:
+            self.scheduler.post_report(self.report_data)
+
+    @staticmethod
+    def _normalize_url(url):
+        if not url:
+            return ""
+        url = str(url)
+        if url.startswith("//"):
+            return "https:" + url
+        return url
+
+    @staticmethod
+    def _extract_shop_id(shop_url):
+        if not shop_url:
+            return ""
+        shop_re = re.search(r"appUid=(\w+)", shop_url)
+        if shop_re:
+            return shop_re.group(1)
+        return hashlib.md5(shop_url.encode("utf-8")).hexdigest()
+
+    @staticmethod
+    def _sql_literal(value):
+        """避免拼接账号名时单引号打断 SQL（非完整防注入，仅兜底）。"""
+        if value is None:
+            return ""
+        return str(value).replace("'", "''").replace("\\", "\\\\")
+
+    def init_session(self):
+        self.session = requests.Session(impersonate="chrome124")
+        self.session.cookies.update(self.cookies or {})
+        self.session.headers.update(headers)
+        if self.proxies:
+            self.session.proxies.update(self.proxies)
+
+    @staticmethod
+    def _is_transport_error(err):
+        msg = str(err or "")
+        return ("curl: (16)" in msg) or ("Failed to perform" in msg)
+
+    def get_token(self, t, app_key, data_str):
+        _m_h5_tk = (self.cookies or {}).get("_m_h5_tk", "")
+        token = _m_h5_tk.split("_")[0] if _m_h5_tk else ""
+        text = f"{token}&{t}&{app_key}&{data_str}"
+        return hashlib.md5(text.encode()).hexdigest()
+
+    def get_html_content(self, res_html):
+        if not res_html:
+            return ""
+        ele_html = etree.HTML(res_html)
+        if ele_html is None:
+            return str(res_html)
+        text_list = ele_html.xpath(".//text()")
+        return "".join(text_list)
+
+    def _build_search_payload(self, keyword, page, page_size=50):
+        return {
+            "appId": MTOP_APP_ID,
+            "params": {
+                "device": "HMA-AL00",
+                "isBeta": "false",
+                "grayHair": "false",
+                "from": "nt_history",
+                "brand": "HUAWEI",
+                "info": "wifi",
+                "index": "4",
+                "rainbow": "",
+                "schemaType": "auction",
+                "elderHome": "false",
+                "isEnterSrpSearch": "true",
+                "newSearch": "false",
+                "network": "wifi",
+                "subtype": "",
+                "hasPreposeFilter": "false",
+                "prepositionVersion": "v2",
+                "client_os": "Android",
+                "gpsEnabled": "false",
+                "searchDoorFrom": "srp",
+                "debug_rerankNewOpenCard": "false",
+                "homePageVersion": "v7",
+                "searchElderHomeOpen": "false",
+                "search_action": "initiative",
+                "sugg": "_4_1",
+                "sversion": "13.6",
+                "style": "list",
+                "ttid": "600000@taobao_pc_10.7.0",
+                "needTabs": "true",
+                "areaCode": "CN",
+                "vm": "nw",
+                "countryNum": "156",
+                "m": "pc",
+                "page": page,
+                "n": 48,
+                "q": keyword,
+                "qSource": "url",
+                "pageSource": "",
+                "channelSrp": "",
+                "tab": "all",
+                "pageSize": str(page_size),
+                "sourceS": "2",
+                "ntoffset": "0",
+                "filterTag": "",
+                "service": "",
+                "prop": "",
+                "loc": "",
+                "categoryp": "",
+                "screenResolution": "1920x1080",
+                "viewResolution": "1092x4722",
+                "userAgent": headers["user-agent"],
+                "couponUnikey": "",
+                "subTabId": "",
+                "np": "",
+                "clientType": "h5",
+                "isNewDomainAb": "false",
+                "forceOldDomain": "false",
+            },
+        }
+
+    def _request_search_page(self, keyword, page):
+        t = str(int(time.time() * 1000))
+        data = self._build_search_payload(keyword, page)
+        data_str = json.dumps(data, separators=(",", ":"))
+        sign = self.get_token(t, MTOP_APP_KEY, data_str)
+        params = {
+            "jsv": "2.7.4",
+            "appKey": MTOP_APP_KEY,
+            "t": t,
+            "sign": sign,
+            "api": "mtop.relationrecommend.wirelessrecommend.recommend",
+            "v": "2.0",
+            "timeout": "10000",
+            "type": "jsonp",
+            "dataType": "jsonp",
+            "callback": "",
+            "data": data_str,
+        }
+        return self.session.get(MTOP_URL, params=params, timeout=30)
+
+    def _parse_jsonp_body(self, res_text):
+        res_text = (res_text or "").strip()
+        json_str = res_text
+        m = re.match(r"^[^(]*\((.*)\)\s*;?\s*$", res_text, re.DOTALL)
+        if m:
+            json_str = m.group(1)
+        return json.loads(json_str)
+
+    def get_search(self):
+        keyword = self.product
+        if self.brand:
+            keyword = (self.brand + " " + self.product).strip()
+
+        if self.product_desc:
+            keyword = (keyword + " " + self.product_desc).strip()
+        for page in range(1, SEARCH_MAX_PAGE + 1):
+            logger.info(f"正在爬取关键词：{keyword}，{page}页数据")
+
+            # input_box = self.driver.ele('xpath=//*[@id="q"]')
+            # input_box.input(keyword)
+            # time.sleep(1)
+            # button = self.driver.ele('xpath=//*[@id="J_TSearchForm"]/div[2]/button')
+            # button.click()
+            # time.sleep(2)
+            base_url = build_taobao_search_url(keyword)
+            tab = self.driver.latest_tab
+
+            tab.listen.start('https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/')  # 开始监听，指定获取包含该文本的数据包
+            if page==1:
+                tab.get(base_url)
+            else:
+                next_btn = tab.ele('xpath=//*[@id="search-content-leftWrap"]//button[contains(@aria-label, "下一页")]')
+                if not next_btn.click():
+                    break
+            for _ in range(5):
+                response = tab.listen.wait()  # 等待并获取一个数据包
+                if len(response.url)>3000:
+                    res = response.response.raw_body
+                    break
+                else:
+                    continue
+            time.sleep(1.5)
+            try:
+                json_data = self._parse_jsonp_body(res)
+                item_array = json_data.get("data", {}).get("itemsArray", [])
+            except Exception as e:
+                logger.warning(
+                    "解析数据异常，%s 账号可能退出登录，尝试重新登录: %s",
+                    self.account_name,
+                    e,
+                )
+                break
+            s2 = 0
+            if not item_array:
+                logger.warning("关键词 %s 第 %s 页未获取到商品数据", keyword, page)
+                return
+            elems =  tab.eles('xpath=//*[@id="content_items_wrapper"]/div//div/a/span/../../../..')
+            title_eles = tab.eles('xpath=//*[@id="content_items_wrapper"]/div//div/a/span')
+            for s,raw in enumerate(item_array):
+                try:
+                    item_id = raw.get("item_id", "")
+
+                    if not item_id:
+                        continue
+                    while raw.get('shopInfo').get('title') != title_eles[s2].text:
+                        s2 += 1
+                    item_title = self.get_html_content(raw.get("title") or "")
+
+                    if self.brand not in item_title:
+                        self.is_no_product += 1
+                        continue
+                    if self.product not in item_title:
+                        self.is_no_product += 1
+                        continue
+                    if "+" in item_title:
+                        continue
+
+                    if self.product_desc:
+                        if self.product_desc in item_title:
+                            crawl_product_desc = self.product_desc
+                        else:
+                            crawl_product_desc = ""
+                    else:
+                        crawl_product_desc = ""
+                    self.is_no_product = 0
+                    status = 1
+                    if self.product_keyword:
+                        search_keyword_list = self.product_keyword.split(",")
+                        for search_keyword in search_keyword_list:
+                            if search_keyword.strip() not in item_title:
+                                status = 0
+                    if status == 0:
+                        continue
+
+                    item_price = raw.get("price")
+                    item_price_show = raw.get("priceShow", {}).get("price", 0)
+                    item_sales = raw.get("realSales") or ""
+
+                    sale_num = ""
+                    sales_m = re.search(r"(.*?)人付款", item_sales)
+                    if sales_m:
+                        sale_num = sales_m.group(1)
+
+                    item_url = self._normalize_url(raw.get("auctionURL"))
+
+                    match = re.search(r'skuId=([^&]*)', unquote(item_url))
+
+                    if match:
+                        if match.group(1) != 'null':
+                            item_id = item_id + '_' + match.group(1)
+
+
+
+                    shop_name = raw.get("shopInfo", {}).get("title", "")
+
+                    area_str = (raw.get("procity", "") or "").strip()
+                    city_id, province_id, city, province = get_city(area_str)
+
+                    shop_url = self._normalize_url(
+                        raw.get("shopInfo", {}).get("url", "")
+                    )
+
+                    structured_list = raw.get("structuredUSPInfo",{})
+                    for structured in structured_list:
+                        if structured.get("propertyName","") == "规格":
+                            crawl_product_desc = structured.get("propertyValueName","")
+
+                    pic_path = raw.get("pic_path", "")
+                    raw_price = item_price_show
+                    if raw_price in (None, ""):
+                        price = Decimal("0.00")
+                    else:
+                        try:
+                            price = Decimal(str(raw_price)).quantize(Decimal("0.00"))
+                        except (InvalidOperation, ValueError):
+                            price = Decimal("0.00")
+                    upload_key = hashlib.md5(item_url.encode("utf-8")).hexdigest()
+                    for i in range(3):
+                        try:
+                            jpg_bytes = elems[s2].get_screenshot(as_bytes="jpg")
+                            snapshot_url = self.ossuploader.upload_from_bytes(jpg_bytes, str(upload_key))
+                            break
+                        except Exception as e:
+                            snapshot_url=''
+                            print(e)
+                    s2+=1
+                    scrape_date = time.strftime("%Y-%m-%d")
+                    update_time = time.strftime("%Y-%m-%d %H:%M:%S")
+                    #snapshot_url = self._normalize_url(pic_path) if pic_path else ""
+
+                    # 字段与 yaofangwang_crawl 对齐；键顺序须与 commons.sql_data.RETRIEVE_SCRAPE_INSERT_COLUMNS 一致
+                    product = {
+                        "platform": self.platform,
+                        "item_id": item_id,
+                        "enterprise_id": self.company_id,
+                        "product_name": item_title,
+                        "spec": crawl_product_desc,
+                        "one_price": "",
+                        "detail_url": item_url,
+                        "shop_name": shop_name,
+                        "anonymous_store_name": "",
+                        "shop_url": shop_url,
+                        "city_name": "",
+                        "city_id": "",
+                        "province_name": "",
+                        "province_id": "",
+                        "shipment_city_name": city,
+                        "shipment_city_id": city_id,
+                        "shipment_province_name": province,
+                        "shipment_province_id": province_id,
+                        "area_info": area_str,
+                        "factory_name": "",
+                        "scrape_date": scrape_date,
+                        "price": price,
+                        "sales": sale_num,
+                        "stock_count": "",
+                        "snapshot_url": snapshot_url,
+                        "approval_num": "",
+                        "produced_time": "",
+                        "deadline": "",
+                        "update_time": update_time,
+                        "insert_time": update_time,
+                        "number": 1,
+                        "product_brand": self.brand or "",
+                        "collect_task_id": self.collect_task_id,
+                        "search_name": self.product,
+                        "company_name": "",
+                        "collect_config_info": json.dumps(
+                            {
+                                "sampling_cycle": self.sampling_cycle,
+                                "sampling_start_time": self.sampling_start_time,
+                                "sampling_end_time": self.sampling_end_time,
+                            }
+                        ),
+                        "account_id": self.account_id,
+                        "collect_region_id": self.collect_region_id,
+                        "collect_round": self.collect_round,
+                        "is_sold_out": 0
+
+                    }
+
+                    try:
+                        self.pipeline.storge_data(product)
+                        logger.info("%s", json.dumps(product, ensure_ascii=False, default=str))
+                    except Exception as e:
+                        logger.exception("写入数据库失败: %s", e)
+                except:
+                    continue
+
+
+            logger.info(
+                "关键词 %s 第 %s 页爬取完成",
+                keyword,
+                page,
+            )
+
+            total_page = (
+                (json_data or {}).get("data", {}).get("mainInfo", {}).get("totalPage")
+            )
+
+            try:
+                total_page_int = int(total_page) if total_page is not None else 50
+            except (TypeError, ValueError):
+                total_page_int = 50
+            self.report_data.update({'total_pages': total_page_int})
+
+            if self.scheduler and self.scheduler.end:
+                logger.info('心跳失败')
+                self._error_report({'is_finished': 0, 'need_reassign': 0, 'current_page': page})
+                break
+
+            if page >= total_page_int:
+                self._success_report({'is_finished': 1, 'need_reassign': 0, 'current_page': page})
+                break
+            if self.is_no_product > 100:
+                self._success_report({'is_finished': 1, 'need_reassign': 0, 'current_page': page})
+                break
+            sleep_second = random.uniform(30, 60)
+            logger.info("第 %s 页爬取完成，休息 %.1fs", page, sleep_second)
+            self._success_report({'is_finished': 0, 'need_reassign': 0, 'current_page': page})
+            time.sleep(sleep_second)
+
+
+    def update_cookie(self):
+        taobao_auto = TaobaoAutoCrawl(self.account_name, self.ip, self.product)
+        self.driver=taobao_auto.run()
+        if not self.driver :
+            return False
+
+        return True
+
+    def get_account(self):
+
+        sql_account = """
+                      SELECT * FROM `retrieve_collect_equipment_account` WHERE `id` = %s and `status` = 0
+                      """
+
+        account_list = self.db.select_data(sql_account,self.account_id)
+        if not account_list:
+            return False
+
+        account_dict = account_list[0]
+        self.ip = account_dict.get("ip")
+        cookie_str = account_dict.get("cookie_str")
+
+        self.ip = account_dict.get("ip")
+        self.account_name = account_dict.get("username")
+        self.login_username = account_dict.get("phone", "")
+        self.login_password = account_dict.get("password", "")
+        self.cookie_stamp = account_dict.get("update_time")
+        if self.ip:
+            account_proxy = f"http://{self.ip}"
+            self.proxies = {"http": account_proxy, "https": account_proxy}
+        else:
+            self.proxies = None
+
+        need_refresh = (
+                not cookie_str
+                or int(time.time()) - int(self.cookie_stamp or 0) > COOKIE_MAX_AGE_SEC
+        )
+        if 1:
+            if not self.update_cookie():
+                return False
+
+
+        logger.info("获取到账号: %s, ip: %s", self.account_name, self.ip)
+        return True
+
+    def run(self):
+        if not self.get_account():
+            logger.info("==================当前无账号可用==================")
+            self.success = False
+            return self.pipeline.crawl_count, self.success
+
+        logger.info("获取到账号:%s,代理ip:%s", self.account_name, self.ip)
+        try:
+            self.get_search()
+            self.driver.quit()
+        except Exception as e:
+            print('数据获取错误',e)
+        logger.info(
+            "任务id:%s, 任务状态已更新, 产品名称:%s, 爬取数据:%s条",
+            self.task_id,
+            self.product,
+            self.pipeline.crawl_count,
+        )
+        return self.pipeline.crawl_count, self.success

spiders/taobao/snapshot_taobao_login.py

@@ -0,0 +1,194 @@
+import time
+import json
+import random
+import signal
+import sys
+import DrissionPage
+from DrissionPage import ChromiumPage, ChromiumOptions
+import re
+import socket
+from commons.conn_mysql import MySQLPoolOnline
+import hashlib
+from commons.Logger import logger
+
+MAX_PAGES = 5
+WAIT_BETWEEN_PAGES = (8, 15)  # 页间等待时间范围（秒）
+SCROLL_DELAY = (0.3, 0.8)  # 滚动延迟范围
+CLICK_DELAY = (0.5, 1.2)  # 点击延迟范围
+BROWSE_TIME = (5, 10)  # 浏览时间范围
+chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
+
+
+class TaobaoAutoCrawl:
+    def __init__(self, account_name, ip, key_word):
+        self.driver = None
+        self.register_signal_handler()
+        self.db = MySQLPoolOnline()
+        self.account_name = account_name
+        self.ip = ip
+        self.keyword = key_word
+
+    @staticmethod
+    def _get_free_port():
+        """获取一个当前可用的本地端口，供 Chrome 调试使用。"""
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.bind(("127.0.0.1", 0))
+            return s.getsockname()[1]
+
+    def init_drissionpage(self):
+        # 避免 auto_port 在部分环境下生成异常地址（无端口）导致 ChromiumPage 初始化失败
+        co = ChromiumOptions().set_browser_path(chrome_path)
+        #debug_port = self._get_free_port()
+        #co.set_argument(f"--remote-debugging-port={debug_port}")
+
+        co.set_user_data_path(f"./spider/taobao/{'TB_103'}")
+        if self.ip:
+            proxy = self.ip.strip()
+            if not proxy.startswith(("http://", "https://")):
+                proxy = f"http://{proxy}"
+            co.set_argument(f"--proxy-server={proxy}")
+        self.driver = ChromiumPage(co)
+
+    def register_signal_handler(self):
+        """ 非常必要，注册信号处理，确保状态保存"""
+
+        def signal_handler(signum, frame):
+            print("\n⚠️ 收到退出信号，正在保存状态并退出...")
+            if self.driver:
+                self.driver.quit()
+            sys.exit(0)
+
+        signal.signal(signal.SIGINT, signal_handler)
+        # Windows 上可能不支持 SIGTERM，做兼容处理
+        if hasattr(signal, "SIGTERM"):
+            signal.signal(signal.SIGTERM, signal_handler)
+
+    # ==================== 人工行为模拟 ====================
+    def random_wait(self, min_sec, max_sec=None):
+        """随机等待"""
+        if max_sec is None:
+            max_sec = min_sec
+        time.sleep(random.uniform(min_sec, max_sec))
+
+    def move_mouse_to_element(self, element):
+        """移动鼠标到元素"""
+        if not element:
+            return
+        try:
+            # 优先使用 DrissionPage 推荐方式
+            self.driver.actions.move_to(element)
+        except Exception:
+            # 兼容旧逻辑：按元素中心点移动（不同版本 move 参数可能不同）
+            box = element.rect
+            try:
+                center_x = int(box.x + box.width / 2)
+                center_y = int(box.y + box.height / 2)
+            except Exception:
+                center_x = int(box["x"] + box["width"] / 2)
+                center_y = int(box["y"] + box["height"] / 2)
+            try:
+                self.driver.actions.move(center_x, center_y)
+            except TypeError:
+                # 某些版本仅支持关键字参数
+                self.driver.actions.move(offset_x=center_x, offset_y=center_y)
+        self.random_wait(0.2, 0.5)
+
+    def human_type(self, element, text):
+        """模拟人类输入"""
+        for char in text:
+            element.send_keys(char)
+            time.sleep(random.uniform(0.1, 0.3))
+
+    def login(self, username, password):
+        self.driver.get("https://login.taobao.com")
+        self.random_wait(5, 8)
+
+        # 输入账号
+        login_name = self.driver.ele("xpath=//input[@name='fm-login-id']", timeout=30)
+        if login_name:
+            self.move_mouse_to_element(login_name)
+            self.human_type(login_name, username)
+            self.random_wait(1, 3)
+
+        # 输入密码
+        login_pass = self.driver.ele("xpath=//input[@name='fm-login-password']", timeout=30)
+        if login_pass:
+            self.move_mouse_to_element(login_pass)
+            self.human_type(login_pass, password)
+            self.random_wait(1, 3)
+
+        # 点击登录
+        login_button = self.driver.ele("xpath=//button[text()='登录']", timeout=30)
+        if login_button:
+            self.move_mouse_to_element(login_button)
+            login_button.click()
+            self.random_wait(1, 3)
+
+        # 处理同意按钮
+        login_agree = self.driver.ele("xpath=//button[text()='同意']", timeout=5)
+        if login_agree:
+            self.move_mouse_to_element(login_agree)
+            login_agree.click()
+            self.random_wait(1, 3)
+
+        # 等待登录结果
+        self.random_wait(10, 20)
+        # 检查是否登录成功
+        user_info = self.driver.ele("xpath=//a[@class='site-nav-login-info-nick']", timeout=10)
+        if user_info:
+            print("登录成功！")
+        else:
+            print("登录失败，请检查账号密码或验证码")
+
+    def get_search(self):
+        url = "https://www.taobao.com"
+        self.driver.get(url, timeout=30)
+        time.sleep(20)
+        time.sleep(random.uniform(3, 8))
+
+        # 刷新一次，否则可能未找到登录状态
+        self.driver.refresh()
+        self.random_wait(5, 10)
+        # login_name = self.driver.ele("xpath=//input[@name='fm-login-id']")
+        # if login_name:
+        #     self.login("aqwwer","wewetrv")
+        #     self.driver.refresh()
+        #     self.random_wait(5, 10)
+
+        ele_iframe = self.driver.ele("xpath=//iframe[@id='baxia-dialog-content']")
+
+
+        ele = self.driver.ele('xpath=//*[contains(@class,"site-nav-login-info-nick")]', timeout=30)
+
+        if ele:
+            cookies_list = self.driver.cookies()
+            cookies_dict = {c['name']: c['value'] for c in cookies_list}
+            timestamp = int(time.time())
+            # 保存 cookie 到文件
+            print(f"{self.account_name},获取 cookie 成功！")
+            logger.info(f"{self.account_name},获取 cookie 成功！")
+
+            self.random_wait(3, 5)
+            return True
+        else:
+            return False
+
+    def run(self):
+        bool_login = False
+        try:
+            self.init_drissionpage()
+            bool_login = self.get_search()
+        except Exception as e:
+            logger.exception(f"{self.account_name} 获取 cookie 异常: {e}")
+        finally:
+            if self.driver:
+                return self.driver
+        return bool(bool_login)
+
+
+if __name__ == '__main__':
+    account_name = "tb_account10"
+    ip = ""
+    keyword = "手机"
+    taobao_crawl = TaobaoAutoCrawl(account_name, ip, keyword)
+    taobao_crawl.run()

spiders/taobao/taobao_crawl.py

@@ -0,0 +1,596 @@
+import hashlib
+import json
+import random
+import re
+import time
+from decimal import Decimal, InvalidOperation
+from curl_cffi import requests
+from lxml import etree
+from commons.Logger import get_spider_logger
+from commons.conn_mysql import MySQLPoolOn2
+from pipelines.drug_pipelines import DrugPipeline
+from spiders.taobao.taobao_login import (TaobaoAutoCrawl)
+from area_info.city_name_to_id import get_city
+from urllib.parse import quote, unquote
+logger = get_spider_logger("taobao")
+
+MTOP_APP_KEY = "12574478"
+MTOP_APP_ID = "34385"
+SEARCH_MAX_PAGE = 20
+REQUEST_RETRY_COUNT = 3
+COOKIE_MAX_AGE_SEC = 3600
+
+headers = {
+    "accept": "*/*",
+    "accept-language": "zh-CN,zh;q=0.9",
+    "referer": "https://s.taobao.com/search?page=1&q=999%E6%84%9F%E5%86%92%E7%81%B5&spm=a21bo.jianhua%2Fa.201867-main.d4_first.42f72a89n1ITMs&tab=mall",
+    "sec-ch-ua": '"Not:A-Brand";v="99", "Google Chrome";v="145", "Chromium";v="145"',
+    "sec-ch-ua-mobile": "?0",
+    "sec-ch-ua-platform": '"Windows"',
+    "sec-fetch-dest": "script",
+    "sec-fetch-mode": "no-cors",
+    "sec-fetch-site": "same-site",
+    "user-agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36"
+    ),
+}
+
+MTOP_URL = (
+    "https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/"
+)
+
+
+def get_product_name(item):
+    """递归查找 item 中 propertyName='产品名称' 的 propertyValueName"""
+    if isinstance(item, dict):
+        if item.get('propertyName') == '产品名称':
+            return item.get('propertyValueName', '')
+        for v in item.values():
+            result = get_product_name(v)
+            if result:
+                return result
+    elif isinstance(item, list):
+        for sub in item:
+            result = get_product_name(sub)
+            if result:
+                return result
+    return ''
+
+
+class TaobaoCrawl:
+    def __init__(self, drug_dict=None,scheduler = None):
+        self.cookies = None
+        self.db = MySQLPoolOn2()
+        self.pipeline = DrugPipeline("taobao")
+        self.session = None
+        self.proxies = None
+        self.account_name = None
+        self.ip = None
+        self.cookie_stamp = None
+        self.platform = 1
+        self.task_dict = drug_dict or {}
+        self.collect_task_id = None
+        self.success = True
+        self.scheduler = scheduler 
+        if self.task_dict:
+            self.get_product_data()
+        self.is_no_product = 0
+
+    def get_product_data(self):
+        self.task_id = self.task_dict["id"]
+        self.company_id = self.task_dict["company_id"]
+        self.product = self.task_dict["product_name"]
+        self.product_desc = self.task_dict.get("product_specs", "")
+        self.brand = self.task_dict.get("product_brand", "")
+        self.product_keyword = self.task_dict.get("product_keyword", "")
+        self.collect_task_id = self.task_dict.get("collect_task_id", "")
+        self.sampling_cycle = self.task_dict.get("sampling_cycle", "")
+        self.sampling_start_time = self.task_dict.get("sampling_start_time", "")
+        self.sampling_end_time = self.task_dict.get("sampling_end_time", "")
+        self.collect_equipment_id = self.task_dict.get("collect_equipment_id", "")
+        self.account_id = self.task_dict.get("collect_equipment_account_id", "")
+        self.collect_region_id = self.task_dict.get("collect_region_id", "")
+        self.collect_round = self.task_dict.get("collect_round", 1)
+        self.start_page = self.task_dict.get(self.task_dict.get("current_page")+1, 1)
+        self.report_data = {'task_id':self.task_id,'platform':self.platform,'username':self.scheduler.username}
+
+    def _error_report(self,data):
+            self.report_data.update(data)
+            self.scheduler.stop()
+            self.scheduler.post_report(self.report_data)
+
+    def _success_report(self,data):
+            self.report_data.update(data)
+            self.scheduler.post_report(self.report_data)
+
+    @staticmethod
+    def _normalize_url(url):
+        if not url:
+            return ""
+        url = str(url)
+        if url.startswith("//"):
+            return "https:" + url
+        return url
+
+    @staticmethod
+    def _extract_shop_id(shop_url):
+        if not shop_url:
+            return ""
+        shop_re = re.search(r"appUid=(\w+)", shop_url)
+        if shop_re:
+            return shop_re.group(1)
+        return hashlib.md5(shop_url.encode("utf-8")).hexdigest()
+
+    @staticmethod
+    def _sql_literal(value):
+        """避免拼接账号名时单引号打断 SQL（非完整防注入，仅兜底）。"""
+        if value is None:
+            return ""
+        return str(value).replace("'", "''").replace("\\", "\\\\")
+
+    def init_session(self):
+        self.session = requests.Session(impersonate="chrome124")
+        self.session.cookies.update(self.cookies or {})
+        self.session.headers.update(headers)
+        if self.proxies:
+            self.session.proxies.update(self.proxies)
+
+    def init_session_lose(self,res,keyword,page):
+        if '哎哟喂,被挤爆啦' in res.text:
+            if not self.update_cookie():
+                return False
+
+            res = self._request_search_page(keyword, page)
+            if '哎哟喂,被挤爆啦' in res.text:
+                return False
+
+        return True
+
+
+    @staticmethod
+    def _is_transport_error(err):
+        msg = str(err or "")
+        return ("curl: (16)" in msg) or ("Failed to perform" in msg)
+
+    def get_token(self, t, app_key, data_str):
+        _m_h5_tk = (self.cookies or {}).get("_m_h5_tk", "")
+        token = _m_h5_tk.split("_")[0] if _m_h5_tk else ""
+        text = f"{token}&{t}&{app_key}&{data_str}"
+        return hashlib.md5(text.encode()).hexdigest()
+
+    def get_html_content(self, res_html):
+        if not res_html:
+            return ""
+        ele_html = etree.HTML(res_html)
+        if ele_html is None:
+            return str(res_html)
+        text_list = ele_html.xpath(".//text()")
+        return "".join(text_list)
+
+    def _build_search_payload(self, keyword, page, page_size=50):
+        return {
+            "appId": MTOP_APP_ID,
+            "params": {
+                "device": "HMA-AL00",
+                "isBeta": "false",
+                "grayHair": "false",
+                "from": "nt_history",
+                "brand": "HUAWEI",
+                "info": "wifi",
+                "index": "4",
+                "rainbow": "",
+                "schemaType": "auction",
+                "elderHome": "false",
+                "isEnterSrpSearch": "true",
+                "newSearch": "false",
+                "network": "wifi",
+                "subtype": "",
+                "hasPreposeFilter": "false",
+                "prepositionVersion": "v2",
+                "client_os": "Android",
+                "gpsEnabled": "false",
+                "searchDoorFrom": "srp",
+                "debug_rerankNewOpenCard": "false",
+                "homePageVersion": "v7",
+                "searchElderHomeOpen": "false",
+                "search_action": "initiative",
+                "sugg": "_4_1",
+                "sversion": "13.6",
+                "style": "list",
+                "ttid": "600000@taobao_pc_10.7.0",
+                "needTabs": "true",
+                "areaCode": "CN",
+                "vm": "nw",
+                "countryNum": "156",
+                "m": "pc",
+                "page": page,
+                "n": 48,
+                "q": keyword,
+                "qSource": "url",
+                "pageSource": "",
+                "channelSrp": "",
+                "tab": "all",
+                "pageSize": str(page_size),
+                "sourceS": "2",
+                "ntoffset": "0",
+                "filterTag": "",
+                "service": "",
+                "prop": "",
+                "loc": "",
+                "categoryp": "",
+                "screenResolution": "1920x1080",
+                "viewResolution": "1092x4722",
+                "userAgent": headers["user-agent"],
+                "couponUnikey": "",
+                "subTabId": "",
+                "np": "",
+                "clientType": "h5",
+                "isNewDomainAb": "false",
+                "forceOldDomain": "false",
+            },
+        }
+
+    def _request_search_page(self, keyword, page):
+        t = str(int(time.time() * 1000))
+        data = self._build_search_payload(keyword, page)
+        data_str = json.dumps(data, separators=(",", ":"))
+        sign = self.get_token(t, MTOP_APP_KEY, data_str)
+        params = {
+            "jsv": "2.7.4",
+            "appKey": MTOP_APP_KEY,
+            "t": t,
+            "sign": sign,
+            "api": "mtop.relationrecommend.wirelessrecommend.recommend",
+            "v": "2.0",
+            "timeout": "10000",
+            "type": "jsonp",
+            "dataType": "jsonp",
+            "callback": "",
+            "data": data_str,
+        }
+        return self.session.get(MTOP_URL, params=params, timeout=30)
+
+    def _parse_jsonp_body(self, res_text):
+        res_text = (res_text or "").strip()
+        json_str = res_text
+        m = re.match(r"^[^(]*\((.*)\)\s*;?\s*$", res_text, re.DOTALL)
+        if m:
+            json_str = m.group(1)
+        return json.loads(json_str)
+
+    def get_search(self):
+        keyword = self.product
+        if self.brand:
+            keyword = (self.brand + " " + self.product).strip()
+
+        if self.product_desc:
+            keyword = (keyword + " " + self.product_desc).strip()
+
+        for page in range(self.start_page,30):
+            logger.info(f"正在爬取关键词：{keyword}，{page}页数据")
+            for attempt in range(1, REQUEST_RETRY_COUNT + 1):
+                try:
+                    if not self.session:
+                        self.init_session()
+                    res = self._request_search_page(keyword, page)
+                    if not self.init_session_lose(res,keyword,page):
+                        #这里要警告，属于账号异常，账号即使刷新还是不恢复
+                        self.success=False
+                        self._error_report({'is_finished':0,'need_reassign':1,'current_page':self.start_page,'exception_type':2})
+                        return
+
+                    if res.status_code != 200:
+                        logger.warning(
+                            "请求失败 HTTP %s，第%s/%s 次重试",
+                            res.status_code,
+                            attempt,
+                            REQUEST_RETRY_COUNT,
+                        )
+                        time.sleep(random.randint(3, 8))
+                        continue
+                except Exception as e:
+                    if self._is_transport_error(e):
+                        logger.warning(
+                            "检测到网络传输异常(curl)，重建会话后重试: %s",
+                            e,
+                        )
+                        self.init_session()
+                    logger.warning(
+                        "请求异常，第%s/%s 次重试: %s",
+                        attempt,
+                        REQUEST_RETRY_COUNT,
+                        e,
+                    )
+                    # 指数退避，避免连续瞬时失败
+                    time.sleep(min(3 * attempt, 10))
+                    continue
+
+                try:
+                    json_data = self._parse_jsonp_body(res.text)
+                    item_array = json_data.get("data", {}).get("itemsArray", [])
+                    break
+                except Exception as e:
+                    logger.warning(
+                        "解析数据异常，%s 账号可能退出登录，尝试重新登录: %s",
+                        self.account_name,
+                        e,
+                    )
+                    if self.update_cookie():
+                        self.init_session()
+                    else:
+                        self._error_report({'is_finished':0,'need_reassign':1,'current_page':page,'exception_type':2})
+                        return
+
+            else:
+                logger.warning("关键词 %s 第 %s 页连续重试失败", keyword, page)
+                continue
+
+            if not item_array:
+                logger.warning("关键词 %s 第 %s 页未获取到商品数据", keyword, page)
+                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page})
+                return
+
+            for raw in item_array:
+                try:
+                    item_id = raw.get("item_id", "")
+                except:
+                    break
+                if not item_id:
+                    continue
+
+                item_title = self.get_html_content(raw.get("title") or "")
+                try:
+                    propertyValueName =get_product_name(raw)
+                except:
+                    propertyValueName = ''
+
+                if self.brand not in item_title:
+                    self.is_no_product += 1
+                    continue
+                if propertyValueName:
+                    if self.product not in propertyValueName:
+                        self.is_no_product += 1
+                        continue
+                else:
+                    if self.product not in item_title:
+                        self.is_no_product += 1
+                        continue
+                if "+" in item_title:
+                    continue
+
+                if self.product_desc:
+                    if self.product_desc in item_title:
+                        crawl_product_desc = self.product_desc
+                    else:
+                        crawl_product_desc = ""
+                else:
+                    crawl_product_desc = ""
+
+                self.is_no_product = 0
+                status = 1
+                if self.product_keyword:
+                    search_keyword_list = self.product_keyword.split(",")
+                    for search_keyword in search_keyword_list:
+                        if search_keyword.strip() not in item_title:
+                            status = 0
+                if status == 0:
+                    continue
+
+                item_price = raw.get("price")
+                item_price_show = raw.get("priceShow", {}).get("price", 0)
+                item_sales = raw.get("realSales") or ""
+
+                sale_num = ""
+                sales_m = re.search(r"(.*?)人付款", item_sales)
+                if sales_m:
+                    sale_num = sales_m.group(1)
+
+                item_url = self._normalize_url(raw.get("auctionURL"))
+                match = re.search(r'skuId=([^&]*)', unquote(item_url))
+
+                if match:
+                    if match.group(1)!='null':
+                        item_id = item_id + '_' + match.group(1)
+
+
+
+                shop_name = raw.get("shopInfo", {}).get("title", "")
+
+                area_str = (raw.get("procity", "") or "").strip()
+                city_id, province_id, city, province = get_city(area_str)
+
+                shop_url = self._normalize_url(
+                    raw.get("shopInfo", {}).get("url", "")
+                )
+
+                structured_list = raw.get("structuredUSPInfo",{})
+                for structured in structured_list:
+                    if structured.get("propertyName","") == "规格":
+                        crawl_product_desc = structured.get("propertyValueName","")
+
+                pic_path = raw.get("pic_path", "")
+                raw_price = item_price_show
+                if raw_price in (None, ""):
+                    price = Decimal("0.00")
+                else:
+                    try:
+                        price = Decimal(str(raw_price)).quantize(Decimal("0.00"))
+                    except (InvalidOperation, ValueError):
+                        price = Decimal("0.00")
+
+                scrape_date = time.strftime("%Y-%m-%d")
+                update_time = time.strftime("%Y-%m-%d %H:%M:%S")
+                snapshot_url = self._normalize_url(pic_path) if pic_path else ""
+
+                # 字段与 yaofangwang_crawl 对齐；键顺序须与 commons.sql_data.RETRIEVE_SCRAPE_INSERT_COLUMNS 一致
+                product = {
+                    "platform": self.platform,
+                    "item_id": item_id,
+                    "enterprise_id": self.company_id,
+                    "product_name": item_title,
+                    "spec": crawl_product_desc,
+                    "one_price": "",
+                    "detail_url": item_url,
+                    "shop_name": shop_name,
+                    "anonymous_store_name": "",
+                    "shop_url": shop_url,
+                    "city_name": "",
+                    "city_id": "",
+                    "province_name": "",
+                    "province_id": "",
+                    "shipment_city_name": city,
+                    "shipment_city_id": city_id,
+                    "shipment_province_name": province,
+                    "shipment_province_id": province_id,
+                    "area_info": area_str,
+                    "factory_name": "",
+                    "scrape_date": scrape_date,
+                    "price": price,
+                    "sales": sale_num,
+                    "stock_count": "",
+                    "snapshot_url": "",
+                    "approval_num": "",
+                    "produced_time": "",
+                    "deadline": "",
+                    "update_time": update_time,
+                    "insert_time": update_time,
+                    "number": 1,
+                    "product_brand": self.brand or "",
+                    "collect_task_id": self.collect_task_id,
+                    "search_name": self.product,
+                    "company_name": "",
+                    "collect_config_info": json.dumps(
+                        {
+                            "sampling_cycle": self.sampling_cycle,
+                            "sampling_start_time": self.sampling_start_time,
+                            "sampling_end_time": self.sampling_end_time,
+                        }
+                    ),
+                    "account_id": self.account_id,
+                    "collect_region_id": self.collect_region_id,
+                    "collect_round": self.collect_round,
+                    "is_sold_out": 0
+
+                }
+
+                try:
+                    self.pipeline.storge_data(product)
+                    logger.info("%s", json.dumps(product, ensure_ascii=False, default=str))
+                except Exception as e:
+                    logger.exception("写入数据库失败: %s", e)
+
+            logger.info(
+                "关键词 %s 第 %s 页爬取完成",
+                keyword,
+                page,
+            )
+
+            total_page = (
+                (json_data or {}).get("data", {}).get("mainInfo", {}).get("totalPage")
+            )
+
+            try:
+                total_page_int = int(total_page) if total_page is not None else 50
+            except (TypeError, ValueError):
+                total_page_int = 50
+            self.report_data.update({'total_pages':total_page_int})
+
+            if self.scheduler.end ==True:
+                logger.info('心跳失败')
+                self._error_report({'is_finished':0,'need_reassign':0,'current_page':page})
+                break
+
+            if page >= total_page_int:
+                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page})
+                break
+            if self.is_no_product > 100:
+                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page})
+                break
+            sleep_second = random.uniform(30, 60)
+            logger.info("第 %s 页爬取完成，休息 %.1fs", page, sleep_second)
+            self._success_report({'is_finished':0,'need_reassign':0,'current_page':page})
+            time.sleep(sleep_second)
+
+    def update_cookie(self):
+        taobao_auto = TaobaoAutoCrawl(self.account_name, self.ip, self.product)
+        if not taobao_auto.run():
+            return False
+        safe_name = self._sql_literal(self.account_name)
+        sql_account = (
+            f"select * from `retrieve_collect_equipment_account` where `username`='{safe_name}'"
+        )
+        account_list = self.db.select_data(sql_account)
+        if not account_list:
+            logger.error("账号 %s 未查询到 cookie 信息", self.account_name)
+            return False
+        cookie_str = account_list[0].get("cookie_str")
+        if not cookie_str:
+            logger.error("账号 %s cookie 为空", self.account_name)
+            return False
+        try:
+            self.cookies = json.loads(cookie_str)
+        except Exception as e:
+            logger.error("账号 %s cookie 解析失败: %s", self.account_name, e)
+            return False
+        return True
+
+    def get_account(self):
+        sql_account = """
+                      SELECT * FROM `retrieve_collect_equipment_account` WHERE `id` = %s and `status` = 0
+                      """
+        account_list = self.db.select_data(sql_account,self.account_id)
+        if not account_list:
+            return False
+
+        account_dict = account_list[0]
+        self.ip = account_dict.get("ip")
+        cookie_str = account_dict.get("cookie_str")
+
+        self.ip = account_dict.get("ip")
+        self.account_name = account_dict.get("username")
+        self.login_username = account_dict.get("phone", "")
+        self.login_password = account_dict.get("password", "")
+        self.cookie_stamp = account_dict.get("update_time")
+        if self.ip:
+            account_proxy = f"http://{self.ip}"
+            self.proxies = {"http": account_proxy, "https": account_proxy}
+        else:
+            self.proxies = None
+
+        need_refresh = (
+                not cookie_str
+                or int(time.time()) - int(self.cookie_stamp or 0) > COOKIE_MAX_AGE_SEC
+        )
+        print(account_dict)
+        if need_refresh:
+            if not self.update_cookie():
+                return False
+        else:
+            try:
+                self.cookies = json.loads(cookie_str)
+            except Exception as e:
+                logger.error("cookie 解析失败，尝试刷新: %s", e)
+                if not self.update_cookie():
+                    return False
+
+        logger.info("获取到账号: %s, ip: %s", self.account_name, self.ip)
+        self.init_session()
+        return True
+
+    def run(self):
+        if not self.get_account():
+            logger.info("==================当前无账号可用==================")
+            self.success = False
+            return self.pipeline.crawl_count, self.success
+
+        logger.info("获取到账号:%s,代理ip:%s", self.account_name, self.ip)
+
+        self.get_search()
+        logger.info(
+            "任务id:%s, 任务状态已更新, 产品名称:%s, 爬取数据:%s条",
+            self.task_id,
+            self.product,
+            self.pipeline.crawl_count,
+        )
+        return self.pipeline.crawl_count, self.success

spiders/taobao/taobao_login.py

@@ -0,0 +1,190 @@
+import time
+import json
+import random
+import signal
+import sys
+from DrissionPage import ChromiumPage, ChromiumOptions
+import re
+import socket
+from commons.conn_mysql import MySQLPoolOn2
+import hashlib
+from commons.Logger import logger
+
+MAX_PAGES = 5
+WAIT_BETWEEN_PAGES = (8, 15)  # 页间等待时间范围（秒）
+SCROLL_DELAY = (0.3, 0.8)  # 滚动延迟范围
+CLICK_DELAY = (0.5, 1.2)  # 点击延迟范围
+BROWSE_TIME = (5, 10)  # 浏览时间范围
+chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
+
+
+class TaobaoAutoCrawl:
+    def __init__(self, account_name, ip, key_word):
+        self.driver = None
+        self.register_signal_handler()
+        self.db = MySQLPoolOn2()
+        self.account_name = account_name
+        self.ip = ip
+        self.keyword = key_word
+
+    @staticmethod
+    def _get_free_port():
+        """获取一个当前可用的本地端口，供 Chrome 调试使用。"""
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.bind(("127.0.0.1", 0))
+            return s.getsockname()[1]
+
+    def init_drissionpage(self):
+        # 避免 auto_port 在部分环境下生成异常地址（无端口）导致 ChromiumPage 初始化失败
+        co = ChromiumOptions().set_browser_path(chrome_path)
+        debug_port = self._get_free_port()
+        co.set_argument(f"--remote-debugging-port={debug_port}")
+        co.set_user_data_path(f"./spider/taobao/{self.account_name}")
+        if self.ip:
+            proxy = self.ip.strip()
+            if not proxy.startswith(("http://", "https://")):
+                proxy = f"http://{proxy}"
+            co.set_argument(f"--proxy-server={proxy}")
+        logger.info(f"启动浏览器: account={self.account_name}, debug_port={debug_port}")
+        self.driver = ChromiumPage(co)
+
+    def register_signal_handler(self):
+        """ 非常必要，注册信号处理，确保状态保存"""
+
+        def signal_handler(signum, frame):
+            print("\n⚠️ 收到退出信号，正在保存状态并退出...")
+            if self.driver:
+                self.driver.quit()
+            sys.exit(0)
+
+        signal.signal(signal.SIGINT, signal_handler)
+        # Windows 上可能不支持 SIGTERM，做兼容处理
+        if hasattr(signal, "SIGTERM"):
+            signal.signal(signal.SIGTERM, signal_handler)
+
+    # ==================== 人工行为模拟 ====================
+    def random_wait(self, min_sec, max_sec=None):
+        """随机等待"""
+        if max_sec is None:
+            max_sec = min_sec
+        time.sleep(random.uniform(min_sec, max_sec))
+
+    def move_mouse_to_element(self, element):
+        """移动鼠标到元素"""
+        if not element:
+            return
+        try:
+            # 优先使用 DrissionPage 推荐方式
+            self.driver.actions.move_to(element)
+        except Exception:
+            # 兼容旧逻辑：按元素中心点移动（不同版本 move 参数可能不同）
+            box = element.rect
+            try:
+                center_x = int(box.x + box.width / 2)
+                center_y = int(box.y + box.height / 2)
+            except Exception:
+                center_x = int(box["x"] + box["width"] / 2)
+                center_y = int(box["y"] + box["height"] / 2)
+            try:
+                self.driver.actions.move(center_x, center_y)
+            except TypeError:
+                # 某些版本仅支持关键字参数
+                self.driver.actions.move(offset_x=center_x, offset_y=center_y)
+        self.random_wait(0.2, 0.5)
+
+    def human_type(self, element, text):
+        """模拟人类输入"""
+        for char in text:
+            element.send_keys(char)
+            time.sleep(random.uniform(0.1, 0.3))
+
+    def login(self, username, password):
+        self.driver.get("https://login.taobao.com")
+        self.random_wait(5, 8)
+
+        # 输入账号
+        login_name = self.driver.ele("xpath=//input[@name='fm-login-id']", timeout=30)
+        if login_name:
+            self.move_mouse_to_element(login_name)
+            self.human_type(login_name, username)
+            self.random_wait(1, 3)
+
+        # 输入密码
+        login_pass = self.driver.ele("xpath=//input[@name='fm-login-password']", timeout=30)
+        if login_pass:
+            self.move_mouse_to_element(login_pass)
+            self.human_type(login_pass, password)
+            self.random_wait(1, 3)
+
+        # 点击登录
+        login_button = self.driver.ele("xpath=//button[text()='登录']", timeout=30)
+        if login_button:
+            self.move_mouse_to_element(login_button)
+            login_button.click()
+            self.random_wait(1, 3)
+
+        # 处理同意按钮
+        login_agree = self.driver.ele("xpath=//button[text()='同意']", timeout=5)
+        if login_agree:
+            self.move_mouse_to_element(login_agree)
+            login_agree.click()
+            self.random_wait(1, 3)
+
+        # 等待登录结果
+        self.random_wait(10, 20)
+        # 检查是否登录成功
+        user_info = self.driver.ele("xpath=//a[@class='site-nav-login-info-nick']", timeout=10)
+        if user_info:
+            print("登录成功！")
+        else:
+            print("登录失败，请检查账号密码或验证码")
+
+    def get_search(self):
+        url = "https://www.taobao.com"
+        self.driver.get(url, timeout=30)
+        time.sleep(30)
+        time.sleep(random.uniform(3, 8))
+
+        # 刷新一次，否则可能未找到登录状态
+        self.driver.refresh()
+        self.random_wait(5, 10)
+        # login_name = self.driver.ele("xpath=//input[@name='fm-login-id']")
+        # if login_name:
+        #     self.login("aqwwer","wewetrv")
+        #     self.driver.refresh()
+        #     self.random_wait(5, 10)
+
+        ele_iframe = self.driver.ele("xpath=//iframe[@id='baxia-dialog-content']")
+        if ele_iframe:
+            update_sql = f""" UPDATE `retrieve_collect_equipment_account` SET `status`= %s WHERE `nickname` = %s; """
+            self.db.execute(update_sql, (1, self.account_name))
+            return False
+
+        ele = self.driver.ele('xpath=//*[contains(@class,"site-nav-login-info-nick")]', timeout=30)
+
+        if ele:
+            cookies_list = self.driver.cookies()
+            cookies_dict = {c['name']: c['value'] for c in cookies_list}
+            timestamp = int(time.time())
+            # 保存 cookie 到文件
+            update_sql = f""" UPDATE `retrieve_collect_equipment_account` SET `update_time` = %s, `cookie_str`= %s,`status`= %s WHERE `username` = %s; """
+            self.db.execute(update_sql, (timestamp, json.dumps(cookies_dict), 0, self.account_name))
+            print(f"{self.account_name},获取 cookie 成功！")
+            logger.info(f"{self.account_name},获取 cookie 成功！")
+
+            self.random_wait(3, 5)
+            return True
+        else:
+            return False
+
+    def run(self):
+        bool_login = False
+        try:
+            self.init_drissionpage()
+            bool_login = self.get_search()
+        except Exception as e:
+            logger.exception(f"{self.account_name} 获取 cookie 异常: {e}")
+        finally:
+            if self.driver:
+                return self.driver
+        return bool(bool_login)