.idea/.gitignore

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

.idea/inspectionProfiles/profiles_settings.xml

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

.idea/misc.xml

@@ -1,7 +0,0 @@
-<?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

@@ -1,8 +0,0 @@
-<?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

@@ -1,12 +0,0 @@
-<?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

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

README.md

@@ -1,39 +0,0 @@
-# 淘宝商品详情采集
-
-淘宝药品/商品数据采集系统，支持列表搜索和详情页 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

@@ -1,51 +0,0 @@
-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

@@ -1,126 +0,0 @@
-"""定时从调度库拉取任务、执行爬虫并上报状态的通用入口。"""
-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

@@ -1 +0,0 @@
-

commons/conn_mysql.py

@@ -1,176 +0,0 @@
-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

@@ -1,69 +0,0 @@
-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

@@ -1,121 +0,0 @@
-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

@@ -1,70 +0,0 @@
-"""
-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

@@ -1,77 +0,0 @@
-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

@@ -1,627 +0,0 @@
-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

@@ -1,652 +0,0 @@
-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

@@ -1,194 +0,0 @@
-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

@@ -1,596 +0,0 @@
-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

@@ -1,190 +0,0 @@
-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)