feat(mysql_pool): 实现带断连重试的SQL执行封装

- 新增 _execute 方法，支持 SQL 执行自动断连重试和指数退避
- 新增 _execute_many 方法，支持 executemany 断连重试和指数退避
- insert_many 方法批量插入时遇重复自动降级逐条插入处理
- update_all 和批量更新接口采用重试封装，失败时降级单条更新
- 异常处理更细化，对完整性错误和连接错误分别处理
- 提升了数据库操作的稳定性和健壮性，避免分钟级网络波动影响

feat(yaml_loader): 优化 yaml 配置文件加载路径及环境变量解析

- 调整正则表达式，支持匹配配置中的环境变量引用格式
- 支持从主脚本目录查找配置文件，补充路径查找逻辑
- 读取 profile 配置时，优先加载并合并同目录下后缀文件
- 方法注释更详细，提升代码可读性和易用性

feat(get_token): 新增卡玩App登录态Token抓取脚本

- 基于 adb root 权限从 MuMu 模拟器读取卡玩 Flutter SharedPreferences
- 通过 grep 精准提取 flutter.auth_token 字段的 Sa-Token 风格 JWT
- 实现设备连接和 adb root 权限自动准备步骤，确保稳定获取 token
- 脚本支持命令行直接打印和爬虫中函数调用获取最新 token
- 增加完整异常处理和错误排查提示，提高使用体验

kawan_spider/YamlLoader.py

@@ -1,74 +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\.]+)\}$')
+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)

kawan_spider/get_token.py

@@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/08/21
+"""卡玩 App token 抓取脚本。
+
+MuMu 模拟器（已 root）里卡玩 App 登录后，登录态 JWT 存在 Flutter 的
+SharedPreferences 文件（FlutterSharedPreferences.xml）的 flutter.auth_token
+字段里。本脚本经 adb 以 root 身份直接读该文件，抠出当前有效的 token。
+
+该 token 是 Sa-Token 风格的 JWT（loginId=app_user:xxx + rnStr + HS256），
+换账号重登会覆盖旧值，所以读到的必然是此刻有效的那一条。请求时按接口实际情况
+放进请求头（卡玩大概率是 satoken 字段），本脚本返回裸 token，不拼任何前缀。
+
+用法：
+    python get_token.py            # 打印当前 token
+    from get_token import get_token  # 爬虫里直接调用，每次取最新 token
+"""
+import sys
+import subprocess
+
+# MuMu 12 模拟器的 adb 地址（多开实例端口可能不同，按需改，常见 16384 / 16416 ...）
+MUMU_ADB_ADDR = "127.0.0.1:16384"
+
+# 卡玩 App 包名
+PACKAGE_NAME = "com.kawan.kawanapp"
+
+# 当前登录态 token 所在的 Flutter SharedPreferences 文件（只存当前账号的一条）
+PREFS_PATH = f"/data/data/{PACKAGE_NAME}/shared_prefs/FlutterSharedPreferences.xml"
+
+# 设备端执行的 grep：只抠 flutter.auth_token 这一项的值
+# 原始 XML 形如 <string name="flutter.auth_token">TOKEN</string>，
+# 输出形如 flutter.auth_token">TOKEN，Python 侧再切出 TOKEN
+REMOTE_GREP_CMD = f"grep -o 'flutter.auth_token\">[^<]*' {PREFS_PATH}"
+
+# grep 输出里 token 前的分隔标记（值紧跟在 "> 之后）
+_VALUE_SEP = '">'
+
+
+def _run(cmd: list[str], timeout: int = 20) -> subprocess.CompletedProcess:
+    """执行一条命令并返回结果对象。
+
+    Args:
+        cmd (list[str]): 完整命令 argv 列表，来自各调用处拼好的 adb 命令。
+        timeout (int, optional): 超时秒数。Defaults to 20。
+
+    Returns:
+        subprocess.CompletedProcess: 执行结果，含 returncode / stdout / stderr。
+
+    Raises:
+        RuntimeError: 未找到 adb 可执行文件，或命令超时时抛出。
+    """
+    try:
+        return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
+    except FileNotFoundError as e:
+        # 环境里没有 adb 可执行文件（未装 platform-tools 或未加入 PATH）
+        raise RuntimeError(
+            f"[{sys._getframe().f_code.co_name}] 未找到 adb，请确认已把 platform-tools 加入 PATH"
+        ) from e
+    except subprocess.TimeoutExpired as e:
+        raise RuntimeError(
+            f"[{sys._getframe().f_code.co_name}] adb 命令超时，请检查模拟器与 adb 连接"
+        ) from e
+
+
+def _prepare_device() -> None:
+    """确保已连上 MuMu 并让 adbd 以 root 运行。
+
+    两步都幂等：已连接会提示 already connected，已 root 会提示 already running as
+    root，均不影响后续读取。连接 / 提权若失败，不在此中断，交由读取阶段统一暴露。
+
+    Returns:
+        None: 无返回值。
+    """
+    # 幂等连接：未连则建立，已连直接复用
+    _run(["adb", "connect", MUMU_ADB_ADDR])
+    # 幂等提权：MuMu 支持 adb root，读 App 私有目录需要 root 身份
+    _run(["adb", "-s", MUMU_ADB_ADDR, "root"])
+
+
+def get_token() -> str:
+    """从已登录的卡玩 App 抓取当前登录态 token。
+
+    先确保连上 MuMu 且 adbd 为 root，再以 root 读取 Flutter 的
+    SharedPreferences，grep 出 flutter.auth_token 的值。整条远程命令作为单个
+    字符串传给 adb shell，保证单引号在设备端正确解析。
+
+    Returns:
+        str: 当前有效的裸 token（Sa-Token 风格 JWT，不含任何前缀）。
+
+    Raises:
+        RuntimeError: 设备未连接 / 未 root / App 未登录 / 未抓到 token 时抛出。
+    """
+    _prepare_device()
+
+    # 整条远程命令作为单个字符串传参，避免 > 与 [^<]* 被本机 shell 或设备 shell 误解析
+    result = _run(["adb", "-s", MUMU_ADB_ADDR, "shell", REMOTE_GREP_CMD])
+
+    if result.returncode != 0:
+        # grep 未匹配（App 未登录 / 文件不存在）也会返回非 0，统一当作失败处理
+        err = result.stderr.strip() or result.stdout.strip() or "未知错误"
+        raise RuntimeError(
+            f"[{sys._getframe().f_code.co_name}] 抓取失败：{err}\n"
+            f"排查：1) 模拟器是否开着且 adb 能连上 {MUMU_ADB_ADDR} "
+            f"2) adb 是否已 root 3) 卡玩 App 是否已登录"
+        )
+
+    # 输出形如 flutter.auth_token">eyJ...，切出 "> 之后的 token（.strip 顺带去掉 \r）
+    token = ""
+    for line in result.stdout.splitlines():
+        line = line.strip()
+        if _VALUE_SEP in line:
+            token = line.split(_VALUE_SEP, 1)[1].strip()
+            if token:
+                break
+
+    if not token:
+        # 匹配到字段但值为空 = App 已退出登录 / 尚未登录
+        raise RuntimeError(
+            f"[{sys._getframe().f_code.co_name}] 未抓到 token，请确认卡玩 App 已登录并稳定运行几秒后再试"
+        )
+
+    return token
+
+
+if __name__ == "__main__":
+    try:
+        print(get_token())
+    except RuntimeError as exc:
+        print(exc)
+        sys.exit(1)

kawan_spider/mysql_pool.py

@@ -1,8 +1,9 @@
 # -*- coding: utf-8 -*-
 # Author : Charley
-# Python : 3.10.8
+# Python : 3.12.10
 # Date   : 2025/3/25 14:14
 import re
+import time
 import pymysql
 import YamlLoader
 from loguru import logger
@@ -50,27 +51,128 @@ class MySQLConnectionPool:
             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: 查询结果
+        执行SQL（带断连重试，指数退避，覆盖分钟级网络波动）。
+
+        Args:
+            query (str): SQL语句。
+            args: SQL参数，可以是元组、列表或 None。
+            commit (bool, optional): 是否提交事务。Defaults to False。
+
+        Returns:
+            cursor: pymysql cursor，调用方可读取 rowcount / lastrowid / fetchone 等。
+
+        Raises:
+            pymysql.err.IntegrityError: 唯一键冲突等完整性错误。
+            pymysql.err.OperationalError: 重试耗尽后仍无法连接数据库。
         """
-        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
+        for attempt in range(15):  # 指数退避：5s→10s→20s…最多 300s，覆盖约 10 分钟波动
+            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, pymysql.err.OperationalError) as e:
+                errno = e.args[0] if e.args else None
+                if errno in (2006, 2013):  # MySQL server gone / lost connection during query
+                    delay = min(5 * (2 ** attempt), 300)
+                    self.log.warning(
+                        f"数据库连接断开 (errno={errno})，{delay}s 后重试 "
+                        f"(attempt {attempt + 1}/15): {e}"
+                    )
+                    time.sleep(delay)
+                    continue
+                self.log.error(f"Error executing query: {e}, Query: {query}")
+                raise
+            except pymysql.err.IntegrityError:
+                raise
+            except Exception as e:
+                self.log.exception(f"Error executing query: {e}, Query: {query}, Args: {args}")
+                raise
+        raise pymysql.err.OperationalError(2013, f"重试 15 次后仍无法连接数据库，Query: {query[:100]}")
+
+    def _execute_many(self, query, args_list, commit=True):
+        """
+        executemany 的断连重试包装，指数退避，覆盖分钟级网络波动。
+
+        Args:
+            query (str): SQL语句。
+            args_list (list): 参数列表。
+            commit (bool, optional): 是否提交事务。Defaults to True。
+
+        Returns:
+            int: 影响行数（cursor.rowcount）。
+
+        Raises:
+            pymysql.err.IntegrityError: 唯一键冲突等完整性错误。
+            pymysql.err.OperationalError: 重试耗尽后仍无法连接数据库。
+        """
+        for attempt in range(15):
+            try:
+                with self.pool.connection() as conn:
+                    with conn.cursor() as cursor:
+                        cursor.executemany(query, args_list)
+                        if commit:
+                            conn.commit()
+                        return cursor.rowcount
+            except (pymysql.err.InterfaceError, pymysql.err.OperationalError) as e:
+                errno = e.args[0] if e.args else None
+                if errno in (2006, 2013):
+                    delay = min(5 * (2 ** attempt), 300)
+                    self.log.warning(
+                        f"数据库连接断开 (errno={errno})，{delay}s 后重试 "
+                        f"(attempt {attempt + 1}/15): {e}"
+                    )
+                    time.sleep(delay)
+                    continue
+                if commit:
+                    try:
+                        conn.rollback()
+                    except Exception:
+                        pass
+                raise
+            except pymysql.err.IntegrityError:
+                if commit:
+                    try:
+                        conn.rollback()
+                    except Exception:
+                        pass
+                raise
+            except Exception as e:
+                if commit:
+                    try:
+                        conn.rollback()
+                    except Exception:
+                        pass
+                self.log.exception(f"Error executing many: {e}, Query: {query}")
+                raise
+        raise pymysql.err.OperationalError(2013, f"重试 15 次后仍无法连接数据库，Query: {query[:100]}")
 
     def select_one(self, query, args=None):
         """
@@ -180,13 +282,14 @@ class MySQLConnectionPool:
             return cursor.lastrowid
         except pymysql.err.IntegrityError as e:
             if "Duplicate entry" in str(e):
-                self.log.warning(f"插入失败：重复条目，已跳过。错误详情: {e}")
+                # 重复条目用 warning 简短输出，不打印堆栈
+                self.log.warning(f"插入跳过-重复条目 Table: {table}, {e.args[1] if len(e.args) > 1 else e}")
                 return -1  # 返回 -1 表示重复条目被跳过
             else:
-                self.log.exception(f"数据库完整性错误: {e}")
+                self.log.error(f"数据库完整性错误 Table: {table}, Error: {e}")
                 raise
         except Exception as e:
-            self.log.exception(f"未知错误: {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,
@@ -220,37 +323,26 @@ class MySQLConnectionPool:
         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
+                total += self._execute_many(query, batch, commit=commit)
             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  # 单条插入时手动捕获重复
+                                        ignore=False
                                     )
                                 else:
-                                    # 原始SQL模式
                                     self.insert_one(query, args)
                                 rowcount += 1
                             except pymysql.err.IntegrityError as e2:
@@ -263,16 +355,10 @@ class MySQLConnectionPool:
                         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}")
@@ -308,17 +394,10 @@ class MySQLConnectionPool:
         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
+                total += self._execute_many(query, batch, commit=commit)
             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:
@@ -334,13 +413,9 @@ class MySQLConnectionPool:
                     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
@@ -356,11 +431,8 @@ class MySQLConnectionPool:
         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)}")
+                self._execute_many(query, batch, commit=True)
+                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}")
                 # 当前批次降级为单条插入
@@ -382,17 +454,11 @@ class MySQLConnectionPool:
         :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._execute_many(query, args_list, commit=True)
             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
@@ -400,11 +466,6 @@ class MySQLConnectionPool:
                 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):
         """
@@ -526,13 +587,9 @@ class MySQLConnectionPool:
 
                 # 执行批量更新
                 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}")
+                    rowcount = self._execute_many(base_query, batch_args, commit=commit)
+                    total += rowcount
+                    self.log.debug(f"Batch update succeeded. Rows: {rowcount}")
                 except Exception as e:
                     if commit:
                         conn.rollback()
@@ -555,16 +612,10 @@ class MySQLConnectionPool:
             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}")
+                    rowcount = self._execute_many(query, batch_args, commit=commit)
+                    total += rowcount
+                    self.log.debug(f"Batch update succeeded. Rows: {rowcount}")
                 except Exception as e:
-                    if commit:
-                        conn.rollback()
                     self.log.error(f"Batch update failed: {e}")
                     # 降级为单条更新
                     for args in batch_args: