feat(mysql): 新增MySQL连接池及配置支持

- 添加 application.yml，支持MySQL数据库配置参数
- 实现 MySQLConnectionPool 类，提供连接池管理和SQL执行功能
- 支持单条及批量插入，更新，查询操作，带重试和错误处理
- 实现插入忽略重复条目和批量插入失败逐条降级逻辑
- 支持字典式和原始SQL两种数据操作方式
- 提供连接池健康检查和关闭功能
- 添加基础使用示例代码
- 新增README介绍项目结构、目标数据、运行方式及代理配置
- 实现scp_core公共模块，用于抓取并解析SCP Auctions拍卖会数据，包含代理和重试机制

scp_spider/README.md

@@ -0,0 +1,120 @@
+# SCP Auctions 拍卖数据爬虫
+
+抓取 **SCP Auctions**（[catalogs.scpauctions.com](https://catalogs.scpauctions.com/auctions/past)）历史拍卖会的已售/流拍拍品数据，含拍卖会信息、拍品列表与拍品详情。提供 **历史全量** 与 **每周新增** 两个任务。
+
+- URL：https://catalogs.scpauctions.com/auctions/past
+- 平台：**Bidsquare** 托管（Apache + 服务端渲染 SSR，无 Cloudflare）
+- 语言 / 环境：Python 3.12.10
+- 请求 / 解析：`curl_cffi`（浏览器指纹）+ `parsel`
+
+---
+
+## 1. 目标数据
+
+| 层级 | 来源页 | 抓取字段 |
+|---|---|---|
+| 拍卖会 | `/auctions/past` | 标题、类型、状态、开始/结束时间、主页链接、图录链接、封面图、**event_id（去重）** |
+| 拍品列表 | 图录页 `catalog` | Lot 号、标题、详情链接、缩略图、出价数、成交状态、成交价、**item_id（去重）** |
+| 拍品详情 | `/online-auctions/...` | 标题、成交价、成交状态、出价数、Collection、Sport、Type、Athlete、Team、多图 |
+
+---
+
+## 2. 站点分析结论（为什么这么抓）
+
+> 用户要求「优先从接口抓取」。经浏览器抓包确认：**本站为 Bidsquare 纯 SSR 页面，数据全部内嵌在 HTML 中，无独立数据 API**（XHR 仅为 GA/埋点）。因此采用 `curl_cffi` 直接 GET 页面 + `parsel` 解析，无 Cloudflare 挑战，响应 200 纯 HTML。
+
+- **去重键**
+  - 拍卖会：`event_id`（拍卖会卡片 `data-event_id`，如 `21706`，也出现在 URL 尾部）
+  - 拍品：`item_id`（lot 卡片 `data-item_id`，如 `9317607`）
+- **分页**
+  - 拍卖会列表：`/auctions/past?page=N`，每页 **24** 场，翻到空页为止
+  - 图录（lot 列表）：`{catalog_url}?page=N&limit=120`，`limit` 上限 **120**，翻到空页为止
+- **详情页属性区字段数不固定**：`.item-attributes li`（`<label>键</label><span>值</span>`）可能是 2~5 个，Athlete/Team 常缺失 → 先建 `label→value` 字典，再按需取值，缺失置 `None`。
+- **多图归一化**：缩略图 `s1.img.bidsquare.com/item/s/...` 与大图 `/item/l/...` 并存，统一替换成 `/item/l/`（large）后去重，逗号拼接入库。
+- **成交状态**：列表卡片底部 `9 BidsSold for$90,000` / `24 BidsUnsold`；详情页 `.bidding-price` 有金额为 `sold`，为空为 `unsold`。
+
+---
+
+## 3. 文件结构
+
+```
+2026-07-15(scp_spider)/
+├── application.yml      # MySQL 配置（mysql_pool 从运行目录读取）
+├── create_table.sql     # 建表：scp_auction（场次）+ scp_lot（拍品）
+├── scp_core.py          # 公用模块：HTTP/代理、列表解析、分页抓取、详情解析、入库
+├── scp_history.py       # 任务一：历史全量（一次性）
+├── scp_spider.py        # 任务二：每周新增（周调度常驻）
+└── logs/                # 运行日志（loguru 按天切分，保留 7 天）
+```
+
+### 数据表
+
+- **`scp_auction`**：一条 = 一场拍卖会，`event_id` 唯一索引。
+- **`scp_lot`**：一条 = 一个拍品，`item_id` 唯一索引；字段分两阶段写入。
+
+两阶段设计：
+- **阶段一（列表）**：抓拍卖会 + lot 列表入库，`scp_lot.state=0`（含成交价/状态/出价，列表页即有）。
+- **阶段二（详情）**：扫库 `state != 1` 的 lot，逐条进详情页补 Collection/Sport/Type/Athlete/Team/多图，成功置 `state=1`，失败置 `2`。
+
+---
+
+## 4. 使用方式
+
+### 4.1 前置
+
+1. 安装依赖：`curl_cffi`、`parsel`、`loguru`、`tenacity`、`schedule`，以及全局公共库 `charley-utils`（提供 `mysql_pool`）。
+2. 建表：执行 `create_table.sql`（DDL 写操作，需人工确认执行）。
+3. 配置 `application.yml` 中的 MySQL 连接。
+4. **代理**：本地开了 VPN，从大陆直连站点不通，请求必须走代理。见下方「代理配置」。
+
+### 4.2 运行
+
+```bash
+# 任务一：历史全量（初始化数据库时跑一次）
+python scp_history.py
+
+# 任务二：每周新增（常驻，先立即跑一次，之后每周一 05:00）
+python scp_spider.py
+```
+
+- 每周新增逻辑：GET `/auctions/past` 首页 → 与库中 `event_id` 求差集 → **仅新增拍卖会才入库查询**；随后补跑阶段二，兜底上一轮失败/中断的详情。
+
+---
+
+## 5. 代理配置
+
+代理集中在 `scp_core.py` 的 `get_proxys()`（带重试）。默认使用住宅代理（北美出口，与 SCP 面向的美区一致）：
+
+```python
+http_proxy = "http://账号:密码@proxy.123proxy.cn:36927"
+# 若改用本地 VPN 客户端 HTTP 端口（如 Clash），改成：
+# http_proxy = https_proxy = "http://127.0.0.1:7890"
+```
+
+切换本地 VPN 端口时，把 `get_proxys()` 内两行替换为对应地址即可，其余代码无需改动。
+
+---
+
+## 6. 反爬与稳健性
+
+- **无 Cloudflare / 无 JS 挑战**：`curl_cffi` 随机浏览器指纹（`impersonate`）+ 通用请求头即可，未写死 UA（避免与 TLS 指纹矛盾）。
+- **重试**：页面 GET、代理获取均由 `tenacity` 包裹重试；周调度主函数失败按小时重试。
+- **幂等**：`insert_many(ignore=True)` 依赖唯一索引去重，重复跑不会产生脏数据。
+- **限频**：站点为静态页，当前未加显式 sleep；如遇风控可在翻页/详情循环处补 `time.sleep`。
+
+---
+
+## 7. 踩坑记录
+
+- **优先接口落空**：抓包确认本站无数据接口，全部 SSR，最终走 HTML 解析。
+- **详情属性字段数量不一致**（2~5 个、可能缺 Athlete/Team）：改为按 `label` 建字典按需取，避免按固定顺序取值错位。
+- **多图有 small/large 两套**：统一归一化到 large 再去重，避免同图重复入库。
+- **时间带时区名**（`EDT`/`EST`）：解析时剥离时区名，保留东部墙钟时间入 `datetime` 字段，原文另存 `*_raw` 字段。
+- **控制台中文/en-dash 乱码**：Windows GBK 终端显示问题，实际数据 UTF-8 正确，入 `utf8mb4` 库无碍。
+
+---
+
+## 8. 举一反三
+
+- 本套「拍卖会 → 图录列表 → 拍品详情」三级 + 两阶段（列表/详情）+ `event_id`/`item_id` 去重 + 首页差集增量，是 Bidsquare 系托管拍卖站的通用范式，换其他 Bidsquare 站点只需调整域名与少量选择器。
+- 若某站改为前端渲染（数据走 XHR/JSON），优先复用「浏览器抓包定位接口 → 直接请求接口」路径，比解析 HTML 更稳。

scp_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)

scp_spider/application.yml

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

scp_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)

scp_spider/scp_core.py

@@ -0,0 +1,561 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/07/15
+"""
+SCP Auctions (catalogs.scpauctions.com) 公用模块：HTTP 配置、拍卖会列表解析、
+场次内 lot 列表分页抓取、详情页解析。被 scp_history.py / scp_spider.py 复用。
+
+目标网站: https://catalogs.scpauctions.com/auctions/past
+逻辑要点:
+    1. 站点为 Bidsquare 托管平台，页面服务端渲染（SSR），无独立数据 API，
+       全部数据直接在 HTML 里，用 parsel 解析即可（本地开了 VPN，请求统一走代理）。
+    2. 拍卖会列表页 /auctions/past 分页：?page=N（每页 24 场），翻到空页为止。
+       每张拍卖会卡片是 <div class="gtm-visible_event" data-event_id=...>：
+         event_id（去重键）、标题、类型、起止时间、主页/图录链接、封面图。
+    3. 图录（lot 列表）页 catalog_url?page=N&limit=120（limit 上限 120）：
+       lot 卡片是 <div id="stl-{item_id}" data-item_id=...>，含 Lot 号 / 标题 /
+       详情链接 / 缩略图 / 出价数 / 成交状态与成交价。翻到空页为止。
+    4. lot 详情页 /online-auctions/{house}/{slug}-{item_id}：
+         标题取 og:title；成交价在 .bidding-price（流拍则为空）；
+         属性（Collection/Sport/Type/Athlete/Team 等）在 .item-attributes li，
+           label/span 成对，字段数量不固定（2~5 个，可能缺 Athlete/Team），
+           故按 label 建字典再按需取值，缺失置空；
+         多图为 s1.img.bidsquare.com/item/{l|s}/... 归一化到 large 尺寸去重。
+"""
+import re
+import random
+from datetime import datetime
+
+from loguru import logger
+from parsel import Selector
+from curl_cffi import requests
+from curl_cffi.requests import BrowserType
+from urllib.parse import urljoin
+from tenacity import retry, stop_after_attempt, wait_fixed
+
+# —— 站点常量 ——
+BASE_URL = "https://catalogs.scpauctions.com"  # 站点根，用于拼接相对链接
+PAST_URL = "https://catalogs.scpauctions.com/auctions/past"  # 拍卖会（历史）列表页
+CATALOG_PAGE_SIZE = 120  # 图录页每页 lot 数（limit 参数，站点上限 120）
+IMG_ITEM_RE = r'https://s1\.img\.bidsquare\.com/item/[ls]/\d+/\d+\.jpe?g(?:\?t=[^"\'\s\\]+)?'  # 拍品图 URL
+
+# 直接用库内置的所有浏览器指纹（伪装真实浏览器 TLS/UA，规避潜在的指纹风控）
+client_identifier_list = [b.value for b in BrowserType]
+
+# 不写死 user-agent：curl_cffi 会按 impersonate 指纹自动带匹配的 UA，
+# 写死会造成 TLS 指纹与 UA 头矛盾，反而更易被识别。只保留通用、不冲突的头。
+headers = {
+    "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
+    "accept-language": "en-US,en;q=0.9",
+}
+
+
+def after_log(retry_state):
+    """tenacity retry 回调，统一打印重试日志。
+
+    Args:
+        retry_state: tenacity.RetryCallState，retry 框架自动传入。
+
+    Returns:
+        None: 仅打印日志，无返回值。
+    """
+    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(2), after=after_log)
+def get_proxys(log):
+    """获取代理字典，全部请求方法的 proxies 参数都走这里。
+
+    本地开了 VPN，从大陆直连站点不通，请求必须走代理。默认用住宅代理（北美出口，
+    与 SCP 面向的美区一致）。若要改用本地 VPN 客户端的 HTTP 代理端口（如 Clash 的
+    127.0.0.1:7890），把 http_proxy / https_proxy 换成对应地址即可。
+
+    Args:
+        log: logger 对象。
+
+    Returns:
+        dict: requests 风格代理字典 {"http": ..., "https": ...}。
+
+    Raises:
+        Exception: 透传内部异常以便 tenacity 触发重试。
+    """
+    # 住宅 ip 池 北美（与参考项目 rea_spider 一致）；如需切本地 VPN 端口在此改
+    http_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36927"
+    https_proxy = "http://u1952150085001297:sJMHl4qc4bM0@proxy.123proxy.cn:36927"
+    # 本地 VPN（Clash）示例，二选一：
+    # http_proxy = https_proxy = "http://127.0.0.1:7890"
+
+    try:
+        return {
+            "http": http_proxy,
+            "https": https_proxy,
+        }
+    except Exception as e:
+        log.error(f"Error getting proxy: {e}")
+        raise e
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
+def _get_selector(log, session, impersonate, url):
+    """GET 一个页面并返回 parsel Selector（带重试，走代理）。
+
+    Args:
+        log: logger 对象。
+        session (requests.Session): curl_cffi 会话对象。
+        impersonate (str): 浏览器指纹标识。
+        url (str): 目标绝对 URL。
+
+    Returns:
+        Selector: 响应 HTML 的 parsel 解析对象。
+
+    Raises:
+        Exception: HTTP 状态非 2xx 时透传异常以触发重试。
+    """
+    resp = session.get(url, headers=headers, impersonate=impersonate,
+                       proxies=get_proxys(log), timeout=30)
+    resp.raise_for_status()
+    return Selector(resp.text)
+
+
+def _parse_datetime(raw):
+    """把站点时间原文解析成 datetime（丢弃时区名，保留东部墙钟时间）。
+
+    Args:
+        raw (str): 时间原文，如 "May 20, 2026 01:00PM EDT" 或带 "Start:" 前缀。
+
+    Returns:
+        datetime | None: 解析成功返回 datetime；失败或空返回 None。
+    """
+    if not raw:
+        return None
+    # 去掉 Start:/End: 前缀与末尾时区名（EDT/EST/…），只留 "May 20, 2026 01:00PM"
+    text = re.sub(r'^\s*(Start|End)\s*:\s*', '', raw.strip(), flags=re.I)
+    m = re.search(r'([A-Za-z]{3,9}\s+\d{1,2},\s*\d{4}\s+\d{1,2}:\d{2}\s*[AP]M)', text, re.I)
+    if not m:
+        return None
+    try:
+        return datetime.strptime(m.group(1).replace("  ", " ").upper(), "%b %d, %Y %I:%M%p")
+    except ValueError:
+        return None
+
+
+def parse_auction_list(selector):
+    """解析 /auctions/past 单页的拍卖会卡片列表。
+
+    Args:
+        selector (Selector): 拍卖会列表页某一页 GET 响应的 parsel 解析对象。
+
+    Returns:
+        list[dict]: 每场一个 dict，字段：event_id, auction_name, auction_type,
+            event_status, start_time_raw, end_time_raw, start_time, end_time,
+            auction_url, catalog_url, cover_img；无卡片返回空列表。
+    """
+    auctions = []
+    for card in selector.css('div.gtm-visible_event[data-event_id]'):
+        event_id = card.attrib.get("data-event_id", "").strip()
+        if not event_id:
+            continue
+
+        # 标题 + 拍卖会主页链接
+        title_a = card.css('.list-top h1 a')
+        auction_name = (title_a.xpath('normalize-space(.)').get() or
+                        card.attrib.get("data-event_name", "")).strip()
+        auction_url = title_a.attrib.get("href", "").strip()
+
+        # 类型（Timed Auction 等）
+        auction_type = (card.css('.list-top label::text').get() or "").strip()
+
+        # 起止时间：.list-top 下多个 <span>，按前缀区分
+        start_raw = end_raw = ""
+        for sp in card.css('.list-top span'):
+            t = (sp.xpath('normalize-space(.)').get() or "").strip()
+            if t.lower().startswith("start"):
+                start_raw = t
+            elif t.lower().startswith("end"):
+                end_raw = t
+
+        # 图录链接（去掉锚点 #catalog）
+        catalog_url = (card.css('a.view-catalog-btn::attr(href)').get() or "").strip()
+        catalog_url = catalog_url.split("#")[0]
+
+        cover_img = (card.css('.slider-for img::attr(src)').get() or "").strip()
+
+        start_dt = _parse_datetime(start_raw)
+        end_dt = _parse_datetime(end_raw)
+        auctions.append({
+            "event_id": event_id,
+            "auction_name": auction_name,
+            "auction_type": auction_type,
+            "event_status": card.attrib.get("data-event_status", "").strip(),
+            "start_time_raw": re.sub(r'^\s*Start\s*:\s*', '', start_raw, flags=re.I).strip(),
+            "end_time_raw": re.sub(r'^\s*End\s*:\s*', '', end_raw, flags=re.I).strip(),
+            "start_time": start_dt.strftime("%Y-%m-%d %H:%M:%S") if start_dt else None,
+            "end_time": end_dt.strftime("%Y-%m-%d %H:%M:%S") if end_dt else None,
+            "auction_url": urljoin(BASE_URL, auction_url) if auction_url else None,
+            "catalog_url": urljoin(BASE_URL, catalog_url) if catalog_url else None,
+            "cover_img": cover_img or None,
+        })
+    return auctions
+
+
+@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
+def get_auction_list(log, session, impersonate, only_first_page=False):
+    """翻页抓取 /auctions/past 全部（或仅首页）拍卖会。
+
+    Args:
+        log: logger 对象。
+        session (requests.Session): curl_cffi 会话对象。
+        impersonate (str): 浏览器指纹标识。
+        only_first_page (bool, optional): True 仅取首页（增量用，最近场次在首页顶部）；
+            False 翻完全部页（全量用）。Defaults to False。
+
+    Returns:
+        list[dict]: 拍卖会 dict 列表，已按 event_id 去重。
+    """
+    scope = "首页" if only_first_page else "全部"
+    log.info(f"获取{scope}拍卖会列表")
+    result, seen = [], set()
+    page = 1
+    while True:
+        url = f"{PAST_URL}?page={page}"
+        sel = _get_selector(log, session, impersonate, url)
+        page_auctions = parse_auction_list(sel)
+        if not page_auctions:
+            break  # 空页 = 已翻过末页
+        new = [a for a in page_auctions if a["event_id"] not in seen]
+        for a in new:
+            seen.add(a["event_id"])
+            result.append(a)
+        if not new:
+            break  # 本页全重复（越界回卷兜底），停止
+        log.info(f"  第 {page} 页解析到 {len(page_auctions)} 场（新增 {len(new)}，累计 {len(result)}）")
+        if only_first_page:
+            break
+        page += 1
+    log.info(f"共解析到 {len(result)} 场拍卖会")
+    return result
+
+
+def _clean_price(text):
+    """把成交价文本清洗成纯数字字符串。
+
+    Args:
+        text (str): 原始价格文本，如 "$90,000"。
+
+    Returns:
+        str: 去掉 $ 和千分位逗号后的数字串，如 "90000"；无价格返回空串。
+    """
+    if not text:
+        return ""
+    m = re.search(r'\$?\s*([\d,]+)', text)
+    return m.group(1).replace(",", "") if m else ""
+
+
+def parse_lot_cards(selector, auction):
+    """解析图录页单页的 lot 卡片。
+
+    Args:
+        selector (Selector): 图录页某一页 GET 响应的 parsel 解析对象。
+        auction (dict): 当前场次 dict（含 event_id / auction_name），回填到每条 lot。
+
+    Returns:
+        list[dict]: 每条 lot 一个 dict，字段：item_id, event_id, auction_name,
+            lot_number, title, detail_url, list_img, sold_status, sold_for, bids；
+            无卡片返回空列表。
+    """
+    rows = []
+    for card in selector.css('div[id^="stl-"][data-item_id]'):
+        item_id = card.attrib.get("data-item_id", "").strip()
+        if not item_id:
+            continue
+
+        title_a = card.css('.lot_title a')
+        title = (title_a.xpath('normalize-space(.)').get() or "").strip()
+        detail_url = (title_a.attrib.get("href") or "").strip()
+
+        # Lot 号："Lot 1" → "1"
+        lot_raw = (card.css('.lot_Num::text').get() or "").strip()
+        lot_number = (re.search(r'(\d+)', lot_raw) or [None, ""])[1] if lot_raw else ""
+
+        list_img = (card.css('.catalog_img img::attr(src)').get() or "").strip()
+
+        # 底部区："9 BidsSold for$90,000" / "24 BidsUnsold"
+        bottom = card.css('.item-list-bottom')
+        bottom_text = (bottom.xpath('normalize-space(.)').get() or "") if bottom else ""
+        bids_m = re.search(r'(\d+)\s*Bids?', bottom_text, re.I)
+        bids = bids_m.group(1) if bids_m else ""
+        if re.search(r'Unsold', bottom_text, re.I):
+            sold_status, sold_for = "unsold", ""
+        elif re.search(r'Sold\s*for', bottom_text, re.I):
+            sold_status = "sold"
+            sold_for = _clean_price(re.sub(r'.*Sold\s*for', '', bottom_text, flags=re.I))
+        else:
+            sold_status, sold_for = "", ""
+
+        rows.append({
+            "item_id": item_id,
+            "event_id": auction["event_id"],
+            "auction_name": auction.get("auction_name"),
+            "lot_number": lot_number,
+            "title": title,
+            "detail_url": urljoin(BASE_URL, detail_url) if detail_url else None,
+            "list_img": list_img or None,
+            "sold_status": sold_status or None,
+            "sold_for": sold_for or None,
+            "bids": bids or None,
+        })
+    return rows
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
+def _fetch_lot_page(log, session, impersonate, catalog_url, page):
+    """抓取图录页的某一页 lot 卡片。
+
+    Args:
+        log: logger 对象。
+        session (requests.Session): curl_cffi 会话对象。
+        impersonate (str): 浏览器指纹标识。
+        catalog_url (str): 图录页绝对 URL（不含 query）。
+        page (int): 页码，从 1 开始。
+
+    Returns:
+        Selector: 该页响应的 parsel 解析对象。
+    """
+    url = f"{catalog_url}?page={page}&limit={CATALOG_PAGE_SIZE}"
+    return _get_selector(log, session, impersonate, url)
+
+
+def fetch_auction_lots(log, session, impersonate, auction):
+    """抓取一个场次下的全部 lot 列表（翻页，去重 item_id）。
+
+    Args:
+        log: logger 对象。
+        session (requests.Session): curl_cffi 会话对象。
+        impersonate (str): 浏览器指纹标识。
+        auction (dict): 场次 dict（含 event_id / auction_name / catalog_url）。
+
+    Returns:
+        list[dict]: 该场次全部 lot 列表（阶段一字段，不含详情）。
+    """
+    catalog_url = auction.get("catalog_url")
+    if not catalog_url:
+        log.warning(f"场次 {auction['event_id']} 无 catalog_url，跳过")
+        return []
+
+    log.info(f"开始抓取场次 {auction['event_id']}({auction.get('auction_name')}) 的 lot 列表")
+    all_lots, seen = [], set()
+    page = 1
+    while True:
+        sel = _fetch_lot_page(log, session, impersonate, catalog_url, page)
+        lots = parse_lot_cards(sel, auction)
+        if not lots:
+            break  # 空页 = 已翻过末页
+        new = [x for x in lots if x["item_id"] not in seen]
+        for x in new:
+            seen.add(x["item_id"])
+        all_lots.extend(new)
+        log.info(f"  第 {page} 页 {len(lots)} 条（新增 {len(new)}，累计 {len(all_lots)}）")
+        if not new:
+            break  # 本页全重复，停止
+        page += 1
+    log.info(f"场次 {auction['event_id']} 共抓 {len(all_lots)} 条 lot")
+    return all_lots
+
+
+def parse_lot_detail(selector):
+    """解析 lot 详情页，抽取标题、成交价、属性、多图。
+
+    详情页属性区（.item-attributes）字段数量不固定：不同拍品可能是 2~5 个，
+    Athlete/Team 等常缺失。故先把所有 li 的 label→span 建成字典，再按需取字段，
+    缺失字段返回 None，保证解析健壮。
+
+    Args:
+        selector (Selector): 详情页 GET 响应的 parsel 解析对象。
+
+    Returns:
+        dict: {
+            "title": 标题, "sold_for": 成交价(纯数字串,流拍为空),
+            "sold_status": "sold"/"unsold", "bids": 出价数,
+            "collection", "sport", "item_type", "athlete", "team": 属性(缺失为 None),
+            "imgs": 多图 URL 逗号拼接
+        }。
+    """
+    # 标题：og:title 最干净（页面首个 h1 是拍卖会名，非拍品标题）
+    title = (selector.css('meta[property="og:title"]::attr(content)').get() or "").strip()
+    if not title:
+        # 兜底：取非拍卖会名的那个 h1
+        h1s = [t.strip() for t in selector.css('h1 ::text, h1::text').getall() if t.strip()]
+        title = h1s[1] if len(h1s) > 1 else (h1s[0] if h1s else "")
+
+    # 成交价：.bidding-price 内首个 $金额；流拍则为空
+    price_text = selector.css('.bidding-price').xpath('normalize-space(.)').get() or ""
+    sold_for = _clean_price(price_text)
+    sold_status = "sold" if sold_for else "unsold"
+
+    # 出价数：页面 "[9 Bids]"
+    body_text = selector.xpath('normalize-space(//body)').get() or ""
+    bids_m = re.search(r'\[?\s*(\d+)\s*Bids?\s*\]?', body_text, re.I)
+    bids = bids_m.group(1) if bids_m else ""
+
+    # 属性：.item-attributes li -> {label: value}，字段数量不固定，按需取
+    attrs = {}
+    for li in selector.css('.item-attributes li'):
+        label = (li.css('label::text').get() or "").strip().rstrip(":")
+        value = (li.css('span').xpath('normalize-space(.)').get() or "").strip()
+        if label:
+            attrs[label.lower()] = value
+
+    # 多图：/item/{l|s}/... 统一归一化到 large 尺寸并去重（保持出现顺序）
+    imgs, seen = [], set()
+    for u in selector.re(IMG_ITEM_RE):
+        large = u.replace("/item/s/", "/item/l/")
+        if large not in seen:
+            seen.add(large)
+            imgs.append(large)
+
+    return {
+        "title": title,
+        "sold_for": sold_for or None,
+        "sold_status": sold_status,
+        "bids": bids or None,
+        "collection": attrs.get("collection") or None,
+        "sport": attrs.get("sport") or None,
+        "item_type": attrs.get("type") or None,
+        "athlete": attrs.get("athlete") or None,
+        "team": attrs.get("team") or None,
+        "imgs": ",".join(imgs) if imgs else None,
+    }
+
+
+@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
+def fetch_lot_detail(log, session, impersonate, detail_url):
+    """GET lot 详情页并解析。
+
+    Args:
+        log: logger 对象。
+        session (requests.Session): curl_cffi 会话对象。
+        impersonate (str): 浏览器指纹标识。
+        detail_url (str): 详情页绝对 URL。
+
+    Returns:
+        dict: parse_lot_detail 的返回结果。
+    """
+    log.debug(f"获取详情 {detail_url}")
+    sel = _get_selector(log, session, impersonate, detail_url)
+    return parse_lot_detail(sel)
+
+
+def save_auction(log, sql_pool, auction):
+    """把一场拍卖会写入 scp_auction_record（幂等，靠 event_id 唯一索引去重）。
+
+    Args:
+        log: logger 对象。
+        sql_pool: MySQL 连接池；传 None 时不入库。
+        auction (dict): parse_auction_list 产出的场次 dict。
+
+    Returns:
+        None: 仅入库，无返回值。
+    """
+    if sql_pool is None:
+        return
+    row = {k: auction.get(k) for k in (
+        "event_id", "auction_name", "auction_type", "event_status",
+        "start_time_raw", "end_time_raw", "start_time", "end_time",
+        "auction_url", "catalog_url", "cover_img",
+    )}
+    sql_pool.insert_many(table="scp_auction_record", data_list=[row], ignore=True)
+
+
+def crawl_one_auction(log, sql_pool, session, impersonate, auction):
+    """抓取单个场次的全部 lot 列表并入库（阶段一：只抓列表，不进详情页）。
+
+    两阶段设计：本函数只负责 lot 列表入库（state 默认 0），详情字段由后续
+    update_details_for_pending 扫库 state != 1 的记录单独补抓。抓完把
+    scp_auction_record.lots_state 置 1，标记该场列表已抓。
+
+    Args:
+        log: logger 对象。
+        sql_pool: MySQL 连接池；传 None 时只返回数据，不入库。
+        session (requests.Session): curl_cffi 会话对象。
+        impersonate (str): 浏览器指纹标识。
+        auction (dict): 场次 dict（含 event_id / auction_name / catalog_url）。
+
+    Returns:
+        list[dict]: 该场次全部 lot 列表数据（阶段一字段，不含详情）。
+    """
+    lots = fetch_auction_lots(log, session, impersonate, auction)
+
+    if sql_pool is not None:
+        if lots:
+            sql_pool.insert_many(table="scp_lot_record", data_list=lots, ignore=True)
+        # 标记该场 lot 列表已抓
+        sql_pool.update_one_or_dict(
+            table="scp_auction_record",
+            data={"lots_state": 1},
+            condition={"event_id": auction["event_id"]},
+        )
+
+    log.info(f"场次 {auction['event_id']}({auction.get('auction_name')}) 共抓 {len(lots)} 条 lot")
+    return lots
+
+
+def get_details(log, detail_url, sql_pool, sql_id):
+    """对单条已入库 lot 补抓详情（阶段二），写回 scp_lot_record。
+
+    Args:
+        log: logger 对象。
+        detail_url (str): 详情页 URL。
+        sql_pool: MySQL 连接池。
+        sql_id: 数据库记录 id。
+
+    Returns:
+        None: 仅入库，无返回值。
+    """
+    log.info(f">>> 补抓详情 {detail_url}")
+    impersonate = random.choice(client_identifier_list)
+    with requests.Session() as session:
+        detail = fetch_lot_detail(log, session, impersonate, detail_url)
+
+    # 详情字段 + 标记已抓；成交价/状态/出价以详情页为准（覆盖列表阶段的值）
+    data = {**detail, "state": 1}
+    sql_pool.update_one_or_dict(
+        table="scp_lot_record",
+        data=data,
+        condition={"id": sql_id},
+    )
+
+
+def update_details_for_pending(log, sql_pool):
+    """扫库里 state != 1 的 lot，逐条补抓详情。
+
+    Args:
+        log: logger 对象。
+        sql_pool: MySQL 连接池。
+
+    Returns:
+        None: 仅入库，无返回值。
+    """
+    log.debug("Updating detail pages ...")
+    rows = sql_pool.select_all(
+        "select id, detail_url from scp_lot_record where state != 1"
+    )
+    for row in rows:
+        sql_id, detail_url = row[0], row[1]
+        try:
+            get_details(log, detail_url, sql_pool, sql_id)
+        except Exception as e:
+            log.error(f"Error getting details for {detail_url}: {e}")
+            sql_pool.update_one_or_dict(
+                table="scp_lot_record",
+                data={"state": 2},
+                condition={"id": sql_id},
+            )

scp_spider/scp_history.py

@@ -0,0 +1,102 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/07/15
+"""
+SCP Auctions (catalogs.scpauctions.com) 全量历史爬虫（一次性脚本）
+逻辑（两阶段）：
+  阶段一 列表：GET /auctions/past 翻页解析「全部」历史拍卖会 → 逐场入库 scp_auction，
+              再逐场翻页抓 lot 列表入库 scp_lot（state 默认 0）
+  阶段二 详情：扫库 scp_lot 中 state != 1 的记录 → 逐条进详情页抓
+              「标题 + 成交价 + 属性(Collection/Sport/Type/Athlete/Team) + 多图」写回
+适用场景：初始化数据库时跑一次。后续每周增量由 scp_spider.py 负责。
+"""
+import sys
+import random
+
+from curl_cffi import requests
+from loguru import logger
+from mysql_pool import MySQLConnectionPool
+
+from scp_core import (
+    client_identifier_list,
+    crawl_one_auction,
+    get_auction_list,
+    save_auction,
+    update_details_for_pending,
+)
+
+logger.remove()
+logger.add("./logs/his_{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")
+
+# —— 调试开关 ——
+# 只抓前 N 个拍卖会，None 表示全抓（生产）；测试时设小一点，比如 1
+DEBUG_AUCTION_LIMIT = None
+
+
+def run_history(log, sql_pool):
+    """全量抓取所有历史拍卖会及其 lot 列表（阶段一）。
+
+    Args:
+        log: logger 对象。
+        sql_pool: MySQL 连接池；传 None 时不入库，仅翻页抓取并 print 样本。
+
+    Returns:
+        None: 结果直接入库，无返回值。
+    """
+    impersonate = random.choice(client_identifier_list)
+    with requests.Session() as session:
+        try:
+            auctions = get_auction_list(log, session, impersonate, only_first_page=False)
+        except Exception as e:
+            log.error(f"获取拍卖会列表失败: {e}")
+            return
+
+        if DEBUG_AUCTION_LIMIT is not None:
+            auctions = auctions[:DEBUG_AUCTION_LIMIT]
+            log.warning(f"[DEBUG] 调试模式，仅抓前 {len(auctions)} 场拍卖会")
+
+        for idx, auc in enumerate(auctions, 1):
+            log.info(f"========== [{idx}/{len(auctions)}] 拍卖会 {auc['event_id']} ({auc['auction_name']}) ==========")
+            try:
+                save_auction(log, sql_pool, auc)  # 先入库拍卖会本身
+                lots = crawl_one_auction(log, sql_pool, session, impersonate, auc)
+                if sql_pool is None:
+                    for row in lots[:2]:
+                        print(row)
+            except Exception as e:
+                log.error(f"拍卖会 {auc['event_id']} 抓取异常: {e}")
+                continue
+
+
+def scp_history_main(log):
+    """全量历史抓取入口。
+
+    Args:
+        log: logger 对象。
+
+    Returns:
+        None: 无返回值。
+    """
+    log.info(f"开始运行 {sys._getframe().f_code.co_name} 全量爬虫任务 ...")
+
+    sql_pool = MySQLConnectionPool(log=log)
+    if not sql_pool:
+        log.error("MySQL数据库连接失败")
+        raise Exception("MySQL数据库连接失败")
+
+    try:
+        # 阶段一：抓拍卖会 + lot 列表入库
+        run_history(log, sql_pool)
+        # 阶段二：扫库 state != 1 的 lot 补抓详情
+        update_details_for_pending(log, sql_pool)
+    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__":
+    scp_history_main(log=logger)

scp_spider/scp_spider.py

@@ -0,0 +1,168 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/07/15
+"""
+SCP Auctions (catalogs.scpauctions.com) 每周增量爬虫（周调度）
+逻辑（两阶段）：
+  1. GET /auctions/past 只解析首页（最新的拍卖会排在首页顶部）
+  2. 查库 select distinct event_id from scp_auction，得到已爬过的场次
+  3. 差集 = 新增拍卖会
+  4. 没有新增 → 本轮无数据可抓，仍补跑一次阶段二兜底后结束
+  5. 阶段一 列表：对每个新增拍卖会入库 scp_auction，再翻页抓 lot 列表入库 scp_lot（state 默认 0）
+  6. 阶段二 详情：扫库 scp_lot 中 state != 1 的记录 → 逐条进详情页抓字段写回
+
+说明：SCP 的拍卖会一旦结束即定型，新场次会出现在 /auctions/past 首页顶部；
+      因此增量只需盯首页，用 event_id 差集识别新增场次即可（老场次早已在库）。
+"""
+import sys
+import time
+import random
+
+import schedule
+from curl_cffi import requests
+from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
+from mysql_pool import MySQLConnectionPool
+
+from scp_core import (
+    client_identifier_list,
+    crawl_one_auction,
+    get_auction_list,
+    save_auction,
+    update_details_for_pending,
+    after_log,
+)
+
+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")
+
+
+def get_existing_event_ids(log, sql_pool):
+    """查库返回已爬过的 event_id 集合。
+
+    Args:
+        log: logger 对象。
+        sql_pool: MySQL 连接池；为 None 时返回空集合（视作库内无任何场次）。
+
+    Returns:
+        set[str]: 已存在的 event_id 字符串集合。
+    """
+    if sql_pool is None:
+        log.warning("sql_pool 为 None，视为库内无任何场次（将全量重抓首页）")
+        return set()
+
+    rows = sql_pool.select_all("select distinct event_id from scp_auction_record")
+    keys = {str(r[0]) for r in rows} if rows else set()
+    log.info(f"库中已存在 {len(keys)} 个 event_id")
+    return keys
+
+
+def diff_new_auctions(log, recent_auctions, existing_keys):
+    """从首页拍卖会中筛出库里没有的新增场次。
+
+    Args:
+        log: logger 对象。
+        recent_auctions (list[dict]): get_auction_list(only_first_page=True) 的返回。
+        existing_keys (set[str]): 已存在的 event_id 集合。
+
+    Returns:
+        list[dict]: 待抓取的新增拍卖会列表。
+    """
+    new_list = [a for a in recent_auctions if a["event_id"] not in existing_keys]
+    log.info(f"新增待抓取拍卖会数: {len(new_list)} -> {[a['event_id'] for a in new_list]}")
+    return new_list
+
+
+def run_incremental(log, sql_pool):
+    """增量抓取主流程（阶段一）。
+
+    Args:
+        log: logger 对象。
+        sql_pool: MySQL 连接池；为 None 时不入库，仅在内存收集并 print 样本。
+
+    Returns:
+        None: 无返回值。
+    """
+    impersonate = random.choice(client_identifier_list)
+    with requests.Session() as session:
+        try:
+            recent = get_auction_list(log, session, impersonate, only_first_page=True)
+        except Exception as e:
+            log.error(f"获取首页拍卖会列表失败: {e}")
+            return
+
+        existing_keys = get_existing_event_ids(log, sql_pool)
+        new_auctions = diff_new_auctions(log, recent, existing_keys)
+
+        if not new_auctions:
+            log.info("本轮无新增拍卖会，跳过 list 抓取")
+            return
+
+        for idx, auc in enumerate(new_auctions, 1):
+            log.info(f"========== [{idx}/{len(new_auctions)}] 新增拍卖会 {auc['event_id']} ({auc['auction_name']}) ==========")
+            try:
+                save_auction(log, sql_pool, auc)
+                crawl_one_auction(log, sql_pool, session, impersonate, auc)
+            except Exception as e:
+                log.error(f"拍卖会 {auc['event_id']} 抓取异常: {e}")
+                continue
+
+
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
+def scp_main(log):
+    """周调度主函数：增量 list + 补详情。
+
+    Args:
+        log: logger 对象。
+
+    Returns:
+        None: 无返回值。
+    """
+    log.info(f"开始运行 {sys._getframe().f_code.co_name} 增量爬虫任务 ...")
+
+    sql_pool = MySQLConnectionPool(log=log)
+    if not sql_pool:
+        log.error("MySQL数据库连接失败")
+        raise Exception("MySQL数据库连接失败")
+
+    try:
+        # 阶段一：抓新增拍卖会的 lot 列表入库
+        try:
+            run_incremental(log, sql_pool)
+        except Exception as e:
+            log.error(f"增量抓取失败: {e}")
+
+        # 阶段二：扫库 state != 1 的 lot 补抓详情（含上一轮失败/中断的兜底）
+        try:
+            update_details_for_pending(log, sql_pool)
+        except Exception as e:
+            log.error(f"详情补抓失败: {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} 运行结束，等待下一轮采集 ...")
+
+
+def schedule_task():
+    """启动调度器：先即时跑一次，之后每周一 05:00 跑一次增量。
+
+    Returns:
+        None: 常驻循环，无返回值。
+    """
+    scp_main(log=logger)
+
+    schedule.every().monday.at("05:00").do(scp_main, log=logger)
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+if __name__ == "__main__":
+    # 测试时直接跑一次
+    # scp_main(log=logger)
+    # 上生产再切回 schedule
+    schedule_task()