refactor(mysql_pool): 优化数据库连接池及批量插入逻辑

- 降低初始化连接池的最小和最大缓存连接数，减少资源占用
- 调整连接参数，添加连接和写操作超时设置，确保连接有效性
- 增强_execute方法，支持连接断开自动重试机制，提高稳定性
- 批量插入函数改进，遇到唯一键冲突时降级为逐条插入，避免失败，提升健壮性
- insert_one_or_dict方法优化，重复条目以警告日志跳过，避免日志堆栈污染
- 批量插入相关方法新增ignore参数，支持INSERT IGNORE语句，灵活控制冲突处理
- 新增close方法，支持安全关闭连接池，释放数据库资源
- 完善日志输出，增加操作细节和错误上下文，便于问题排查
- pcg_spider.py中修改SQL查询条件，避免选取state为0的任务，修正业务逻辑
- YamlLoader.py重构yaml路径解析，支持主脚本目录路径查找，提高配置加载灵活性
- 修复YamlLoader中正则表达式与getValueAsX方法的格式和逻辑，提升环境变量解析准确性
- 统一文件编码和异常处理，清理冗余代码及调试注释，提高代码规范性和可读性

pcg_spider/YamlLoader.py

@@ -1,74 +1,98 @@
 # -*- 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\.]+)\}$')
+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):
+    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'] != None:
+            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'] != None:
+            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, env: str = None):
+
+    def getValueAsBool(self, key: str):
         try:
             match = regex.match(self.config[key])
             group = match.groupdict()
-            if group['ENV'] != None:
+            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:
+
+
+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):
-        with open(path) as fd:
-            conf = yaml.load(fd, Loader=yaml.FullLoader)
+        return path
 
-    if profile != None:
-        result = path.split('.')
+    # 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) as fd:
+            with open(profiledYaml, encoding='utf-8') 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)

pcg_spider/mysql_pool.py

@@ -23,7 +23,7 @@ class MySQLConnectionPool:
     MySQL连接池
     """
 
-    def __init__(self, mincached=4, maxcached=5, maxconnections=10, log=None):
+    def __init__(self, mincached=1, maxcached=2, maxconnections=3, log=None):
         """
         初始化连接池
         :param mincached: 初始化时，链接池中至少创建的链接，0表示不创建
@@ -44,30 +44,75 @@ class MySQLConnectionPool:
             user=sql_user,
             password=sql_password,
             database=sql_db,
-            ping=0  # 每次连接使用时自动检查有效性（0=不检查,1=执行query前检查,2=每次执行前检查）
+            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
+        执行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.error(f"Error executing query: {e}, Query: {query}, Args: {args}")
-            raise e
+        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):
         """
@@ -101,7 +146,7 @@ class MySQLConnectionPool:
 
     def insert_all(self, query, args_list):
         """
-        执行批量插入语句，如果失败则逐条插入
+        执行批量插入语句,如果失败则逐条插入
         :param query: 插入语句
         :param args_list: 插入参数列表
         """
@@ -112,17 +157,33 @@ class MySQLConnectionPool:
             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.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.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.")
+            self.log.exception(f"批量插入失败: {e}")
+            raise e
         finally:
             if cursor:
                 cursor.close()
@@ -147,20 +208,13 @@ class MySQLConnectionPool:
             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}")
@@ -168,19 +222,18 @@ class MySQLConnectionPool:
             return cursor.lastrowid
         except pymysql.err.IntegrityError as e:
             if "Duplicate entry" in str(e):
-                self.log.warning(f"插入失败：重复条目，已跳过。错误详情: {e}")
-                # print("插入失败：重复条目", 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"数据库完整性错误: {e}")
-                # print("插入失败：完整性错误", e)
-                raise e
+                self.log.error(f"数据库完整性错误 Table: {table}, Error: {e}")
+                raise
         except Exception as e:
-            self.log.error(f"未知错误: {e}", exc_info=True)
-            # print("插入失败：未知错误", e)
-            raise 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):
+    def insert_many(self, table=None, data_list=None, query=None, args_list=None, batch_size=1000, commit=True,
+                    ignore=False):
         """
         批量插入（支持字典列表或原始SQL）
         :param table: 表名（字典插入时必需）
@@ -201,7 +254,6 @@ class MySQLConnectionPool:
 
             # 构建 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:
@@ -217,48 +269,71 @@ class MySQLConnectionPool:
                         if commit:
                             conn.commit()
                         total += cursor.rowcount
-            except pymysql.Error as e:
+            except pymysql.err.IntegrityError 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
+                    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.error(f"数据库错误: {e}")
+                    # 其他完整性错误
+                    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}")
+            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):
+    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: 表名（字典插入时必需）
+        批量插入(支持字典列表或原始SQL) - 备用方法
+        :param table: 表名(字典插入时必需)
         :param data_list: 字典列表 [{列名: 值}]
-        :param query: 直接SQL语句（与data_list二选一）
-        :param args_list: SQL参数列表（query使用时必需）
+        :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:
@@ -266,41 +341,51 @@ class MySQLConnectionPool:
                 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})"
+            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:
-                        # 添加调试日志：输出 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)}")
+            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()
-                # 降级为单条插入，并记录每个错误
-                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}")
+                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):
@@ -310,6 +395,7 @@ class MySQLConnectionPool:
         :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:
@@ -317,6 +403,7 @@ class MySQLConnectionPool:
                     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}")
                 # 当前批次降级为单条插入
@@ -553,6 +640,17 @@ class MySQLConnectionPool:
             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标识符安全校验"""
@@ -560,3 +658,14 @@ class MySQLConnectionPool:
             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)

pcg_spider/pcg_spider.py

@@ -164,7 +164,7 @@ def pcg_main(log):
 
     try:
         while True:
-            sql_cert_id_list = sql_pool.select_all("select cert_id from pcg_task where state = 0 limit 10000")
+            sql_cert_id_list = sql_pool.select_all("select cert_id from pcg_task where state != 0 limit 10000")
             sql_cert_id_list = [i[0] for i in sql_cert_id_list]
             log.info(f"开始处理 {len(sql_cert_id_list)} 条数据")
             if not sql_cert_id_list: