update yueka spider 12.24.1

yueka_spider/YamlLoader.py

@@ -0,0 +1,78 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.10.8
+# 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 readYaml(path: str = 'application.yml', profile: str = None) -> YamlConfig:
+    if os.path.exists(path):
+        with open(path) as fd:
+            conf = yaml.load(fd, Loader=yaml.FullLoader)
+    
+    if profile is not None:
+        result = path.split('.')
+        profiledYaml = f'{result[0]}-{profile}.{result[1]}'
+        if os.path.exists(profiledYaml):
+            with open(profiledYaml) as fd:
+                conf.update(yaml.load(fd, Loader=yaml.FullLoader))
+    
+    return YamlConfig(conf)
+
+# res = readYaml()
+# mysqlConf = res.get('mysql')
+# print(mysqlConf)
+
+# print(res.getValueAsString("host"))
+# mysqlYaml = mysqlConf.getValueAsString("host")
+# print(mysqlYaml)
+# host = mysqlYaml.get("host").split(':')[-1][:-1]
+# port = mysqlYaml.get("port").split(':')[-1][:-1]
+# username = mysqlYaml.get("username").split(':')[-1][:-1]
+# password = mysqlYaml.get("password").split(':')[-1][:-1]
+# mysql_db = mysqlYaml.get("db").split(':')[-1][:-1]
+# print(host,port,username,password)

yueka_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}

yueka_spider/mysql_pool.py

@@ -0,0 +1,574 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.10.8
+# 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=4, maxcached=5, maxconnections=10, 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=0  # 每次连接使用时自动检查有效性（0=不检查,1=执行query前检查,2=每次执行前检查）
+        )
+
+    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:
+                conn.rollback()
+            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}, Rows: {len(args_list)}")
+            self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data insert_all 入库中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+        except Exception as e:
+            conn.rollback()
+            self.log.error(f"Batch insertion failed after 5 attempts. Trying single inserts. Error: {e}")
+            # 如果批量插入失败，则逐条插入
+            rowcount = 0
+            for args in args_list:
+                self.insert_one(query, args)
+                rowcount += 1
+            self.log.debug(f"Batch insertion failed. Inserted {rowcount} rows individually.")
+        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))
+
+            # query = f"INSERT INTO {self._safe_identifier(table)} ({keys}) VALUES ({values})"
+            # 构建 INSERT IGNORE 语句
+            ignore_clause = "IGNORE" if ignore else ""
+            # insert_sql = f"INSERT {ignore_clause} INTO {table} ({columns}) VALUES ({placeholders})"
+            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")
+
+        # 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
+
+        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):
+                self.log.warning(f"插入失败：重复条目，已跳过。错误详情: {e}")
+                # print("插入失败：重复条目", e)
+                return -1  # 返回 -1 表示重复条目被跳过
+            else:
+                self.log.exception(f"数据库完整性错误: {e}")
+                # print("插入失败：完整性错误", e)
+                raise
+        except Exception as e:
+            # self.log.error(f"未知错误: {str(e)}", exc_info=True)
+            self.log.exception(f"未知错误: {e}")  # 记录完整异常信息
+            # print("插入失败：未知错误", 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 ""
+            # insert_sql = f"INSERT {ignore_clause} INTO {table} ({columns}) VALUES ({placeholders})"
+            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.Error as e:
+                if "Duplicate entry" in str(e):
+                    # self.log.warning(f"检测到重复条目，开始逐条插入。错误详情: {e}")
+                    raise  e
+                    # rowcount = 0
+                    # for args in batch:
+                    #     try:
+                    #         self.insert_one_or_dict(table=table, data=dict(zip(data_list[0].keys(), args)),
+                    #                                 commit=commit)
+                    #         rowcount += 1
+                    #     except pymysql.err.IntegrityError as e2:
+                    #         if "Duplicate entry" in str(e2):
+                    #             self.log.warning(f"跳过重复条目: {args}")
+                    #         else:
+                    #             self.log.error(f"插入失败: {e2}, 参数: {args}")
+                    # total += rowcount
+                else:
+                    self.log.exception(f"数据库错误: {e}")
+                    if commit:
+                        conn.rollback()
+                    raise e
+                # 重新抛出异常，供外部捕获
+                # 降级为单条插入
+                # for args in batch:
+                #     try:
+                #         self.insert_one_or_dict(table=None, query=query, args=args, commit=commit)
+                #         total += 1
+                #     except Exception as e2:
+                #         self.log.error(f"Single insert failed: {e2}")
+                        # continue
+        self.log.info(f"sql insert_many, Table: {table}, 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):
+        """
+        批量插入（支持字典列表或原始SQL）
+        :param table: 表名（字典插入时必需）
+        :param 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")
+            keys = ', '.join([self._safe_identifier(k) for k in data_list[0].keys()])
+            values = ', '.join(['%s'] * len(data_list[0]))
+            query = f"INSERT 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:
+                        # 添加调试日志：输出 SQL 和参数示例
+                        # self.log.debug(f"Batch insert SQL: {query}")
+                        # self.log.debug(f"Sample args: {batch[0] if batch else 'None'}")
+                        cursor.executemany(query, batch)
+                        if commit:
+                            conn.commit()
+                        total += cursor.rowcount
+                        # self.log.debug(f"Batch insert succeeded. Rows: {cursor.rowcount}")
+            except Exception as e:  # 明确捕获数据库异常
+                self.log.exception(f"Batch insert failed: {e}")  # 使用 exception 记录堆栈
+                self.log.error(f"Failed SQL: {query}, Args count: {len(batch)}")
+                if commit:
+                    conn.rollback()
+                # 降级为单条插入，并记录每个错误
+                rowcount = 0
+                for args in batch:
+                    try:
+                        self.insert_one(query, args)
+                        rowcount += 1
+                    except Exception as e2:
+                        self.log.error(f"Single insert failed: {e2}, Args: {args}")
+                total += rowcount
+                self.log.debug(f"Inserted {rowcount}/{len(batch)} rows individually.")
+        self.log.info(f"sql insert_many, 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: 每次插入的条数
+        """
+        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()
+            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
+
+    @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)

yueka_spider/settings.py

@@ -0,0 +1,564 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.10.8
+# Date   : 2025/3/24 15:05
+import inspect
+import requests
+from loguru import logger
+from bs4 import BeautifulSoup
+from tenacity import retry, stop_after_attempt, wait_fixed
+
+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")
+
+HEADERS = {
+    "User-Agent": "Dart/3.5 (dart:io)",
+    "Accept-Encoding": "gzip",
+    "Content-Type": "application/json",
+    "deviceid": "110bec91-951f-4346-8195-95700fdb8c1d",
+    "brand": "google",
+    "os": "android",
+    "content-type": "application/json; charset=utf-8",
+    # "authori-zation": "a-209a2707889f473688a742c4bac06eed",
+    "systemversion": "30",
+    "theme": "light",
+    "lang": "zh",
+    "verse-ua": "5282360bfafdd013cec4258b31c53f13",
+    "version": "1.0.7",
+    "isphysicaldevice": "true",
+    "sktime": "1766401868617",
+    "cid": "42661545",
+    "sk": "f0635a5c5fee2a9d137358264721e7ff" # 每次版本变化 也需要修改
+}
+
+
+
+def after_log(retry_state):
+    """
+    retry 回调
+    :param retry_state: RetryCallState 对象
+    """
+    # 检查 args 是否存在且不为空
+    if retry_state.args and len(retry_state.args) > 0:
+        log = retry_state.args[0]  # 获取传入的 logger
+    else:
+        log = logger  # 使用全局 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):
+    """
+    获取代理
+    :return: 代理
+    """
+    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
+
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
+def make_request(log, method, url, params=None, data=None, headers=None, proxies=None, timeout=22, token=None):
+    """
+    通用请求函数
+    :param log: logger对象
+    :param method: 请求方法 ('GET' 或 'POST')
+    :param url: 请求的URL
+    :param params: GET请求的查询参数
+    :param data: POST请求的数据
+    :param headers: 请求头
+    :param proxies: 代理
+    :param timeout: 请求超时时间
+    :param token: token
+    :return: 响应的JSON数据
+    """
+    if headers is None:
+        headers = HEADERS
+        if 'getHitCardReport' or 'getCardPublicly' or 'productDetailDynamics'  or 'productShowList' in url:
+            if not token:
+                token = "a-87768ad775bf4513bec14f2c4fd2cd0c"
+            headers["authori-zation"] = token
+
+    if proxies is None:
+        proxies = get_proxys(log)
+
+    try:
+        with requests.Session() as session:
+            if method.upper() == 'GET':
+                if proxies is None:
+                    response = session.get(url, headers=headers, params=params, timeout=timeout)
+                else:
+                    response = session.get(url, headers=headers, params=params, proxies=proxies, timeout=timeout)
+            elif method.upper() == 'POST':
+                if proxies is None:
+                    response = session.post(url, headers=headers, json=data, timeout=timeout)
+                    # print(response.text)
+                else:
+                    response = session.post(url, headers=headers, json=data, proxies=proxies, timeout=timeout)
+            else:
+                log.error(f"Unsupported request method: {method}")
+                return None
+
+            response.raise_for_status()
+            data = response.json()
+            # print(response.text)
+            if data["code"] == 200:
+                log.info(f"Successfully fetched {method} request to {url}")
+                return data
+            else:
+                log.warning(f"Warning {inspect.currentframe().f_code.co_name}: {data['message']}")
+                return {}
+    except requests.exceptions.RequestException as e:
+        log.error(f"Error making {method} request to {url}: {e}")
+        raise e
+    except ValueError as e:
+        log.error(f"Error parsing JSON for {method} request to {url}: {e}")
+        raise e
+    except Exception as e:
+        log.error(f"Error making {method} request to {url}: {e}")
+        raise e
+
+
+def get_play_back(log, product_id, token):
+    """
+    获取 视频回放链接
+    :param log: logger对象
+    :param product_id: product_id
+    :param token: token
+    """
+    log.info(f"Starting to fetch playback for product_id {product_id}")
+    url = "https://api.joycard.xyz/api/front/c/product/productDetailDynamics"
+    params = {
+        # "code": "LCS1254174"
+        "code": product_id
+    }
+    try:
+        response = make_request(log, 'GET', url, params=params, token=token)
+        if response:
+            items = response.get("data", {})
+            normalLiving = items.get("normalLiving", {})
+            playback = normalLiving.get("playback")
+            return playback
+        else:
+            return None
+    except Exception as e:
+        log.error(f"Error fetching playback {product_id}: {e}")
+        return None
+
+
+def clean_texts(html_text):
+    """
+    使用 BeautifulSoup 解析并获取纯文本
+    :param html_text: 待解析的HTML格式的数据
+    :return: clean_text -> 解析后的数据
+    """
+    if not html_text:
+        return ""
+    soup = BeautifulSoup(html_text, 'html.parser')
+    # clean_text = soup.get_text(separator=' ', strip=True)
+    clean_text = soup.get_text(strip=True)
+    # 替换   为普通空格
+    clean_text = clean_text.replace(' ', ' ')
+    return clean_text
+
+
+def parse_product_items(log, items, sql_pool, product_id, token):
+    """
+    解析 产品信息
+    :param log: logger对象
+    :param items: 请求response
+    :param sql_pool: MySQL连接池对象
+    :param product_id: product_id
+    :param token: token
+    """
+    if not items:
+        log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No items found")
+        return
+    no = items.get("id")
+    create_time = items.get("publishTime")
+    title = items.get("productName")
+    img = items.get("productImageIndex")
+    price_sale = items.get("unitPriceStr")
+    total_price = items.get("totalSalePrice")
+    sale_num = items.get("saleCount")  # 售出数量
+    spec_config = items.get("hitCardStandard")  # 规格
+    sort = items.get("series")  # 分类  0:全部  1:原盒  2:幸运盒  3:福盒?
+    state = items.get("status")
+    shop_id = items.get("merchantCode")
+    shop_name = items.get("merchantName")
+    category = items.get("brandId")
+    on_sale_time = items.get("onlineTime")
+    end_time = items.get("endTime")
+    finish_time = items.get("finishTime")
+    # content = items.get("purchaseNotes")
+    # if content:
+    #     content = content.replace("<p>", "").replace("</p>", "")
+    # brief = items.get("brief")
+    product_detail = items.get("productDetail")
+    if product_detail:
+        product_detail = clean_texts(product_detail)
+        # print('product_detail:',product_detail)
+
+    video_url = get_play_back(log, product_id, token)
+
+    hit_card_desc = items.get("hitCardDesc")  # 赠品介绍
+    open_mode = items.get("openMode")  # 随机球队
+    open_mode_comment = items.get("openModeComment")  # 随机球队 说明
+
+    random_mode = items.get("randomMode")  # 即买即随
+    random_mode_comment = items.get("randomModeComment")  # 即买即随 说明
+
+    info_dict = {
+        "no": no,
+        "create_time": create_time,
+        "title": title,
+        "img": img,
+        "price_sale": price_sale,
+        "total_price": total_price,
+        "sale_num": sale_num,
+        "spec_config": spec_config,
+        "sort": sort,
+        "state": state,
+        "shop_id": shop_id,
+        "shop_name": shop_name,
+        "category": category,
+        "on_sale_time": on_sale_time,
+        "end_time": end_time,
+        "finish_time": finish_time,
+        "product_detail": product_detail,
+        "video_url": video_url,
+        "hit_card_desc": hit_card_desc,
+        "open_mode": open_mode,
+        "open_mode_comment": open_mode_comment,
+        "random_mode": random_mode,
+        "random_mode_comment": random_mode_comment,
+    }
+    # print(info_dict)
+    # sql_pool.insert_one_or_dict(table="yueka_product_record", data=info_dict)
+    sql_pool.update_one_or_dict(table="yueka_product_record", data=info_dict, condition={"product_id": product_id})
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
+def get_product_details(log, product_id, sql_pool, token):
+    """
+    获取 商品详情 单条 信息
+    :param log: logger对象
+    :param product_id: product_id
+    :param sql_pool: MySQL连接池对象
+    :param token: token
+    """
+    log.debug(f"Getting product details for {product_id}")
+    url = "https://api.joycard.xyz/api/front/c/product/productDetail"
+    params = {
+        # "code": "LCS1254079"
+        "code": product_id
+    }
+    try:
+        response = make_request(log, 'GET', url, params=params, token=token)
+        if response:
+            parse_product_items(log, response.get("data"), sql_pool, product_id, token)
+        else:
+            log.error(f"Error getting product details for {product_id}: {response.get('msg')}")
+    except Exception as e:
+        log.error(f"Error getting product details for {product_id}: {e}")
+
+
+def get_product_detail_list(log, sql_pool, token):
+    """
+    获取 商品详情 列表 信息
+    :param log: logger对象
+    :param sql_pool: MySQL连接池对象
+    :param token: token
+    """
+    sql_product_id_list = sql_pool.select_all("SELECT product_id FROM yueka_product_record WHERE no IS NULL")
+    sql_product_id_list = [item[0] for item in sql_product_id_list]
+    for product_id in sql_product_id_list:
+        try:
+            get_product_details(log, product_id, sql_pool, token)
+        except Exception as e:
+            log.error(f"Error get_product_detail_list fetching product {product_id}: {e}")
+            continue
+
+
+def parse_player_items(log, items, sql_pool, product_id):
+    """
+    解析 卡密公示 信息
+    :param log: logger对象
+    :param items: 请求response
+    :param product_id: product_id
+    :param sql_pool: MySQL连接池对象
+    """
+    if not items:
+        log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No items found")
+        return
+    player_list = []
+    for item in items:
+        # print(item)
+        user_code = item.get("userCode")
+        user_id = item.get("userId")
+        user_name = item.get("nickName")
+        num = item.get("cardCount")
+        # info = (product_id, user_code, num, user_id, user_name)
+        info_dict = {
+            "product_id": product_id,
+            "user_code": user_code,
+            "num": num,
+            "user_id": user_id,
+            "user_name": user_name
+        }
+        # print(info_dict)
+        player_list.append(info_dict)
+    sql_pool.insert_many(table='yueka_player_record', data_list=player_list)
+    sql_pool.update_one("update yueka_product_record set km_state = 1 where product_id = %s", (product_id,))
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
+def get_player_list(log, product_id, sql_pool, token):
+    """
+    抓取 kami公示 信息
+    :param log: logger对象
+    :param product_id: product_id
+    :param sql_pool: MySQL连接池对象
+    :param token: token
+    """
+    log.debug(f"Getting player list for {product_id}")
+    url = "https://api.joycard.xyz/api/front/c/card/getCardPublicly"
+    last_id = 0  # 初始lastId为0
+    total_players = 0
+
+    while True:
+        data = {
+            "keyword": "",
+            "lastUserId": last_id,
+            "productCode": product_id,
+            "publiclyType": 2,  # 1:赠品维度 2:玩家维度
+        }
+        # print(data)
+
+        try:
+            response = make_request(log, 'POST', url, data=data, token=token)
+            if not response:
+                log.error(f"Error getting player list for {product_id}: Empty response")
+                break
+
+            items = response.get("data", [])
+            if not items:
+                log.info(f"No more players found for product {product_id}")
+                sql_pool.update_one("update yueka_product_record set km_state = 3 where product_id = %s", (product_id,))
+                break
+
+            # 处理当前页数据
+            parse_player_items(log, items, sql_pool, product_id)
+            total_players += len(items)
+
+            # 如果获取数量超过50条，说明已经获取到所有数据，结束循环
+            if total_players > 50:
+                log.debug(f"Total players found for product {product_id}: {total_players}")
+                break
+
+            # 如果获取数量不足20条，说明是最后一页
+            if len(items) < 20:
+                log.info(f"Last page detected for product {product_id} (got {len(items)} items)")
+                break
+
+            # 更新lastId为最后一条的userId
+            last_id = items[-1].get("userId")
+            # print(last_id)
+            if not last_id:
+                log.error("API response missing userId in last item, cannot paginate")
+                break
+
+            # 避免频繁请求
+            # time.sleep(0.5)
+
+        except Exception as e:
+            log.error(f"Error getting player list for {product_id} at lastId {last_id}: {e}")
+            break
+
+    log.info(f"Finished fetching players for product {product_id}, total: {total_players}")
+
+
+def get_players(log, sql_pool, token):
+    """
+    抓取 kami公示 信息
+    :param log: logger对象
+    :param sql_pool: MySQL连接池对象
+    :param token: token
+    """
+    product_list = sql_pool.select_all("SELECT product_id FROM yueka_product_record WHERE km_state IN (0, 3)")
+    product_list = [product_id[0] for product_id in product_list]
+
+    # token = sql_pool.select_one("SELECT token FROM yueka_token")
+    # token = token[0]
+    if not product_list:
+        log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No product_id found")
+        return
+    else:
+        log.info(f"Start fetching players data. Total products: {len(product_list)}")
+        for product_id in product_list:
+            try:
+                get_player_list(log, product_id, sql_pool, token)
+            except Exception as e:
+                log.error(f"Error fetching product {product_id}: {e}")
+                continue
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
+def get_report_one_page(log, sql_pool, productCode, page, last_id, token):
+    """
+    获取 拆卡报告 单页的信息
+    :param log: logger对象
+    :param sql_pool: MySQL连接池对象
+    :param productCode: product_id
+    :param page: 页码
+    :param last_id: last_id
+    :param token: token
+    """
+    url = "https://api.joycard.xyz/api/front/c/card/getHitCardReport"
+    data = {
+        "keyword": "",
+        "page": page,
+        "lastId": last_id,
+        # "productCode": "LCS1254213"
+        "productCode": productCode
+    }
+    log.info(f"Getting report data for: {productCode}, Page: {page}")
+
+    try:
+        response = make_request(log, 'POST', url, data=data, token=token)
+        # print(response)
+        if response:
+            items = response.get("data", [])
+            if items:
+                info_list = []
+                for item in items:
+                    card_id = item.get("orderNo")
+                    card_name = item.get("cardSecret")
+                    create_time = item.get("drawTime")
+                    imgs = item.get("hitPic")
+                    user_id = item.get("userCode")
+                    user_name = item.get("nickName")
+                    shop_id = item.get("merchantCode")
+                    shop_name = item.get("merchantName")
+                    card_desc = item.get("hitCardDesc")
+                    # info = (card_id, card_name, create_time, imgs, user_id, user_name, shop_id, shop_name, card_desc)
+                    info_dict = {
+                        "product_id": productCode,
+                        "card_id": card_id,
+                        "card_name": card_name,
+                        "create_time": create_time,
+                        "imgs": imgs,
+                        "user_id": user_id,
+                        "user_name": user_name,
+                        "shop_id": shop_id,
+                        "shop_name": shop_name,
+                        "card_desc": card_desc
+                    }
+                    # print(info_dict)
+                    info_list.append(info_dict)
+                sql_pool.insert_many(table='yueka_report_record', data_list=info_list)
+                log.info(f"Successfully saved {len(items)} report items")
+                return items[-1].get("userCode"), len(items)
+            else:
+                log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No items found")
+                sql_pool.update_one("update yueka_product_record set report_state = 3 where product_id = %s",
+                                    (productCode,))
+                return 0, 0
+        else:
+            log.error(f"Error getting report data: {response.get('msg')}")
+            return 0
+    except Exception as e:
+        log.error(f"Error getting report data: {e}")
+        raise e
+
+
+def get_report_list(log, sql_pool, product_id, token):
+    """
+    抓取 拆卡报告 单个product_id 所有页码的 信息
+    :param log: logger对象
+    :param sql_pool: MySQL连接池对象
+    :param product_id: product_id
+    :param token: token
+    """
+    # log.info(f"Start fetching report data. Product id: {product_id}")
+    page = 1
+    last_id = 0
+    # while True:
+    try:
+        last_d, len_item = get_report_one_page(log, sql_pool, product_id, page, last_id, token)
+
+        # if len_item != 0 and len_item < 20:
+        log.info(f"Finished fetching report data for product {product_id}, total: {len_item}")
+        sql_pool.update_one("update yueka_product_record set report_state = 1 where product_id = %s", (product_id,))
+
+        # # 如果获取数量不足20条，说明是最后一页  ***暂时没找到第二页的***
+        # if len_item < 20:
+        #     log.info(f"Last page detected for product {product_id} (got {len_item} items)")
+        #     break
+        #
+        # # 更新lastId为最后一条的userId
+        # last_id = last_d
+        # if not last_id:
+        #     log.error("API response missing userId in last item, cannot paginate")
+        #     break
+        #
+        # page += 1
+    except Exception as e:
+        log.error(f"Error getting report data: {e}")
+        # break
+
+
+def get_reports(log, sql_pool, token):
+    """
+    抓取 拆卡报告 信息
+    :param log: logger对象
+    :param sql_pool: MySQL连接池对象
+    :param token: token
+    """
+    product_list = sql_pool.select_all("SELECT product_id FROM yueka_product_record WHERE report_state IN (0, 3)")
+    product_list = [product_id[0] for product_id in product_list]
+
+    # token = sql_pool.select_one("SELECT token FROM yueka_token")
+    # token = token[0]
+    if not product_list:
+        log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No product_id found")
+        return
+    else:
+        log.info(f"Start fetching report data. Total products: {len(product_list)}")
+        for product_id in product_list:
+            try:
+                get_report_list(log, sql_pool, product_id, token)
+            except Exception as e:
+                log.error(f"Error fetching product {product_id}: {e}")
+                continue
+
+
+if __name__ == '__main__':
+    pass
+
+    # pid = 'LCS1254213'
+    # pid = 'LCS1253418'
+    # pid = 'LCS1256332'
+    # from mysql_pool import MySQLConnectionPool
+    # sql_pool_ = MySQLConnectionPool(log=logger)
+
+    # get_reports(logger, None)
+    # get_player_list(logger, pid, None)
+    # get_product_details(logger, 'LCS1255968', sql_pool_)

yueka_spider/start_yueka_spider.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.10.8
+# Date   : 2025/3/25 18:44
+import time
+import schedule
+import threading
+from yk_login import login_main
+from settings import *
+from yk_new_daily_spider import yueka_main
+
+
+def run_threaded(job_func, *args, **kwargs):
+    """
+    在新线程中运行给定的函数，并传递参数。
+
+    :param job_func: 要运行的目标函数
+    :param args: 位置参数
+    :param kwargs: 关键字参数
+    """
+    job_thread = threading.Thread(target=job_func, args=args, kwargs=kwargs)
+    job_thread.start()
+
+
+def schedule_task():
+    """
+    爬虫模块的启动文件
+    """
+    # 立即运行一次任务
+    # login_main()
+    # time.sleep(5)
+    
+    run_threaded(yueka_main, log=logger)
+    
+    # 每天  定时更新token
+    schedule.every().day.at("00:01").do(run_threaded, login_main)
+    
+    # 设置定时任务
+    schedule.every().day.at("00:06").do(run_threaded, yueka_main, log=logger)
+    
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+if __name__ == '__main__':
+    schedule_task()

yueka_spider/yk_login.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.10.8
+# Date   : 2025/3/24 17:53
+import requests
+# from loguru import logger
+from mysql_pool import MySQLConnectionPool
+from settings import *
+
+# logger.remove()
+# logger.add("./logs/login_{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_token(log):
+    log.debug("Start fetching token...........")
+    url = "https://api.joycard.xyz/api/front/login/pwd"
+    data = {
+        "phone": "19521500850",
+        "pwd": "Pass2025",
+        "spreaderInviteCode": ""
+    }
+    response = requests.post(url, headers=HEADERS, json=data)
+    # print(response.text)
+    resp_json = response.json()
+    if resp_json["code"] == 200:
+        token = resp_json["data"]["token"]
+        log.debug("Fetch token success!")
+        return token
+    else:
+        log.error(f"Fetch token failed! Msg:{resp_json['message']}")
+        return None
+
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
+def login_main():
+    # 配置 MySQL 连接池
+    sql_pool = MySQLConnectionPool(log=logger)
+    if not sql_pool.check_pool_health():
+        logger.error("数据库连接池异常")
+        raise RuntimeError("数据库连接池异常")
+
+    token = get_token(logger)
+    if token is None:
+        logger.error("获取token失败")
+        raise RuntimeError("获取token失败")
+    else:
+        sql_pool.update_one_or_dict(table="yueka_token", data={"token": token}, condition={"id": 1})
+        logger.debug("------------------ 更新token成功 ------------------")
+
+
+if __name__ == '__main__':
+    login_main()
+    # get_token(logger)

yueka_spider/yk_new_daily_spider.py

@@ -0,0 +1,248 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.10.8
+# Date   : 2025/12/22 10:44
+import time
+from mysql_pool import MySQLConnectionPool
+from settings import *
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
+def get_all_sold_one_page(log, page: int, last_id, lastSalePrice):
+    """
+    获取指定页面的已售数据
+    :param log: logger对象
+    :param page: 页码
+    :param last_id: last_id
+    :param lastSalePrice: lastSalePrice
+    :return: 该页的数据, totalPage, total
+    """
+    log.info(f"Starting < get_all_sold_one_page > to fetch page {page}")
+    url = "https://api.joycard.xyz/api/front/c/product/productShowList"
+    #      https://api.joycard.xyz/api/front/c/product/productShowList
+    data = {
+        "lastId": last_id,
+        "lastSalePrice": lastSalePrice,
+        "limit": 20,
+        "openMode": "",
+        "page": page,
+        "saleStatus": "2",
+        "sort": "0"
+    }
+    # {
+    #   "lastId": 0,
+    #   "lastSalePrice": "",
+    #   "limit": 20,
+    #   "openMode": "",
+    #   "page": 1,
+    #   "saleStatus": "1",
+    #   "sort": "0"
+    # }
+    
+    try:
+        # data = json.dumps(data, separators=(',', ':'))
+        response = make_request(log, 'POST', url, data=data)
+        # print(response)
+        if response:
+            items = response["data"]["list"]
+            total_page = response["data"]["totalPage"]
+            total = response["data"]["total"]
+            log.info(f"Successfully fetched page {page}: {len(items)} items")
+            return items, total_page, total
+        else:
+            return [], 0, 0
+    except requests.exceptions.RequestException as e:
+        log.error(f"Error fetching page {page}: {e}")
+        raise e
+    except ValueError as e:
+        log.error(f"Error parsing JSON for page {page}: {e}")
+        raise e
+
+
+def get_shop_detail(log, shop_id):
+    """
+    获取店铺详情信息
+    :param log:
+    :param shop_id:
+    :return:
+    """
+    log.info(f"Start fetching shop {shop_id}")
+    url = f"https://api.joycard.xyz/api/front/c/merchant/{shop_id}"
+    try:
+        response = make_request(log, 'GET', url)
+        if response:
+            item = response["data"]
+            fans_num = item.get("fansNum")
+            group_num = item.get("salesQuantity")
+            create_time = item.get("createTime")
+            log.info(f"Successfully fetched shop {shop_id}")
+            return fans_num, group_num, create_time
+        else:
+            return None, None, None
+    except Exception as e:
+        log.error(f"Error fetching shop {shop_id}: {e}")
+        return None, None, None
+
+
+def parse_shop_items(log, items, sql_pool):
+    if not items:
+        log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No items found")
+        return
+    
+    info_list = []
+    for item in items:
+        shop_id = item.get("merchantCode")
+        
+        fans_num, group_num, create_time = get_shop_detail(log, shop_id)
+        
+        shop_name = item.get("merchantName")
+        shop_info_dict = {
+            "shop_id": shop_id,
+            "shop_name": shop_name,
+            "fans_num": fans_num,
+            "group_num": group_num,
+            "create_time": create_time
+        }
+        info_list.append(shop_info_dict)
+    
+    # 插入数据
+    if info_list:
+        sql_pool.insert_many(table="yueka_shop_record", data_list=info_list, ignore=True)
+
+
+def get_product(log, items, sql_pool, last_product_id):
+    if not items:
+        log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No items found")
+        return True
+    
+    should_stop = False
+    info_list = []
+    for item in items:
+        product_id = item.get("code")
+        if not product_id:
+            log.warning(f"Warning {inspect.currentframe().f_code.co_name}: No product_id found")
+            continue
+        info_list.append(
+            {
+                "product_id": product_id,
+            }
+        )
+        
+        # 判断是否是昨天的最后一条id, 如果是  则停止翻页 / 只有当 last_product_id 不为 None 时才判断是否停止翻页
+        if last_product_id and product_id == last_product_id:
+            log.info(
+                f"----------------- The product_id {product_id} is the last product_id:{last_product_id}, stop fetching -----------------")
+            should_stop = True
+            
+    # 插入数据
+    if info_list:
+        sql_pool.insert_many(table="yueka_product_record", data_list=info_list, ignore=True)
+    
+    # 如果 items 数量小于 20，说明已经到达最后一页
+    if len(items) < 20:
+        log.info(
+            f"----------------- {len(items)} items found, less than 20, stop fetching -----------------")
+        should_stop = True
+    
+    return should_stop
+
+
+def get_all_sold_data(log, sql_pool, last_product_id):
+    """
+    获取 全部类别的已售数据
+    :param sql_pool: MySQL连接池对象
+    :param log: logger对象
+    :param last_product_id: last_product_id，如果为 None 则表示从头开始采集
+    """
+    page = 1
+    max_page = 500
+    last_id = 0
+    lastSalePrice = ''
+    
+    while page <= max_page:
+        items, total_page, total = get_all_sold_one_page(log, page, last_id, lastSalePrice)
+        if not items:
+            break
+        
+        # 处理 items 数据
+        parse_shop_items(log, items, sql_pool)
+        
+        stop_page = get_product(log, items, sql_pool, last_product_id)
+        if stop_page:
+            log.info(
+                f"----------------- The product_id {last_product_id} is the last product_id, stop fetching -----------------")
+            break
+        
+        # 更新lastId为最后一条的userId
+        last_id = items[-1].get("id")
+        lastSalePrice = items[-1].get("unitPriceStr")
+        if not last_id:
+            log.error("API response missing userId in last item, cannot paginate")
+            break
+        if not lastSalePrice:
+            log.error("API response missing lastSalePrice in last item, cannot paginate")
+            break
+        
+        page += 1
+        
+        log.info(f"Finished fetching all data. Total pages: {total_page}, total items: {total}")
+
+
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
+def yueka_main(log):
+    """
+    主函数
+    :param log: logger对象
+    """
+    log.info(
+        f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................')
+    
+    # 配置 MySQL 连接池
+    sql_pool = MySQLConnectionPool(log=log)
+    if not sql_pool.check_pool_health():
+        log.error("数据库连接池异常")
+        raise RuntimeError("数据库连接池异常")
+    
+    try:
+        token = sql_pool.select_one("SELECT token FROM yueka_token")
+        token = token[0]
+        try:
+            # 获取最后一条pid的数据
+            last_product_id_result = sql_pool.select_one(
+                "SELECT product_id FROM yueka_product_record ORDER BY finish_time DESC LIMIT 1")
+            # 如果表中没有数据，last_product_id_result 为 None
+            last_product_id = last_product_id_result[0] if last_product_id_result else None
+            
+            get_all_sold_data(log, sql_pool, last_product_id)
+        except Exception as e:
+            log.error(f"Error fetching last_product_id: {e}")
+        time.sleep(5)
+        
+        # 获取商品详情
+        try:
+            get_product_detail_list(log, sql_pool, token)
+        except Exception as e:
+            log.error(f"Error fetching product_detail_list: {e}")
+        time.sleep(5)
+        
+        # 获取商品玩家
+        try:
+            get_players(log, sql_pool, token)
+        except Exception as e:
+            log.error(f"Error fetching players: {e}")
+        time.sleep(5)
+        
+        # 获取拆卡报告
+        try:
+            get_reports(log, sql_pool, token)
+        except Exception as e:
+            log.error(f"Error fetching reports: {e}")
+    except Exception as e:
+        log.error(f'{inspect.currentframe().f_code.co_name} error: {e}')
+    finally:
+        log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............')
+
+
+if __name__ == '__main__':
+    yueka_main(logger)
+    # get_all_sold_one_page(logger, 1, 0, '')