feat(spider): 新增潮谷街爬虫及数据库连接池配置

- 添加application.yml，配置MySQL连接参数，支持环境变量覆盖
- 新增cgj_daily_spider.py，实现潮谷街每日增量采集功能
- cgj_daily_spider包括商户发现、商品采集、玩家采集和直播回放补采
- 使用requests和tenacity实现接口请求及重试机制，支持代理配置
- 基于MySQLConnectionPool管理数据库连接，批量写入商户、商品和玩家数据
- 增加日志记录，支持每天生成日志文件和日志级别过滤
- cgj_history_spider.py实现潮谷街历史全量采集，复用日增量模块功能
- 新增mysql_pool.py，封装支持连接池的MySQL操作类，配置从yaml读取
- 主流程及定时任务调度实现每日定时增量采集，历史脚本单次全量运行

chaogujie_spider/README.md

@@ -0,0 +1,104 @@
+# 潮谷街 App 抓包逆向分析文档
+
+> 记录日期：2026/07/03
+
+## 1. 目标信息
+
+- **App**：潮谷街（吃谷商城）
+- **包名**：`com.chaogujie.cgjec`
+- **版本**：1.0.15（versionCode 17，minSdk 21，targetSdk 34）
+- **技术栈**：**Flutter**（`libflutter.so` + `libapp.so`，14MB Dart AOT 产物）
+- **测试设备**：Pixel 5（redfin）/ Android 11 / arm64-v8a，已 root（Magisk + LSPosed）
+- **抓包工具**：Reqable
+- **反爬 / 安全措施**（apk native 库暴露）：
+  - `libturingmfa.so` —— 腾讯天御 MFA 风控（设备指纹 / 防注入 / 防 root）
+  - `libzxprotect.so` —— 加固 / 保护壳
+  - `libYTCommonLiveness.so` / `libkyctoolkit.so` —— 活体检测 / KYC 实名
+  - `libImSDK.so` / `libliteavsdk.so` —— 腾讯 IM + 直播
+  - `libBugly_Native.so` —— 崩溃上报
+
+## 2. 分析过程
+
+### 现象
+
+用 Reqable 抓包时，潮谷街一打开抓包就提示「网络请求失败，请稍后重试」，Reqable 里对应请求状态为 `Aborted`，详情显示 **「客户端 SSL 握手失败」**。同一环境下另一个测试 App 可以正常抓包。
+
+已尝试但无效的操作：安装并勾选 JustTrustMePro（LSPosed 模块）、重启手机——抓包一开 App 就网络异常。
+
+### 定位技术栈
+
+拉取 apk 后扫描 native 库，锁定关键证据：
+
+```bash
+unzip -l 潮谷街_1.0.15.apk | grep -iE "lib/arm64-v8a/"
+# lib/arm64-v8a/libflutter.so           ← Flutter 引擎
+# lib/arm64-v8a/libapp.so  (14MB)       ← Flutter 业务代码(Dart AOT)
+# lib/arm64-v8a/libdart_native_imsdk.so
+```
+
+**确认是 Flutter 应用** —— 这是整个排查的分水岭。
+
+### 逐层验证证书链
+
+1. 检查用户 CA store：`/data/misc/user/0/cacerts-added/` —— **空**。
+2. 检查系统 CA store：`mount | grep cacerts` —— 已被 Magisk 以 tmpfs 挂载（装了证书模块）。
+3. 检查设备已启用 Magisk 模块：`movecert`、`reqable-magisk`、`zygisk_shamiko`、`zygisk_lsposed`、`zygiskfrida`。
+4. 解析系统 CA 里的 Reqable 证书 `aa381f31.0`：
+
+   ```
+   CN     = Reqable CA (Dec 18, 2025)
+   有效期 = 2025-12-18 ~ 2033-05-31（有效）
+   SHA1   = 7C:E8:1E:44:8F:77:88:21:36:FC:64:6E:F7:CE:EB:97:CF:03:EA:43
+   ```
+
+5. 与 PC 端当前 CA（`AppData/Roaming/Reqable/certificate/reqable-root.crt`）指纹对比 —— **完全一致**。
+
+结论：证书装进了系统 CA、指纹匹配、在有效期内，普通 App 也能抓 HTTPS，证书链本身没问题。
+
+## 3. 逆向思路
+
+关键认知：**Flutter 的网络请求走它自己打包在 `libflutter.so` 里的 BoringSSL，TLS 握手与证书校验全在 native 层完成，完全不经过 Java 层的 `javax.net.ssl` / OkHttp / WebView。**
+
+由此推导：
+
+- 任何 **Java 层的 unpinning 插件**（JustTrustMe / JustTrustMePro / LSPosed 证书类模块）对 Flutter **一律无效** —— 这就解释了为什么怎么勾选、重启都不生效。
+- Flutter（Dart `HttpClient`）在 Android 上只信任**系统 CA store**，不读用户证书。所以证书必须装到系统级。
+
+中途一度怀疑是 Flutter 的额外 native pinning（需要 Frida hook `libflutter.so`），并已排查设备的 Frida 环境（客户端 17.9.1、`zygiskfrida` 模块在位）。但最终发现真正原因更简单，见踩坑记录。
+
+## 4. 解决方案
+
+**升级 Reqable 证书 Magisk 模块**：旧版（0.1）→ 新版（1.2）。升级后 Flutter 立刻能读到系统 CA 里的 Reqable 证书，抓包恢复正常。
+
+抓到的接口域名：
+
+| 域名 | 用途 |
+|---|---|
+| `cgj.chaogujieapp.com` | 主 API |
+| `cgjsj.chaogujieapp.com` | 数据 / 实时接口 |
+| `5fdc92.chaogujieapp.com` | 疑似 CDN / 资源 |
+| `license.vod2.myqcloud.com` | 腾讯云点播 license（直播） |
+| `android.bugly.qq.com` | 崩溃上报（可忽略） |
+
+## 5. 踩坑记录
+
+1. **「指纹一致 + mount 已挂载」≠「Flutter 能读到」**
+   本次最大的坑：旧版证书模块（0.1）在 Android 11 上虽然把 Reqable CA 挂进了系统 CA store（`mount` 看得到、指纹也对得上），但那种挂载方式对 Flutter 的 BoringSSL 读取路径**不生效**。升级到 1.2 版本后，模块修正了挂载方式（大概率适配了 Android 10+ 的 CA store 读取路径 / 挂载时机），Flutter 才认。排查时不能只看「证书在不在、指纹对不对」，还要确认**证书模块版本够不够新**。
+
+2. **JustTrustMePro 对 Flutter 无效不是配置问题**
+   反复检查 LSPosed 作用域、重启，都改变不了结果——因为方案本身打不中 Flutter 的 native TLS，不是勾选姿势的问题。
+
+3. **潮谷街其实没有额外 native pinning**
+   走的是 Flutter 默认「读系统 CA」校验。所以最终没用上 Frida —— 中途「怀疑 pinning、准备 hook `libflutter.so`」的方向偏保守了。
+
+## 6. 举一反三
+
+**Flutter App 抓包失败（客户端 SSL 握手失败）标准排查顺序：**
+
+1. 先判断是不是 Flutter：`unzip -l xxx.apk | grep libflutter.so`。是 Flutter，就**跳过所有 Java 层方案**（JustTrustMe 系列别再试）。
+2. 确认证书装进**系统 CA store**（不是用户 CA）—— Flutter 只读系统 CA。
+3. 确认**证书模块版本够新**（本次的坑）——「已挂载 + 指纹一致」不代表 Flutter 读得到。
+4. 确认走 **VPN / Tun 透明代理**模式 —— Flutter 默认不读系统 WiFi 代理，普通代理模式流量根本不过 Reqable。
+5. 以上全对仍失败，才是真 native pinning，此时唯一可靠手段：**Frida hook `libflutter.so` 的 BoringSSL 证书校验函数**（参考 NVISO 的 disable-flutter-tls-verification 脚本）；若担心风控检测（如天御 `libturingmfa`），配合 `zygisk_shamiko` + Zygisk DenyList 隐藏。
+
+**通用结论**：unpinning 方案能不能生效，取决于它作用的层次是否覆盖目标 App 的网络栈——Java 层框架管不到 native 层的 BoringSSL。

chaogujie_spider/YamlLoader.py

@@ -0,0 +1,98 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2025/12/22 10:44
+import os, re
+import yaml
+
+regex = re.compile(r'^\$\{(?P<ENV>[A-Z_\-]+:)?(?P<VAL>[\w.]+)}$')
+
+
+class YamlConfig:
+    def __init__(self, config):
+        self.config = config
+
+    def get(self, key: str):
+        return YamlConfig(self.config.get(key))
+
+    def getValueAsString(self, key: str):
+        try:
+            match = regex.match(self.config[key])
+            group = match.groupdict()
+            if group['ENV'] is not None:
+                env = group['ENV'][:-1]
+                return os.getenv(env, group['VAL'])
+            return None
+        except:
+            return self.config[key]
+
+    def getValueAsInt(self, key: str):
+        try:
+            match = regex.match(self.config[key])
+            group = match.groupdict()
+            if group['ENV'] is not None:
+                env = group['ENV'][:-1]
+                return int(os.getenv(env, group['VAL']))
+            return 0
+        except:
+            return int(self.config[key])
+
+    def getValueAsBool(self, key: str):
+        try:
+            match = regex.match(self.config[key])
+            group = match.groupdict()
+            if group['ENV'] is not None:
+                env = group['ENV'][:-1]
+                return bool(os.getenv(env, group['VAL']))
+            return False
+        except:
+            return bool(self.config[key])
+
+
+def _resolve_path(path: str) -> str:
+    """
+    解析 yaml 文件路径，按优先级查找：
+      1) 绝对路径或 cwd 下存在 → 直接用（保留旧行为，向后兼容）
+      2) 调用方主脚本所在目录 → 兜底，方便打包后从任意 cwd 启动
+    :param path: (str) 用户传入的路径，默认 'application.yml'
+    :return: (str) 实际可读取的完整路径；找不到则返回原 path 让 open() 抛错
+    """
+    # 1) 旧行为：cwd 或绝对路径
+    if os.path.exists(path):
+        return path
+
+    # 2) 主脚本目录（__main__.__file__）
+    try:
+        import __main__
+        main_file = getattr(__main__, '__file__', None)
+        if main_file:
+            candidate = os.path.join(os.path.dirname(os.path.abspath(main_file)), path)
+            if os.path.exists(candidate):
+                return candidate
+    except Exception:
+        pass
+
+    return path
+
+
+def readYaml(path: str = 'application.yml', profile: str = None) -> YamlConfig:
+    """
+    读取 yaml 配置。
+    :param path: (str) yaml 文件路径，默认 'application.yml'。
+                       优先 cwd / 绝对路径（保留旧行为），找不到再 fallback 到主脚本所在目录。
+    :param profile: (str) 可选环境后缀，如 'dev' 会额外加载 'application-dev.yml' 并 update
+    :return: (YamlConfig) 配置访问对象
+    :raises FileNotFoundError: cwd 和主脚本目录都找不到时抛出
+    """
+    real_path = _resolve_path(path)
+    with open(real_path, encoding='utf-8') as fd:
+        conf = yaml.load(fd, Loader=yaml.FullLoader)
+
+    if profile is not None:
+        result = real_path.rsplit('.', 1)
+        profiledYaml = f'{result[0]}-{profile}.{result[1]}'
+        if os.path.exists(profiledYaml):
+            with open(profiledYaml, encoding='utf-8') as fd:
+                conf.update(yaml.load(fd, Loader=yaml.FullLoader))
+
+    return YamlConfig(conf)

chaogujie_spider/application.yml

@@ -0,0 +1,6 @@
+mysql:
+  host: ${MYSQL_HOST:100.64.0.25}
+  port: ${MYSQL_PROT:3306}
+  username: ${MYSQL_USERNAME:crawler}
+  password: ${MYSQL_PASSWORD:Pass2022}
+  db: ${MYSQL_DATABASE:crawler}

chaogujie_spider/cgj_daily_spider.py

@@ -0,0 +1,764 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/07/06
+"""潮谷街（吃谷商城）每日增量采集爬虫。
+
+三级采集管道（结构对齐 kaji 的 kj_daily_spider）：
+    1. 商户发现：翻页遍历「shop 列表页」group_buying_products（product_types/category_id 模式），
+       从每个商品里提取 merchant_id / merchant_name，写入 cgj_shop_record。
+    2. 商品采集：遍历库内每个商户的「历史成交」group_buying_products（merchant_id + status 模式），
+       逐商品补 product/info 详情后写入 cgj_product_record。
+    3. 玩家采集：遍历 cgj_product_record 中未采集过玩家的商品，抓 announcement_by_user
+       的买家名单写入 cgj_player_record。
+
+鉴权：潮谷街这几个接口 access-token 为空、无签名/加密，属公开（游客）接口，
+只需带固定的 devicecode / version 等请求头即可，适合长期无人值守。
+
+翻页约定（三个列表接口统一）：请求体/参数带 cursor（首页为空串），响应返回
+    data.list（本页数据）、data.cursor（下一页游标）、data.done（true=到底）。
+    循环条件：done 为 True 或 cursor 为空即停止。
+
+【字段解析】save_shops / build_product / save_players 里的 `for item in items:` 循环中，
+"""
+import sys
+import time
+from datetime import datetime
+
+import requests
+import schedule
+from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
+from mysql_pool import MySQLConnectionPool
+
+logger.remove()
+logger.add("./logs/{time:YYYYMMDD}.log", encoding="utf-8", rotation="00:00",
+           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
+           level="DEBUG", retention="7 day")
+
+# ==================== 基础配置 ====================
+# 业务域名（来源：抓包，四个接口都走这个域）
+BASE = "https://5fdc92.chaogujieapp.com"
+
+PAGE_SIZE = 20  # 列表接口每页条数（抓包实测 20）
+PLAYER_PAGE_SIZE = 20  # 玩家名单每页条数（抓包实测 20）
+# MAX_PAGES = 500  # 单个列表翻页保护上限，防异常时无限翻页
+# 增量早停阈值：某商户历史成交连续这么多页「全是已入库商品(0 新)」即停止翻页。
+# 用「连续多页」而非「单页」，兼容历史成交列表里新商品跨页穿插的情况；调大更稳、调小更省。
+STOP_AFTER_DUPE_PAGES = 2
+MAX_PAGES = 5  # 单个列表翻页保护上限，防异常时无限翻页  每日任务时设置更小限制
+
+# 「历史成交」过滤值（来源：抓包固定 status=1796）。九尾等仅有在售无历史成交的商户返回 0 条，符合预期。
+HISTORY_STATUS = 1796
+
+# 设备码 / App 版本（来源：抓包，公开接口无需登录 token）
+DEVICE_CODE = "8420C51411ECF16C802600868F2D214A"
+APP_VERSION = "1.0.15"
+
+# 是否为每个商品补拉 product/info 详情（拿系列/规格/简介/视频/上下架时间等）。
+# 关闭可大幅减少请求量，但详情维度字段会缺失。
+FETCH_DETAIL = True
+# 是否为每个商品补拉直播回放（live/user_live/detail，room_id 取自详情的 live_room_id）。
+# 依赖 FETCH_DETAIL=True——要先有详情才能拿到 live_room_id。
+FETCH_REPLAY = True
+# 是否使用代理。潮谷街实测可直连；若遇 IP 风控再置 True 并配置 get_proxys。
+USE_PROXY = False
+
+# 固定 UA（来源：抓包，潮谷街为 Flutter WebView）
+UA = ("Mozilla/5.0 (Linux; Android 11; Pixel 5 Build/RQ3A.211001.001; wv) "
+      "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/148.0.7778.120 "
+      "Mobile Safari/537.36")
+
+# 基础请求头（来源：抓包）。POST 的 content-type 由 requests 的 json= 自动补，不放这里。
+BASE_HEADERS = {
+    "user-agent": UA,
+    "accept-encoding": "gzip",
+    "channel-version": APP_VERSION,  # 抓包 channel-version
+    "access-token": "",  # 抓包为空：公开/游客接口，无需登录
+    "source-id": "1",  # 抓包固定
+    "language": "zh",  # 抓包固定
+    "version": APP_VERSION,  # 抓包 App 版本
+    "devicecode": DEVICE_CODE,  # 抓包设备码
+    "platform-id": "1",  # 抓包固定（1=Android）
+    "channel-name": "huawei",  # 抓包渠道
+}
+
+
+def after_log(retry_state):
+    """tenacity 重试回调，记录每次尝试的结果。
+
+    Args:
+        retry_state: tenacity 传入的 RetryCallState 对象，含调用参数与结果。
+    """
+    # 约定业务函数首个位置参数为 log；取不到时回退全局 logger
+    if retry_state.args and len(retry_state.args) > 0:
+        log = retry_state.args[0]
+    else:
+        log = logger
+
+    if retry_state.outcome.failed:
+        log.warning(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times")
+    else:
+        log.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded")
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
+def get_proxys(log):
+    """获取隧道代理配置（默认不启用，见 USE_PROXY）。
+
+    Args:
+        log: 日志对象。
+
+    Returns:
+        dict: requests 可用的 proxies 字典。
+
+    Raises:
+        Exception: 组装代理配置异常时向上抛出以触发重试。
+    """
+    tunnel = "x371.kdltps.com:15818"
+    kdl_username = "t13753103189895"
+    kdl_password = "o0yefv6z"
+    try:
+        proxies = {
+            "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel},
+            "https": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password,
+                                                             "proxy": tunnel},
+        }
+        return proxies
+    except Exception as e:
+        log.error(f"Error getting proxy: {e}")
+        raise e
+
+
+def ts_to_dt(ts) -> str | None:
+    """将秒级 Unix 时间戳转成 'YYYY-MM-DD HH:MM:SS' 字符串。
+
+    Args:
+        ts (int | str | None): 秒级时间戳；为 None / 0 / 空时返回 None。
+
+    Returns:
+        str | None: 格式化时间字符串；无有效时间戳时返回 None。
+    """
+    if not ts:
+        return None
+    try:
+        return datetime.fromtimestamp(int(ts)).strftime("%Y-%m-%d %H:%M:%S")
+    except (ValueError, TypeError, OSError):
+        return None
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
+def cgj_request(log, path: str, method: str = "GET",
+                params: dict = None, json_body: dict = None,
+                extra_headers: dict = None) -> dict | None:
+    """潮谷街通用请求函数（带重试）。
+
+    潮谷街接口无签名/加密，仅需固定请求头，故本函数只负责拼 URL、发请求、判状态码。
+
+    Args:
+        log: 日志对象。
+        path (str): 接口相对路径（以 / 开头，不含域名），如 "/api/home/group_buying_products"。
+        method (str, optional): 请求方法，"GET" 或 "POST"。Defaults to "GET"。
+        params (dict, optional): URL query 参数。Defaults to None。
+        json_body (dict, optional): POST 的 JSON body。Defaults to None。
+        extra_headers (dict, optional): 追加 / 覆盖的请求头。Defaults to None。
+
+    Returns:
+        dict | None: 响应 JSON；非 200 时抛异常触发重试。
+
+    Raises:
+        RuntimeError: HTTP 状态码非 200 时抛出。
+    """
+    url = f"{BASE}{path}"
+    req_headers = BASE_HEADERS.copy()
+    if extra_headers:
+        req_headers.update(extra_headers)
+
+    proxies = get_proxys(log) if USE_PROXY else None
+    if method.upper() == "POST":
+        resp = requests.post(url, headers=req_headers, json=json_body, params=params, timeout=(5, 30), proxies=proxies)
+    else:
+        resp = requests.get(url, headers=req_headers, params=params, timeout=(5, 30), proxies=proxies)
+
+    if resp.status_code != 200:
+        log.error(f"请求失败 {resp.status_code}: {url}")
+        raise RuntimeError(f"HTTP {resp.status_code}")
+    return resp.json()
+
+
+def get_group_buying_page(log, cursor: str = "", merchant_id=None) -> dict | None:
+    """获取 group_buying_products 一页（发现模式 / 商户历史成交模式二合一）。
+
+    该接口一站两用：
+        - 不传 merchant_id：全局「shop 列表页」，用 product_types/category_id 过滤，用于商户发现。
+        - 传 merchant_id：某商户「历史成交」，用 status=HISTORY_STATUS 过滤。
+    两种模式响应结构一致（data.list / data.cursor / data.done）。
+
+    Args:
+        log: 日志对象。
+        cursor (str, optional): 翻页游标，首页传空串。Defaults to ""。
+        merchant_id (int | str | None, optional): 商户 id；None=商户发现模式。Defaults to None。
+
+    Returns:
+        dict | None: 响应 JSON；失败时由 cgj_request 抛异常触发重试。
+    """
+    if merchant_id is not None:
+        body = {
+            "cursor": cursor,
+            "size": PAGE_SIZE,
+            "image_size": "68",
+            "status": HISTORY_STATUS,
+            "merchant_id": merchant_id,
+        }
+    else:
+        body = {
+            "cursor": cursor,
+            "size": PAGE_SIZE,
+            "image_size": "178",
+            "product_types": [1],
+            "category_id": 1,
+        }
+    return cgj_request(log, "/api/home/group_buying_products", method="POST", json_body=body)
+
+
+# ==================== 一、商户发现（shop 列表页） ====================
+def save_shops(log, items: list, sql_pool, seen: set) -> int:
+    """从 shop 列表页的商品项提取商户并写入 cgj_shop_record（存在则更新店名并复活）。
+
+    ------------------------------------------------------------------
+    商户维度目前只取驱动管道必需的 merchant_id / merchant_name，
+    如需更多商户字段（如 merchant_image_url），在下面 for 循环里往 data_dict 加即可，
+    并同步 schema.sql 的 cgj_shop_record 列与下方 upsert 的 SQL。
+    ------------------------------------------------------------------
+
+    Args:
+        log: 日志对象。
+        items (list[dict]): group_buying_products 返回的 data.list，每项含 merchant_id / merchant_name。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+        seen (set): 跨页去重的 merchant_id 集合，避免重复写库。
+
+    Returns:
+        int: 本次实际写库的商户数（已去重）。
+    """
+    info_list = []
+    for item in items:
+        merchant_id = item.get("merchant_id")  # 商户 id（管道必需）
+        merchant_name = item.get("merchant_name")  # 商户名（管道必需）
+
+        data_dict = {"shop_id": merchant_id, "shop_name": merchant_name}
+
+        if not merchant_id or merchant_id in seen:
+            log.info(f"商户 {merchant_id} 已存在 seen，跳过")
+            continue
+        seen.add(merchant_id)
+        info_list.append(data_dict)
+
+    if not info_list:
+        log.info("本页无新商户，不写库")
+        return 0
+
+    # 存在则更新店名并把 is_deleted 刷回 0：能出现在列表 = 商户在营业（曾注销的在此复活）
+    sql = ("INSERT INTO cgj_shop_record (shop_id, shop_name) VALUES (%s, %s) "
+           "ON DUPLICATE KEY UPDATE shop_name = VALUES(shop_name), is_deleted = 0")
+    args_list = [(d["shop_id"], d["shop_name"]) for d in info_list]
+    # print(args_list)
+    sql_pool.insert_many(query=sql, args_list=args_list)
+    return len(info_list)
+
+
+def get_shop_list(log, sql_pool) -> int:
+    """翻页遍历 shop 列表页，发现并写入全部商户。
+
+    翻页靠响应的 done / cursor 判断，merchant_id 唯一键去重兜底。
+
+    Args:
+        log: 日志对象。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 去重后发现的商户总数。
+    """
+    seen = set()
+    cursor = ""
+    page = 0
+
+    while page < MAX_PAGES:
+        try:
+            resp = get_group_buying_page(log, cursor, merchant_id=None)
+        except Exception as e:
+            log.error(f"shop 列表页 cursor={cursor!r} 请求失败: {e}")
+            break
+        if not resp or resp.get("code") != 0:
+            log.info(f"shop 列表页返回异常: {resp.get('msg') if resp else None}")
+            break
+
+        data = resp.get("data") or {}
+        items = data.get("list") or []
+        if not items:
+            log.info(f"shop 列表页第 {page + 1} 页无数据，停止翻页")
+            break
+
+        save_shops(log, items, sql_pool, seen)
+        log.info(f"shop 列表页第 {page + 1} 页完成，本页 {len(items)} 条，累计商户 {len(seen)}")
+
+        if data.get("done") or not data.get("cursor"):
+            break
+        cursor = data["cursor"]
+        page += 1
+
+    return len(seen)
+
+
+# ==================== 二、商品采集（商户历史成交 + 详情） ====================
+def get_product_detail(log, product_id) -> dict:
+    """获取商品详情 product/info，返回 data 节点（含 product_info / merchant_info 等）。
+
+    Args:
+        log: 日志对象。
+        product_id (int | str): 商品 id。
+
+    Returns:
+        dict: 详情 data 字典；无数据时返回空字典。
+    """
+    resp = cgj_request(log, "/api/product/info", method="GET", params={"id": product_id})
+    return (resp or {}).get("data") or {}
+
+
+def get_live_detail(log, room_id: str) -> dict:
+    """获取直播回放详情 live/user_live/detail，返回 data 节点。
+
+    room_id 取自商品详情的 data.live_room_id。回放视频地址在 data.offline_address_info[].url
+    （腾讯云 vod，name="默认"），另带 status / start_time / end_time / duration 等直播信息。
+
+    Args:
+        log: 日志对象。
+        room_id (str): 直播间 id（商品详情的 live_room_id）。
+
+    Returns:
+        dict: 回放详情 data 字典；room_id 为空或无数据时返回空字典。
+    """
+    if not room_id:
+        return {}
+    resp = cgj_request(log, "/api/live/user_live/detail", method="POST", json_body={"room_id": room_id})
+    return (resp or {}).get("data") or {}
+
+
+def build_product(item: dict, detail: dict, shop_id, shop_name: str, live_detail: dict = None) -> dict:
+    """把历史成交列表项 + 商品详情组装成 cgj_product_record 一行。
+
+    ------------------------------------------------------------------
+    列表项字段（item.*）来自 group_buying_products；详情字段（detail.*）来自 product/info。
+    ------------------------------------------------------------------
+
+    Args:
+        item (dict): 历史成交列表里的商品项（group_buying_products data.list 项）。
+        detail (dict): product/info 的 data 节点；FETCH_DETAIL 关闭或失败时为空字典。
+        shop_id (int | str): 商户 id。
+        shop_name (str): 商户名。
+        live_detail (dict, optional): live/user_live/detail 的 data 节点（回放信息）；
+            FETCH_REPLAY 关闭 / 无 live_room_id / 失败时为空字典。Defaults to None。
+
+    Returns:
+        dict: 与 cgj_product_record 列对应的数据字典。
+    """
+    # 详情里的嵌套节点（按需取用）
+    product_info = detail.get("product_info") or {}
+    basic_info = product_info.get("basic_info") or {}
+    # print(product_info)
+    # print('-------------------')
+    # print(basic_info)
+
+    # 直播回放（待确认）：回放地址在 offline_address_info[].url（腾讯云 vod），取第一个（name="默认"）
+    live_detail = live_detail or {}
+    offline_list = live_detail.get("offline_address_info") or []
+    replay_url = offline_list[0].get("url") if offline_list else None
+
+    row = {
+        # ---- 管道必需（勿删）----
+        "pid": item.get("id"),  # 商品 id
+        "shop_id": shop_id,  # 商户 id
+        "shop_name": shop_name,  # 商户名
+
+        # ---- 列表项字段（item.*，待确认）----
+        "title": item.get("name"),  # 商品名
+        "price": item.get("price"),  # 价格
+        "imgs": item.get("image_url"),  # 主图
+        "stock_quantity": item.get("stock_quantity"),  # 库存
+        "sold_quantity": item.get("sold_quantity"),  # 已售
+        "remaining_quantity": item.get("remaining_quantity"),  # 剩余
+        "card_edition_name": item.get("card_edition_name"),  # 卡版名
+
+        # ---- 详情字段（detail.*，待确认；FETCH_DETAIL 关闭时为空）----
+        "specs": basic_info.get("specs"),  # 规格
+        "status": product_info.get("status"),  # 商品状态
+        "listing_beg_time": product_info.get("listing_beg_time"),  # 上架时间(详情直接给文本)
+        "listing_end_time": product_info.get("listing_end_time"),  # 下架时间(详情文本)
+
+        # ---- 直播回放字段（来自 live/user_live/detail；FETCH_REPLAY 关闭时为空）----
+        "live_room_id": detail.get("live_room_id"),  # 直播间 id（补采回放用，勿删）
+        "replay_url": replay_url,  # 回放视频地址(offline_address_info[0].url，多个只取下标0)
+        "live_start_time": live_detail.get("start_time"),  # 开播时间
+        "live_end_time": live_detail.get("end_time"),  # 结束时间
+    }
+    return row
+
+
+def filter_new_pids(shop_id, page_pids: list, sql_pool) -> set:
+    """在「本页」的 pid 里查出哪些是新的（库中不存在），只查这一页、不整表拉取。
+
+    为什么按页查而非一次性把该商户全部 pid 拉成集合：某些商户历史成交可能几十万条，
+    整表拉取会 SELECT 出几十万行并在内存堆一个几十 MB 的集合，每轮每店都来一遍不划算。
+    改为每页用 `pid IN (...)` 批量查已存在的（凭 pid 唯一索引很快），内存只占一页的量，
+    商户体量再大也不受影响；增量本就只翻头几页（连续旧页即早停），每店只几次小查询。
+
+    历史成交列表顺序不稳定、新商品会跨页穿插，所以判重用「集合」而非「单个停止线精确匹配」，
+    再配合调用方的「连续 STOP_AFTER_DUPE_PAGES 页全旧」早停。
+
+    Args:
+        shop_id (int | str): 商户 id。
+        page_pids (list): 本页商品 pid 列表（原始类型，可能为 int）。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        set[int]: 本页中「库里还没有」的 pid 集合；本页为空时返回空集合。
+            pid 库内与 API 均为 int，直接按 int 比较（索引最优）。
+    """
+    pids = [int(p) for p in page_pids if p is not None]
+    if not pids:
+        return set()
+    placeholders = ",".join(["%s"] * len(pids))
+    rows = sql_pool.select_all(
+        f"SELECT pid FROM cgj_product_record WHERE shop_id = %s AND pid IN ({placeholders})",
+        (shop_id, *pids),
+    )
+    existing = {r[0] for r in rows} if rows else set()
+    return {p for p in pids if p not in existing}
+
+
+def get_sold_list(log, shop_id, shop_name: str, sql_pool, incremental: bool = True) -> int:
+    """遍历某商户历史成交翻页，逐商品补详情后写入 cgj_product_record。
+
+    Args:
+        log: 日志对象。
+        shop_id (int | str): 商户 id。
+        shop_name (str): 商户名。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+        incremental (bool, optional): True(daily) 碰到最新已采 pid 早停，只采新商品；
+            False(history) 全量深翻所有页。Defaults to True。
+
+    Returns:
+        int: 本商户写入的商品数。
+    """
+    cursor = ""
+    page = 0
+    saved = 0
+    dupe_pages = 0  # 连续「全是已入库商品(0 新)」的页数，达到阈值即增量早停
+
+    while page < MAX_PAGES:
+        try:
+            resp = get_group_buying_page(log, cursor, merchant_id=shop_id)
+            # print(resp)
+        except Exception as e:
+            log.error(f"商户 {shop_id} 历史成交 cursor={cursor!r} 请求失败: {e}")
+            break
+
+        if not resp or resp.get("code") != 0:
+            log.info(f"商户 {shop_id} 历史成交返回异常: {resp.get('msg') if resp else None}")
+            break
+
+        data = resp.get("data") or {}
+        items = data.get("list") or []
+        if not items:
+            log.info(f"商户 {shop_id} 历史成交第 {page + 1} 页无数据，停止翻页")
+            break
+
+        # 增量判重：只查本页 pid 里哪些是新的（不整表拉取，适配商户体量增长）
+        new_pids = filter_new_pids(shop_id, [it.get("id") for it in items], sql_pool) if incremental else None
+
+        batch = []
+        new_on_page = 0  # 本页「新商品」数（用于连续全旧页早停判断）
+        for item in items:
+            pid = item.get("id")
+            # 已入库的商品直接跳过，连详情/回放都不再请求（省大量无用请求）
+            if incremental and int(pid) not in new_pids:
+                continue
+            new_on_page += 1
+
+            detail = {}
+            if FETCH_DETAIL:
+                try:
+                    detail = get_product_detail(log, pid)
+                except Exception as e:
+                    log.error(f"商品 {pid} 详情请求失败: {e}")
+
+            # 直播回放：room_id 取自详情的 live_room_id，再单独请求 live/user_live/detail
+            live_detail = {}
+            if FETCH_REPLAY and detail:
+                room_id = detail.get("live_room_id")
+                if room_id:
+                    try:
+                        live_detail = get_live_detail(log, room_id)
+                    except Exception as e:
+                        log.error(f"商品 {pid} 回放详情请求失败: {e}")
+
+            row = build_product(item, detail, shop_id, shop_name, live_detail)
+            if row:
+                batch.append(row)
+
+        if batch:
+            # 已存在（pid 唯一）则跳过：历史成交为终态，无需覆盖
+            sql_pool.insert_many(table="cgj_product_record", data_list=batch, ignore=True)
+            saved += len(batch)
+
+        log.info(f"商户 {shop_id} 历史成交第 {page + 1} 页完成，本页 {len(items)} 商品，新增 {new_on_page} 个")
+
+        # 增量早停：连续 STOP_AFTER_DUPE_PAGES 页都没有新商品 → 已翻到旧数据区，停止翻页
+        if incremental:
+            if new_on_page == 0:
+                dupe_pages += 1
+                if dupe_pages >= STOP_AFTER_DUPE_PAGES:
+                    log.info(f"商户 {shop_id} 连续 {dupe_pages} 页无新商品，增量早停")
+                    break
+            else:
+                dupe_pages = 0
+
+        if data.get("done") or not data.get("cursor"):
+            break
+        cursor = data["cursor"]
+        page += 1
+
+    return saved
+
+
+# ==================== 三、玩家采集（买家名单） ====================
+def get_player_page(log, product_id, cursor: str = "") -> dict | None:
+    """获取某商品玩家（买家）名单的一页 announcement_by_user。
+
+    Args:
+        log: 日志对象。
+        product_id (int | str): 商品 id。
+        cursor (str, optional): 翻页游标，首页传空串。Defaults to ""。
+
+    Returns:
+        dict | None: 响应 JSON（含 data.list / cursor / done）；失败由 cgj_request 抛异常触发重试。
+    """
+    return cgj_request(
+        log, "/api/product/announcement_by_user", method="GET",
+        params={"id": product_id, "cursor": cursor, "size": PLAYER_PAGE_SIZE},
+    )
+
+
+def save_players(product_id, items: list, sql_pool) -> int:
+    """解析玩家（买家）名单并写入 cgj_player_record。
+
+    ------------------------------------------------------------------
+    并同步 schema.sql 的 cgj_player_record 列。
+    ------------------------------------------------------------------
+
+    Args:
+        product_id (int | str): 商品 id（写入 pid 列）。
+        items (list[dict]): announcement_by_user 的 data.list，每项含 view_key/total/buyer_name/buyer_image_url。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本次写入的玩家记录数。
+    """
+    info_list = []
+    for item in items:
+        # print(item)
+        info_list.append({
+            "pid": product_id,  # 商品 id（管道必需，item 里没有）
+            "view_key": item.get("view_key"),  # 唯一标识
+            "total": item.get("total"),  # 份数
+            "buyer_name": item.get("buyer_name")  # 买家昵称（脱敏）
+        })
+        # print(info_list)
+
+    if info_list:
+        sql_pool.insert_many(table="cgj_player_record", data_list=info_list, ignore=True)
+    return len(info_list)
+
+
+def get_player_list(log, product_id, sql_pool) -> bool:
+    """遍历某商品玩家名单翻页并写库。
+
+    Args:
+        log: 日志对象。
+        product_id (int | str): 商品 id。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        bool: True 表示抓到玩家数据，False 表示无数据。
+    """
+    cursor = ""
+    page = 0
+    has_data = False
+
+    while page < MAX_PAGES:
+        try:
+            resp = get_player_page(log, product_id, cursor)
+        except Exception as e:
+            log.error(f"商品 {product_id} 玩家名单 cursor={cursor!r} 请求失败: {e}")
+            break
+
+        if not resp or resp.get("code") != 0:
+            log.info(f"商品 {product_id} 暂无玩家: {resp.get('msg') if resp else None}")
+            break
+
+        data = resp.get("data") or {}
+        items = data.get("list") or []
+        if not items:
+            log.info(f"商品 {product_id} 玩家名单翻页完成")
+            break
+
+        has_data = True
+        save_players(product_id, items, sql_pool)
+
+        if data.get("done") or not data.get("cursor"):
+            log.info(f"商品 {product_id} 玩家名单翻页完成")
+            break
+        cursor = data["cursor"]
+        page += 1
+
+    return has_data
+
+
+# ==================== 四、回放补采 ====================
+def refill_replays(log, sql_pool) -> int:
+    """补采回放地址：对已采玩家(player_state=1)但 replay_url 仍空的商品，
+    重取 live_room_id → live/user_live/detail → offline_address_info[0].url → 更新 replay_url。
+
+    覆盖场景：首次入库时直播还没结束 / 回放 VOD 还没生成，offline_address_info 为空、
+    replay_url 拿不到；直播结束、回放就绪后由本函数补回。库里已存 live_room_id 则直接用，
+    没有（如首采详情失败）则回退重取商品详情拿 room_id。replay_url 仍空则留到下一轮再试。
+
+    Args:
+        log: 日志对象。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本轮成功补采的回放数。
+    """
+    rows = sql_pool.select_all(
+        "SELECT pid, live_room_id FROM cgj_product_record WHERE replay_url IS NULL AND player_state = 1"
+    )
+    rows = rows or []
+    log.info(f"待补回放商品 {len(rows)} 个")
+    filled = 0
+    for pid, room_id in rows:
+        try:
+            # 库里没存 room_id（首采详情失败等）→ 回退重取详情
+            if not room_id:
+                detail = get_product_detail(log, pid)
+                room_id = detail.get("live_room_id")
+            if not room_id:
+                continue
+
+            live_detail = get_live_detail(log, room_id)
+            offline_list = live_detail.get("offline_address_info") or []
+            url = offline_list[0].get("url") if offline_list else None
+            live_start_time = live_detail.get("start_time")  # 开播时间
+            live_end_time = live_detail.get("end_time")  # 结束时间
+            if url:
+                sql_pool.update_one(
+                    "UPDATE cgj_product_record SET replay_url = %s, live_start_time = %s, live_end_time = %s WHERE pid = %s",
+                    (url, live_start_time, live_end_time, pid),
+                )
+                filled += 1
+                log.info(f"商品 {pid} 回放已补采 → {url}")
+        except Exception as e:
+            log.error(f"商品 {pid} 回放补采失败: {e}")
+    return filled
+
+
+# ==================== 主流程 ====================
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
+def cgj_main(log):
+    """潮谷街每日采集主函数：商户发现 → 商品采集(+详情) → 玩家采集。
+
+    Args:
+        log: 日志对象。
+
+    Raises:
+        RuntimeError: 数据库连接池异常时抛出以触发重试。
+    """
+    log.info(f"开始运行 {sys._getframe().f_code.co_name} 潮谷街采集任务" + "." * 40)
+
+    sql_pool = MySQLConnectionPool(log=log)
+    if not sql_pool.check_pool_health():
+        log.error("数据库连接池异常")
+        raise RuntimeError("数据库连接池异常")
+
+    try:
+        # 1) 商户发现
+        try:
+            n = get_shop_list(log, sql_pool)
+            log.info(f"商户发现完成，去重商户 {n} 个")
+        except Exception as e:
+            log.error(f"get_shop_list error: {e}")
+
+        time.sleep(5)
+
+        # 2) 商品采集：遍历库内所有商户的历史成交
+        try:
+            shop_rows = sql_pool.select_all("SELECT shop_id, shop_name FROM cgj_shop_record WHERE is_deleted = 0")
+            log.info(f"待采集商户 {len(shop_rows)} 个")
+            for shop_id, shop_name in shop_rows:
+                try:
+                    cnt = get_sold_list(log, shop_id, shop_name, sql_pool)
+                    log.info(f"商户 {shop_id} {shop_name} 商品采集完成，写入 {cnt} 个")
+                except Exception as e:
+                    log.error(f"get_sold_list error（商户 {shop_id}）: {e}")
+        except Exception as e:
+            log.error(f"iterate_shop_list error: {e}")
+
+        time.sleep(5)
+
+        # 3) 玩家采集：遍历尚未成功采集玩家的商品
+        try:
+            prod_rows = sql_pool.select_all("SELECT pid FROM cgj_product_record WHERE player_state != 1")
+            pids = [row[0] for row in prod_rows] if prod_rows else []
+            log.info(f"待采集玩家的商品 {len(pids)} 个")
+            for pid in pids:
+                try:
+                    # 先置 1 表示开始采集
+                    sql_pool.update_one("UPDATE cgj_product_record SET player_state = 1 WHERE pid = %s", (pid,))
+                    has_data = get_player_list(log, pid, sql_pool)
+                    if not has_data:
+                        # 无玩家置 2，下轮仍会重试
+                        sql_pool.update_one("UPDATE cgj_product_record SET player_state = 2 WHERE pid = %s", (pid,))
+                except Exception as pid_error:
+                    log.error(f"商品 {pid} 玩家采集失败: {pid_error}")
+                    try:
+                        sql_pool.update_one("UPDATE cgj_product_record SET player_state = 3 WHERE pid = %s", (pid,))
+                    except Exception as update_error:
+                        log.error(f"更新商品 {pid} 状态失败: {update_error}")
+        except Exception as e:
+            log.error(f"iterate_player_list error: {e}")
+
+        # 4) 回放补采：玩家采集之后，对已采玩家(player_state=1)但 replay_url 仍空的商品重取回放
+        try:
+            n = refill_replays(log, sql_pool)
+            log.info(f"回放补采完成，本轮 {n} 条")
+        except Exception as e:
+            log.error(f"refill_replays error: {e}")
+    except Exception as e:
+        log.error(f"{sys._getframe().f_code.co_name} error: {e}")
+    finally:
+        log.info(f"潮谷街采集 {sys._getframe().f_code.co_name} 运行结束，等待下一轮" + "." * 20)
+
+
+def schedule_task():
+    """定时任务入口：每天 00:01 运行一次 cgj_main。"""
+    # 立即运行一次（调试时取消注释）
+    # cgj_main(log=logger)
+
+    schedule.every().day.at("00:01").do(cgj_main, log=logger)
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+if __name__ == "__main__":
+    # cgj_main(logger)
+    schedule_task()

chaogujie_spider/cgj_history_spider.py

@@ -0,0 +1,100 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/07/06
+"""潮谷街历史全量采集脚本（一次性）。
+
+与 cgj_daily_spider（每天增量）的区别：本脚本对每个商户的历史成交「全量深翻」所有页
+（调 get_sold_list(incremental=False)，不走停止线早停），把历史商品 / 玩家一次性灌满。
+跑一次即止，不带 schedule。
+
+公共函数（请求 / 翻页 / 详情 / 玩家）全部从 cgj_daily_spider 复用，避免重复维护。
+
+运行：python cgj_history_spider.py
+"""
+import sys
+
+from loguru import logger
+from mysql_pool import MySQLConnectionPool
+
+from cgj_daily_spider import (
+    get_shop_list,
+    get_sold_list,
+    get_player_list,
+    refill_replays,
+)
+
+
+def history_main(log):
+    """历史全量采集主函数：商户发现 → 全量深翻商品(+详情) → 全量采玩家。
+
+    Args:
+        log: 日志对象。
+
+    Raises:
+        RuntimeError: 数据库连接池异常时抛出。
+    """
+    log.info(f"开始运行 {sys._getframe().f_code.co_name} 潮谷街历史全量采集" + "." * 40)
+
+    sql_pool = MySQLConnectionPool(log=log)
+    if not sql_pool.check_pool_health():
+        log.error("数据库连接池异常")
+        raise RuntimeError("数据库连接池异常")
+
+    try:
+        # 1) 商户发现：先用 shop 列表页把当前商户补进 cgj_shop_record（历史脚本也要完整商户清单）
+        try:
+            n = get_shop_list(log, sql_pool)
+            log.info(f"商户发现完成，去重商户 {n} 个")
+        except Exception as e:
+            log.error(f"get_shop_list error: {e}")
+
+        # 2) 商品全量：遍历所有商户，深翻历史成交全部页（incremental=False，不早停）
+        try:
+            shop_rows = sql_pool.select_all("SELECT shop_id, shop_name FROM cgj_shop_record WHERE is_deleted = 0")
+            log.info(f"待全量采集商户 {len(shop_rows)} 个")
+            for shop_id, shop_name in shop_rows:
+                try:
+                    cnt = get_sold_list(log, shop_id, shop_name, sql_pool, incremental=False)
+                    log.info(f"商户 {shop_id} {shop_name} 全量商品采集完成，写入 {cnt} 个")
+                except Exception as e:
+                    log.error(f"get_sold_list error（商户 {shop_id}）: {e}")
+        except Exception as e:
+            log.error(f"iterate_shop_list error: {e}")
+
+        # 3) 玩家全量：遍历尚未成功采集玩家的商品（player_state != 1）
+        try:
+            prod_rows = sql_pool.select_all("SELECT pid FROM cgj_product_record WHERE player_state != 1")
+            pids = [row[0] for row in prod_rows] if prod_rows else []
+            log.info(f"待采集玩家的商品 {len(pids)} 个")
+            for pid in pids:
+                try:
+                    # 先置 1 表示开始采集
+                    sql_pool.update_one("UPDATE cgj_product_record SET player_state = 1 WHERE pid = %s", (pid,))
+                    has_data = get_player_list(log, pid, sql_pool)
+                    if not has_data:
+                        # 无玩家置 2，下次仍会重试
+                        sql_pool.update_one("UPDATE cgj_product_record SET player_state = 2 WHERE pid = %s", (pid,))
+                except Exception as pid_error:
+                    log.error(f"商品 {pid} 玩家采集失败: {pid_error}")
+                    try:
+                        sql_pool.update_one("UPDATE cgj_product_record SET player_state = 3 WHERE pid = %s", (pid,))
+                    except Exception as update_error:
+                        log.error(f"更新商品 {pid} 状态失败: {update_error}")
+        except Exception as e:
+            log.error(f"iterate_player_list error: {e}")
+
+        # 4) 回放补采：对已采玩家(player_state=1)但 replay_url 仍空的商品重取回放
+        try:
+            n = refill_replays(log, sql_pool)
+            log.info(f"回放补采完成，本轮 {n} 条")
+        except Exception as e:
+            log.error(f"refill_replays error: {e}")
+    except Exception as e:
+        log.error(f"{sys._getframe().f_code.co_name} error: {e}")
+    finally:
+        log.info(f"潮谷街历史全量采集 {sys._getframe().f_code.co_name} 运行结束")
+
+
+if __name__ == "__main__":
+    history_main(logger)

chaogujie_spider/mysql_pool.py

@@ -0,0 +1,671 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2025/3/25 14:14
+import re
+import pymysql
+import YamlLoader
+from loguru import logger
+from dbutils.pooled_db import PooledDB
+
+# 获取yaml配置
+yaml = YamlLoader.readYaml()
+mysqlYaml = yaml.get("mysql")
+sql_host = mysqlYaml.getValueAsString("host")
+sql_port = mysqlYaml.getValueAsInt("port")
+sql_user = mysqlYaml.getValueAsString("username")
+sql_password = mysqlYaml.getValueAsString("password")
+sql_db = mysqlYaml.getValueAsString("db")
+
+
+class MySQLConnectionPool:
+    """
+    MySQL连接池
+    """
+
+    def __init__(self, mincached=1, maxcached=2, maxconnections=3, log=None):
+        """
+        初始化连接池
+        :param mincached: 初始化时，链接池中至少创建的链接，0表示不创建
+        :param maxcached: 池中空闲连接的最大数目（0 或 None 表示池大小不受限制）
+        :param maxconnections: 允许的最大连接数（0 或 None 表示任意数量的连接）
+        :param log: 自定义日志记录器
+        """
+        # 使用 loguru 的 logger，如果传入了其他 logger，则使用传入的 logger
+        self.log = log or logger
+        self.pool = PooledDB(
+            creator=pymysql,
+            mincached=mincached,
+            maxcached=maxcached,
+            maxconnections=maxconnections,
+            blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
+            host=sql_host,
+            port=sql_port,
+            user=sql_user,
+            password=sql_password,
+            database=sql_db,
+            ping=2,  # 每次执行前检查连接有效性，防止使用已断开的连接
+            connect_timeout=5,  # 连接超时时间（秒）
+            # read_timeout=30,  # 读取超时时间（秒）
+            write_timeout=30  # 写入超时时间（秒）
+        )
+
+    # def _execute(self, query, args=None, commit=False):
+    #     """
+    #     执行SQL
+    #     :param query: SQL语句
+    #     :param args: SQL参数
+    #     :param commit: 是否提交事务
+    #     :return: 查询结果
+    #     """
+    #     try:
+    #         with self.pool.connection() as conn:
+    #             with conn.cursor() as cursor:
+    #                 cursor.execute(query, args)
+    #                 if commit:
+    #                     conn.commit()
+    #                 self.log.debug(f"sql _execute, Query: {query}, Rows: {cursor.rowcount}")
+    #                 return cursor
+    #     except Exception as e:
+    #         if commit and conn:
+    #             conn.rollback()
+    #         self.log.exception(f"Error executing query: {e}, Query: {query}, Args: {args}")
+    #         raise e
+
+    def _execute(self, query, args=None, commit=False):
+        """
+        执行SQL（带断连重试）
+        :param query: SQL语句
+        :param args: SQL参数
+        :param commit: 是否提交事务
+        :return: 查询结果
+        """
+        conn = None
+        for attempt in range(2):  # 最多重试1次
+            try:
+                with self.pool.connection() as conn:
+                    with conn.cursor() as cursor:
+                        cursor.execute(query, args)
+                        if commit:
+                            conn.commit()
+                        self.log.debug(f"sql _execute, Query: {query}, Rows: {cursor.rowcount}")
+                        return cursor
+            except pymysql.err.InterfaceError as e:
+                # 连接已断开，重试一次
+                if attempt == 0:
+                    self.log.warning(f"数据库连接断开，正在重试... Error: {e}")
+                    continue
+                self.log.error(f"重试后仍失败: {e}, Query: {query}")
+                raise e
+            except pymysql.err.IntegrityError:
+                # 完整性错误（如重复条目）交由上层处理，避免在此打印完整堆栈污染日志
+                if commit and conn:
+                    try:
+                        conn.rollback()
+                    except Exception:
+                        pass
+                raise
+            except Exception as e:
+                if commit and conn:
+                    try:
+                        conn.rollback()
+                    except Exception:
+                        pass
+                self.log.exception(f"Error executing query: {e}, Query: {query}, Args: {args}")
+                raise e
+
+    def select_one(self, query, args=None):
+        """
+        执行查询，返回单个结果
+        :param query: 查询语句
+        :param args: 查询参数
+        :return: 查询结果
+        """
+        cursor = self._execute(query, args)
+        return cursor.fetchone()
+
+    def select_all(self, query, args=None):
+        """
+        执行查询，返回所有结果
+        :param query: 查询语句
+        :param args: 查询参数
+        :return: 查询结果
+        """
+        cursor = self._execute(query, args)
+        return cursor.fetchall()
+
+    def insert_one(self, query, args):
+        """
+        执行单条插入语句
+        :param query: 插入语句
+        :param args: 插入参数
+        """
+        self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data insert_one 入库中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+        cursor = self._execute(query, args, commit=True)
+        return cursor.lastrowid  # 返回插入的ID
+
+    def insert_all(self, query, args_list):
+        """
+        执行批量插入语句,如果失败则逐条插入
+        :param query: 插入语句
+        :param args_list: 插入参数列表
+        """
+        conn = None
+        cursor = None
+        try:
+            conn = self.pool.connection()
+            cursor = conn.cursor()
+            cursor.executemany(query, args_list)
+            conn.commit()
+            self.log.debug(f"sql insert_all, SQL: {query[:100]}..., Rows: {cursor.rowcount}")
+            self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data insert_all 入库中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+        except pymysql.err.IntegrityError as e:
+            if "Duplicate entry" in str(e):
+                conn.rollback()
+                self.log.warning(f"批量插入遇到重复,开始逐条插入。错误: {e}")
+                rowcount = 0
+                for args in args_list:
+                    try:
+                        self.insert_one(query, args)
+                        rowcount += 1
+                    except pymysql.err.IntegrityError as e2:
+                        if "Duplicate entry" in str(e2):
+                            self.log.debug(f"跳过重复条目: {e2}")
+                        else:
+                            self.log.error(f"插入失败: {e2}")
+                    except Exception as e2:
+                        self.log.error(f"插入失败: {e2}")
+                self.log.info(f"逐条插入完成: {rowcount}/{len(args_list)}条")
+            else:
+                conn.rollback()
+                self.log.exception(f"数据库完整性错误: {e}")
+                raise e
+        except Exception as e:
+            conn.rollback()
+            self.log.exception(f"批量插入失败: {e}")
+            raise e
+        finally:
+            if cursor:
+                cursor.close()
+            if conn:
+                conn.close()
+
+    def insert_one_or_dict(self, table=None, data=None, query=None, args=None, commit=True, ignore=False):
+        """
+        单条插入（支持字典或原始SQL）
+        :param table: 表名（字典插入时必需）
+        :param data: 字典数据 {列名: 值}
+        :param query: 直接SQL语句（与data二选一）
+        :param args: SQL参数（query使用时必需）
+        :param commit: 是否自动提交
+        :param ignore: 是否使用ignore
+        :return: 最后插入ID
+        """
+        if data is not None:
+            if not isinstance(data, dict):
+                raise ValueError("Data must be a dictionary")
+
+            keys = ', '.join([self._safe_identifier(k) for k in data.keys()])
+            values = ', '.join(['%s'] * len(data))
+
+            # 构建 INSERT IGNORE 语句
+            ignore_clause = "IGNORE" if ignore else ""
+            query = f"INSERT {ignore_clause} INTO {self._safe_identifier(table)} ({keys}) VALUES ({values})"
+            args = tuple(data.values())
+        elif query is None:
+            raise ValueError("Either data or query must be provided")
+
+        try:
+            cursor = self._execute(query, args, commit)
+            self.log.info(f"sql insert_one_or_dict, Table: {table}, Rows: {cursor.rowcount}")
+            self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data insert_one_or_dict 入库中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+            return cursor.lastrowid
+        except pymysql.err.IntegrityError as e:
+            if "Duplicate entry" in str(e):
+                # 重复条目用 warning 简短输出，不打印堆栈
+                self.log.warning(f"插入跳过-重复条目 Table: {table}, {e.args[1] if len(e.args) > 1 else e}")
+                return -1  # 返回 -1 表示重复条目被跳过
+            else:
+                self.log.error(f"数据库完整性错误 Table: {table}, Error: {e}")
+                raise
+        except Exception as e:
+            self.log.error(f"insert_one_or_dict 失败 Table: {table}, Error: {e}")
+            raise
+
+    def insert_many(self, table=None, data_list=None, query=None, args_list=None, batch_size=1000, commit=True,
+                    ignore=False):
+        """
+        批量插入（支持字典列表或原始SQL）
+        :param table: 表名（字典插入时必需）
+        :param data_list: 字典列表 [{列名: 值}]
+        :param query: 直接SQL语句（与data_list二选一）
+        :param args_list: SQL参数列表（query使用时必需）
+        :param batch_size: 分批大小
+        :param commit: 是否自动提交
+        :param ignore: 是否使用ignore
+        :return: 影响行数
+        """
+        if data_list is not None:
+            if not data_list or not isinstance(data_list[0], dict):
+                raise ValueError("Data_list must be a non-empty list of dictionaries")
+
+            keys = ', '.join([self._safe_identifier(k) for k in data_list[0].keys()])
+            values = ', '.join(['%s'] * len(data_list[0]))
+
+            # 构建 INSERT IGNORE 语句
+            ignore_clause = "IGNORE" if ignore else ""
+            query = f"INSERT {ignore_clause} INTO {self._safe_identifier(table)} ({keys}) VALUES ({values})"
+            args_list = [tuple(d.values()) for d in data_list]
+        elif query is None:
+            raise ValueError("Either data_list or query must be provided")
+
+        total = 0
+        for i in range(0, len(args_list), batch_size):
+            batch = args_list[i:i + batch_size]
+            try:
+                with self.pool.connection() as conn:
+                    with conn.cursor() as cursor:
+                        cursor.executemany(query, batch)
+                        if commit:
+                            conn.commit()
+                        total += cursor.rowcount
+            except pymysql.err.IntegrityError as e:
+                # 处理唯一索引冲突
+                if "Duplicate entry" in str(e):
+                    if ignore:
+                        # 如果使用了 INSERT IGNORE,理论上不会进这里,但以防万一
+                        self.log.warning(f"批量插入遇到重复条目(ignore模式): {e}")
+                    else:
+                        # 没有使用 IGNORE,降级为逐条插入
+                        self.log.warning(f"批量插入遇到重复条目,开始逐条插入。错误: {e}")
+                        if commit:
+                            conn.rollback()
+                        
+                        rowcount = 0
+                        for j, args in enumerate(batch):
+                            try:
+                                if data_list:
+                                    # 字典模式
+                                    self.insert_one_or_dict(
+                                        table=table,
+                                        data=dict(zip(data_list[0].keys(), args)),
+                                        commit=commit,
+                                        ignore=False  # 单条插入时手动捕获重复
+                                    )
+                                else:
+                                    # 原始SQL模式
+                                    self.insert_one(query, args)
+                                rowcount += 1
+                            except pymysql.err.IntegrityError as e2:
+                                if "Duplicate entry" in str(e2):
+                                    self.log.debug(f"跳过重复条目[{i+j+1}]: {e2}")
+                                else:
+                                    self.log.error(f"插入失败[{i+j+1}]: {e2}")
+                            except Exception as e2:
+                                self.log.error(f"插入失败[{i+j+1}]: {e2}")
+                        total += rowcount
+                        self.log.info(f"批次逐条插入完成: 成功{rowcount}/{len(batch)}条")
+                else:
+                    # 其他完整性错误
+                    self.log.exception(f"数据库完整性错误: {e}")
+                    if commit:
+                        conn.rollback()
+                    raise e
+            except Exception as e:
+                # 其他数据库错误
+                self.log.exception(f"批量插入失败: {e}")
+                if commit:
+                    conn.rollback()
+                raise e
+        if table:
+            self.log.info(f"sql insert_many, Table: {table}, Total Rows: {total}")
+        else:
+            self.log.info(f"sql insert_many, Query: {query}, Total Rows: {total}")
+        return total
+
+    def insert_many_two(self, table=None, data_list=None, query=None, args_list=None, batch_size=1000, commit=True,
+                        ignore=False):
+        """
+        批量插入(支持字典列表或原始SQL) - 备用方法
+        :param table: 表名(字典插入时必需)
+        :param data_list: 字典列表 [{列名: 值}]
+        :param query: 直接SQL语句(与data_list二选一)
+        :param args_list: SQL参数列表(query使用时必需)
+        :param batch_size: 分批大小
+        :param commit: 是否自动提交
+        :param ignore: 是否使用INSERT IGNORE
+        :return: 影响行数
+        """
+        if data_list is not None:
+            if not data_list or not isinstance(data_list[0], dict):
+                raise ValueError("Data_list must be a non-empty list of dictionaries")
+            keys = ', '.join([self._safe_identifier(k) for k in data_list[0].keys()])
+            values = ', '.join(['%s'] * len(data_list[0]))
+            ignore_clause = "IGNORE" if ignore else ""
+            query = f"INSERT {ignore_clause} INTO {self._safe_identifier(table)} ({keys}) VALUES ({values})"
+            args_list = [tuple(d.values()) for d in data_list]
+        elif query is None:
+            raise ValueError("Either data_list or query must be provided")
+    
+        total = 0
+        for i in range(0, len(args_list), batch_size):
+            batch = args_list[i:i + batch_size]
+            try:
+                with self.pool.connection() as conn:
+                    with conn.cursor() as cursor:
+                        cursor.executemany(query, batch)
+                        if commit:
+                            conn.commit()
+                        total += cursor.rowcount
+            except pymysql.err.IntegrityError as e:
+                if "Duplicate entry" in str(e) and not ignore:
+                    self.log.warning(f"批量插入遇到重复,降级为逐条插入: {e}")
+                    if commit:
+                        conn.rollback()
+                    rowcount = 0
+                    for args in batch:
+                        try:
+                            self.insert_one(query, args)
+                            rowcount += 1
+                        except pymysql.err.IntegrityError as e2:
+                            if "Duplicate entry" in str(e2):
+                                self.log.debug(f"跳过重复条目: {e2}")
+                            else:
+                                self.log.error(f"插入失败: {e2}")
+                        except Exception as e2:
+                            self.log.error(f"插入失败: {e2}")
+                    total += rowcount
+                else:
+                    self.log.exception(f"数据库完整性错误: {e}")
+                    if commit:
+                        conn.rollback()
+                    raise e
+            except Exception as e:
+                self.log.exception(f"批量插入失败: {e}")
+                if commit:
+                    conn.rollback()
+                raise e
+        self.log.info(f"sql insert_many_two, Table: {table}, Total Rows: {total}")
+        return total
+
+    def insert_too_many(self, query, args_list, batch_size=1000):
+        """
+        执行批量插入语句，分片提交, 单次插入大于十万+时可用, 如果失败则降级为逐条插入
+        :param query: 插入语句
+        :param args_list: 插入参数列表
+        :param batch_size: 每次插入的条数
+        """
+        self.log.info(f"sql insert_too_many, Query: {query}, Total Rows: {len(args_list)}")
+        for i in range(0, len(args_list), batch_size):
+            batch = args_list[i:i + batch_size]
+            try:
+                with self.pool.connection() as conn:
+                    with conn.cursor() as cursor:
+                        cursor.executemany(query, batch)
+                        conn.commit()
+                        self.log.debug(f"insert_too_many -> Total Rows: {len(batch)}")
+            except Exception as e:
+                self.log.error(f"insert_too_many error. Trying single insert. Error: {e}")
+                # 当前批次降级为单条插入
+                for args in batch:
+                    self.insert_one(query, args)
+
+    def update_one(self, query, args):
+        """
+        执行单条更新语句
+        :param query: 更新语句
+        :param args: 更新参数
+        """
+        self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data update_one 更新中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+        return self._execute(query, args, commit=True)
+
+    def update_all(self, query, args_list):
+        """
+        执行批量更新语句，如果失败则逐条更新
+        :param query: 更新语句
+        :param args_list: 更新参数列表
+        """
+        conn = None
+        cursor = None
+        try:
+            conn = self.pool.connection()
+            cursor = conn.cursor()
+            cursor.executemany(query, args_list)
+            conn.commit()
+            self.log.debug(f"sql update_all, SQL: {query}, Rows: {len(args_list)}")
+            self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data update_all 更新中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+        except Exception as e:
+            conn.rollback()
+            self.log.error(f"Error executing query: {e}")
+            # 如果批量更新失败，则逐条更新
+            rowcount = 0
+            for args in args_list:
+                self.update_one(query, args)
+                rowcount += 1
+            self.log.debug(f'Batch update failed. Updated {rowcount} rows individually.')
+        finally:
+            if cursor:
+                cursor.close()
+            if conn:
+                conn.close()
+
+    def update_one_or_dict(self, table=None, data=None, condition=None, query=None, args=None, commit=True):
+        """
+        单条更新（支持字典或原始SQL）
+        :param table: 表名（字典模式必需）
+        :param data: 字典数据 {列名: 值}（与 query 二选一）
+        :param condition: 更新条件，支持以下格式：
+            - 字典: {"id": 1} → "WHERE id = %s"
+            - 字符串: "id = 1" → "WHERE id = 1"（需自行确保安全）
+            - 元组: ("id = %s", [1]) → "WHERE id = %s"（参数化查询）
+        :param query: 直接SQL语句（与 data 二选一）
+        :param args: SQL参数（query 模式下必需）
+        :param commit: 是否自动提交
+        :return: 影响行数
+        :raises: ValueError 参数校验失败时抛出
+        """
+        # 参数校验
+        if data is not None:
+            if not isinstance(data, dict):
+                raise ValueError("Data must be a dictionary")
+            if table is None:
+                raise ValueError("Table name is required for dictionary update")
+            if condition is None:
+                raise ValueError("Condition is required for dictionary update")
+
+            # 构建 SET 子句
+            set_clause = ", ".join([f"{self._safe_identifier(k)} = %s" for k in data.keys()])
+            set_values = list(data.values())
+
+            # 解析条件
+            condition_clause, condition_args = self._parse_condition(condition)
+            query = f"UPDATE {self._safe_identifier(table)} SET {set_clause} WHERE {condition_clause}"
+            args = set_values + condition_args
+
+        elif query is None:
+            raise ValueError("Either data or query must be provided")
+
+        # 执行更新
+        cursor = self._execute(query, args, commit)
+        # self.log.debug(
+        #     f"Updated table={table}, rows={cursor.rowcount}, query={query[:100]}...",
+        #     extra={"table": table, "rows": cursor.rowcount}
+        # )
+        return cursor.rowcount
+
+    def _parse_condition(self, condition):
+        """
+        解析条件为 (clause, args) 格式
+        :param condition: 字典/字符串/元组
+        :return: (str, list) SQL 子句和参数列表
+        """
+        if isinstance(condition, dict):
+            clause = " AND ".join([f"{self._safe_identifier(k)} = %s" for k in condition.keys()])
+            args = list(condition.values())
+        elif isinstance(condition, str):
+            clause = condition  # 注意：需调用方确保安全
+            args = []
+        elif isinstance(condition, (tuple, list)) and len(condition) == 2:
+            clause, args = condition[0], condition[1]
+            if not isinstance(args, (list, tuple)):
+                args = [args]
+        else:
+            raise ValueError("Condition must be dict/str/(clause, args)")
+        return clause, args
+
+    def update_many(self, table=None, data_list=None, condition_list=None, query=None, args_list=None, batch_size=500,
+                    commit=True):
+        """
+        批量更新（支持字典列表或原始SQL）
+        :param table: 表名（字典插入时必需）
+        :param data_list: 字典列表 [{列名: 值}]
+        :param condition_list: 条件列表（必须为字典，与data_list等长）
+        :param query: 直接SQL语句（与data_list二选一）
+        :param args_list: SQL参数列表（query使用时必需）
+        :param batch_size: 分批大小
+        :param commit: 是否自动提交
+        :return: 影响行数
+        """
+        if data_list is not None:
+            if not data_list or not isinstance(data_list[0], dict):
+                raise ValueError("Data_list must be a non-empty list of dictionaries")
+            if condition_list is None or len(data_list) != len(condition_list):
+                raise ValueError("Condition_list must be provided and match the length of data_list")
+            if not all(isinstance(cond, dict) for cond in condition_list):
+                raise ValueError("All elements in condition_list must be dictionaries")
+
+            # 获取第一个数据项和条件项的键
+            first_data_keys = set(data_list[0].keys())
+            first_cond_keys = set(condition_list[0].keys())
+
+            # 构造基础SQL
+            set_clause = ', '.join([self._safe_identifier(k) + ' = %s' for k in data_list[0].keys()])
+            condition_clause = ' AND '.join([self._safe_identifier(k) + ' = %s' for k in condition_list[0].keys()])
+            base_query = f"UPDATE {self._safe_identifier(table)} SET {set_clause} WHERE {condition_clause}"
+            total = 0
+
+            # 分批次处理
+            for i in range(0, len(data_list), batch_size):
+                batch_data = data_list[i:i + batch_size]
+                batch_conds = condition_list[i:i + batch_size]
+                batch_args = []
+
+                # 检查当前批次的结构是否一致
+                can_batch = True
+                for data, cond in zip(batch_data, batch_conds):
+                    data_keys = set(data.keys())
+                    cond_keys = set(cond.keys())
+                    if data_keys != first_data_keys or cond_keys != first_cond_keys:
+                        can_batch = False
+                        break
+                    batch_args.append(tuple(data.values()) + tuple(cond.values()))
+
+                if not can_batch:
+                    # 结构不一致，转为单条更新
+                    for data, cond in zip(batch_data, batch_conds):
+                        self.update_one_or_dict(table=table, data=data, condition=cond, commit=commit)
+                        total += 1
+                    continue
+
+                # 执行批量更新
+                try:
+                    with self.pool.connection() as conn:
+                        with conn.cursor() as cursor:
+                            cursor.executemany(base_query, batch_args)
+                            if commit:
+                                conn.commit()
+                            total += cursor.rowcount
+                            self.log.debug(f"Batch update succeeded. Rows: {cursor.rowcount}")
+                except Exception as e:
+                    if commit:
+                        conn.rollback()
+                    self.log.error(f"Batch update failed: {e}")
+                    # 降级为单条更新
+                    for args, data, cond in zip(batch_args, batch_data, batch_conds):
+                        try:
+                            self._execute(base_query, args, commit=commit)
+                            total += 1
+                        except Exception as e2:
+                            self.log.error(f"Single update failed: {e2}, Data: {data}, Condition: {cond}")
+            self.log.info(f"Total updated rows: {total}")
+            return total
+        elif query is not None:
+            # 处理原始SQL和参数列表
+            if args_list is None:
+                raise ValueError("args_list must be provided when using query")
+
+            total = 0
+            for i in range(0, len(args_list), batch_size):
+                batch_args = args_list[i:i + batch_size]
+                try:
+                    with self.pool.connection() as conn:
+                        with conn.cursor() as cursor:
+                            cursor.executemany(query, batch_args)
+                            if commit:
+                                conn.commit()
+                            total += cursor.rowcount
+                            self.log.debug(f"Batch update succeeded. Rows: {cursor.rowcount}")
+                except Exception as e:
+                    if commit:
+                        conn.rollback()
+                    self.log.error(f"Batch update failed: {e}")
+                    # 降级为单条更新
+                    for args in batch_args:
+                        try:
+                            self._execute(query, args, commit=commit)
+                            total += 1
+                        except Exception as e2:
+                            self.log.error(f"Single update failed: {e2}, Args: {args}")
+            self.log.info(f"Total updated rows: {total}")
+            return total
+        else:
+            raise ValueError("Either data_list or query must be provided")
+
+    def check_pool_health(self):
+        """
+        检查连接池中有效连接数
+
+        # 使用示例
+        # 配置 MySQL 连接池
+        sql_pool = MySQLConnectionPool(log=log)
+        if not sql_pool.check_pool_health():
+            log.error("数据库连接池异常")
+            raise RuntimeError("数据库连接池异常")
+        """
+        try:
+            with self.pool.connection() as conn:
+                conn.ping(reconnect=True)
+                return True
+        except Exception as e:
+            self.log.error(f"Connection pool health check failed: {e}")
+            return False
+
+    def close(self):
+        """
+        关闭连接池，释放所有连接
+        """
+        try:
+            if hasattr(self, 'pool') and self.pool:
+                self.pool.close()
+                self.log.info("数据库连接池已关闭")
+        except Exception as e:
+            self.log.error(f"关闭连接池失败: {e}")
+
+    @staticmethod
+    def _safe_identifier(name):
+        """SQL标识符安全校验"""
+        if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', name):
+            raise ValueError(f"Invalid SQL identifier: {name}")
+        return name
+
+
+if __name__ == '__main__':
+    sql_pool = MySQLConnectionPool()
+    data_dic = {'card_type_id': 111, 'card_type_name': '补充包 继承的意志【OPC-13】', 'card_type_position': 964,
+                'card_id': 5284, 'card_name': '蒙奇·D·路飞', 'card_number': 'OP13-001', 'card_rarity': 'L',
+                'card_img': 'https://source.windoent.com/OnePiecePc/Picture/1757929283612OP13-001.png',
+                'card_life': '4', 'card_attribute': '打', 'card_power': '5000', 'card_attack': '-',
+                'card_color': '红/绿', 'subscript': 4, 'card_features': '超新星/草帽一伙',
+                'card_text_desc': '【咚!!×1】【对方的攻击时】我方处于活跃状态的咚!!不多于5张的场合，可以将我方任意张数的咚!!转为休息状态。每有1张转为休息状态的咚!!，本次战斗中，此领袖或我方最多1张拥有《草帽一伙》特征的角色力量+2000。',
+                'card_offer_type': '补充包 继承的意志【OPC-13】', 'crawler_language': '简中'}
+    sql_pool.insert_one_or_dict(table="one_piece_record", data=data_dic)

chaogujie_spider/requirements.txt

@@ -0,0 +1,8 @@
+-i https://mirrors.aliyun.com/pypi/simple/
+DBUtils==3.1.2
+loguru==0.7.3
+PyMySQL==1.1.2
+PyYAML==6.0.3
+requests==2.33.1
+schedule==1.2.2
+tenacity==9.1.4