提交

超星数据移植与定时同步/修复说明.md

@@ -0,0 +1,92 @@
+# 请求学生信息API.py 修复说明
+
+## 问题描述
+脚本执行30多分钟后报错：
+- `(1053, 'Server shutdown in progress')` - MySQL服务器正在关闭
+- `(2013, 'Lost connection to MySQL server during query')` - 连接丢失
+
+## 根本原因
+长时间运行的脚本，MySQL连接超时或断开，导致数据库操作失败。原代码没有：
+1. 连接保活机制
+2. 自动重连机制
+3. 异常恢复能力
+
+## 解决方案
+
+### 1. 添加数据库连接管理
+在`__init__`方法中添加：
+```python
+self._db_connection = None
+self._last_ping_time = 0
+self._ping_interval = 60  # 每60秒ping一次数据库
+```
+
+### 2. 实现连接池和保活机制
+新增`_get_db_connection()`方法：
+- 创建带超时配置的数据库连接
+- 定期ping数据库保持连接活跃（每60秒）
+- 自动检测并重连断开的连接
+
+### 3. 实现重试机制
+新增`_execute_with_retry()`方法：
+- 捕获特定的连接错误（2013, 2006, 1053）
+- 最多重试3次
+- 递增等待时间（2秒、4秒、6秒）
+
+### 4. 重构主要方法
+
+#### `get_and_save_students_batch()`
+- 使用`_execute_with_retry()`包装数据库操作
+- 在批量提交时自动ping连接
+- 统一的异常处理和资源释放
+
+#### `sync_student_msg()`
+- 使用`_execute_with_retry()`包装整个同步逻辑
+- 确保连接断开时自动重试
+
+#### `_insert_into_target()`
+- 在每100条记录批量提交前ping连接
+- 确保长时间运行时连接保持活跃
+
+### 5. 改进错误处理
+- 添加详细的错误堆栈信息（traceback）
+- 失败时返回已保存的数量而不是0
+- 更好的日志输出
+
+## 主要改进点
+
+| 改进项 | 说明 |
+|--------|------|
+| 连接超时配置 | connect_timeout=30, read_timeout=60, write_timeout=60 |
+| 连接保活 | 每60秒自动ping数据库 |
+| 自动重连 | 连接断开时自动重建连接 |
+| 重试机制 | 遇到连接错误最多重试3次 |
+| 批量提交优化 | 每批提交前ping连接 |
+| 错误日志 | 添加完整的traceback信息 |
+| 资源管理 | 使用finally确保连接关闭 |
+
+## 使用建议
+
+1. **运行前检查MySQL配置**：
+   - `wait_timeout` 和 `interactive_timeout` 建议设置为 28800（8小时）
+   - `max_allowed_packet` 建议设置足够大（如16M）
+
+2. **监控日志**：
+   - 观察"创建新的数据库连接"消息
+   - 检查"连接ping失败"警告
+   - 注意重试次数和等待时间
+
+3. **调整参数**（如需要）：
+   - `_ping_interval`: 默认60秒，可根据MySQL的wait_timeout调整
+   - `max_retries`: 默认3次，可根据网络稳定性调整
+   - `batch_size`: 默认100条，可根据数据大小调整
+
+## 测试建议
+
+1. 运行脚本观察是否能正常完成
+2. 检查日志中的连接管理信息
+3. 验证数据完整性
+4. 如果仍有问题，可以：
+   - 减小`batch_size`（如改为50）
+   - 减小`_ping_interval`（如改为30秒）
+   - 增加MySQL的`wait_timeout`值

超星数据移植与定时同步/请求学生信息API.py

@@ -14,6 +14,8 @@ from typing import Dict, Any, Optional, List, Tuple
 
 import bcrypt
 import pymysql
+import pymysql.cursors
+from pymysql.cursors import DictCursor
 import requests
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import unpad
@@ -52,6 +54,10 @@ class ChaoxingStudentAPI:
         self.api_path = "xsjbxxpi_66247"
         # 生成AES解密密钥：appId去除'-'后取前16位
         self.aes_key = app_id.replace("-", "")[:16].encode('utf-8')
+        # 数据库连接对象
+        self._db_connection = None
+        self._last_ping_time = 0
+        self._ping_interval = 60  # 每60秒ping一次数据库
     
     def generate_enc(self, page_num: int, page_size: int) -> str:
         """
@@ -254,6 +260,100 @@ class ChaoxingStudentAPI:
         print(f"\n总共获取 {len(all_data)} 条学生信息")
         return all_data
     
+    def _get_db_connection(self, force_new: bool = False):
+        """
+        获取数据库连接，自动处理重连
+        
+        Args:
+            force_new: 是否强制创建新连接
+            
+        Returns:
+            数据库连接对象
+        """
+        if force_new and self._db_connection:
+            try:
+                self._db_connection.close()
+            except:
+                pass
+            self._db_connection = None
+        
+        if self._db_connection is None:
+            print("创建新的数据库连接...")
+            self._db_connection = pymysql.connect(
+                **DB_CONFIG,
+                cursorclass=DictCursor,
+                autocommit=False,
+                connect_timeout=30,
+                read_timeout=60,
+                write_timeout=60
+            )
+            self._last_ping_time = time.time()
+            print("✓ 数据库连接已建立")
+        else:
+            # 检查连接是否需要ping
+            current_time = time.time()
+            if current_time - self._last_ping_time > self._ping_interval:
+                try:
+                    self._db_connection.ping(reconnect=True)
+                    self._last_ping_time = current_time
+                except Exception as e:
+                    print(f"连接ping失败，重新建立连接: {e}")
+                    self._db_connection = None
+                    return self._get_db_connection(force_new=True)
+        
+        return self._db_connection
+    
+    def _close_db_connection(self):
+        """
+        关闭数据库连接
+        """
+        if self._db_connection:
+            try:
+                self._db_connection.close()
+                print("✓ 数据库连接已关闭")
+            except:
+                pass
+            self._db_connection = None
+    
+    def _execute_with_retry(self, func, max_retries: int = 3):
+        """
+        执行数据库操作并在失败时重试
+        
+        Args:
+            func: 要执行的函数，接收connection参数
+            max_retries: 最大重试次数
+            
+        Returns:
+            函数执行结果
+        """
+        last_error = None
+        for attempt in range(max_retries):
+            try:
+                connection = self._get_db_connection()
+                result = func(connection)
+                return result
+            except (pymysql.err.OperationalError, pymysql.err.InterfaceError) as e:
+                last_error = e
+                error_code = e.args[0] if e.args else 0
+                # 2013: Lost connection to MySQL server during query
+                # 2006: MySQL server has gone away
+                # 1053: Server shutdown in progress
+                if error_code in (2013, 2006, 1053):
+                    print(f"数据库连接错误 (尝试 {attempt+1}/{max_retries}): {e}")
+                    self._db_connection = None
+                    if attempt < max_retries - 1:
+                        wait_time = (attempt + 1) * 2  # 递增等待时间
+                        print(f"等待 {wait_time} 秒后重试...")
+                        time.sleep(wait_time)
+                    continue
+                else:
+                    raise
+            except Exception as e:
+                print(f"数据库操作失败: {e}")
+                raise
+        
+        raise last_error if last_error else Exception("数据库操作失败")
+    
     def clear_table(self) -> bool:
         """
         清空学生数据表
@@ -380,70 +480,70 @@ class ChaoxingStudentAPI:
         total_pages = None
         
         try:
-            connection = pymysql.connect(**DB_CONFIG)
-            try:
-                current_time = int(time.time() * 1000)  # 毫秒时间戳
-                table_name = TABLE_MAPPING["student"]
+            current_time = int(time.time() * 1000)  # 毫秒时间戳
+            table_name = TABLE_MAPPING["student"]
+            
+            # 构建插入SQL
+            insert_sql = f"""
+            INSERT INTO `{table_name}` (
+                `inserttime`, `updatetime`,
+                `xsjbxx_xm`, `xsjbxx_xh`, `xsjbxx_lxdh`, `xsjbxx_sfzjlx`,
+                `xsjbxx_sfzjh`, `dataMigrateUniqueId`, `xsjbxx_bjxx`, `xsjbxx_bjbh`,
+                `xsjbxx_szxq`, `xsjbxx_xqdm`, `xsjbxx_sznj`, `xsjbxx_rxnj`,
+                `xsjbxx_yxxx`, `xsjbxx_yxdm`, `xsjbxx_zyxx`, `xsjbxx_zydm`,
+                `xsjbxx_bzrgh`, `xsjbxx_xsdqzt`, `xsjbxx_xssfzx`, `xsjbxx_xjydlx`,
+                `xsjbxx_pycc`, `xsjbxx_xz`, `xsjbxx_xb`, `xsjbxx_csrq`,
+                `xsjbxx_gj`, `xsjbxx_jtzz`, `xsjbxx_jtyzbm`, `xsjbxx_jtlxr`,
+                `xsjbxx_jtdh`, `xsjbxx_fmxm1`, `xsjbxx_fmxm2`, `xsjbxx_cygx1`,
+                `xsjbxx_cygx2`, `xsjbxx_cysfwjhr1`, `xsjbxx_cysfwjhr2`,
+                `xsjbxx_fmsjh1`, `xsjbxx_fmsjh2`, `xsjbxx_fmzjlx1`, `xsjbxx_fmzjlx2`,
+                `xsjbxx_fmzjh1`, `xsjbxx_fmzjh2`, `xsjbxx_xmpy`, `xsjbxx_xjzt`
+            ) VALUES (
+                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
+                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
+                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
+                %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
+                %s, %s, %s, %s, %s
+            )
+            """
+            
+            while True:
+                print(f"\n正在获取第 {page_num} 页数据...")
+                result = self.get_student_list(page_num, max_page_size)
                 
-                # 构建插入SQL
-                insert_sql = f"""
-                INSERT INTO `{table_name}` (
-                    `inserttime`, `updatetime`,
-                    `xsjbxx_xm`, `xsjbxx_xh`, `xsjbxx_lxdh`, `xsjbxx_sfzjlx`,
-                    `xsjbxx_sfzjh`, `dataMigrateUniqueId`, `xsjbxx_bjxx`, `xsjbxx_bjbh`,
-                    `xsjbxx_szxq`, `xsjbxx_xqdm`, `xsjbxx_sznj`, `xsjbxx_rxnj`,
-                    `xsjbxx_yxxx`, `xsjbxx_yxdm`, `xsjbxx_zyxx`, `xsjbxx_zydm`,
-                    `xsjbxx_bzrgh`, `xsjbxx_xsdqzt`, `xsjbxx_xssfzx`, `xsjbxx_xjydlx`,
-                    `xsjbxx_pycc`, `xsjbxx_xz`, `xsjbxx_xb`, `xsjbxx_csrq`,
-                    `xsjbxx_gj`, `xsjbxx_jtzz`, `xsjbxx_jtyzbm`, `xsjbxx_jtlxr`,
-                    `xsjbxx_jtdh`, `xsjbxx_fmxm1`, `xsjbxx_fmxm2`, `xsjbxx_cygx1`,
-                    `xsjbxx_cygx2`, `xsjbxx_cysfwjhr1`, `xsjbxx_cysfwjhr2`,
-                    `xsjbxx_fmsjh1`, `xsjbxx_fmsjh2`, `xsjbxx_fmzjlx1`, `xsjbxx_fmzjlx2`,
-                    `xsjbxx_fmzjh1`, `xsjbxx_fmzjh2`, `xsjbxx_xmpy`, `xsjbxx_xjzt`
-                ) VALUES (
-                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
-                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
-                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
-                    %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
-                    %s, %s, %s, %s, %s
-                )
-                """
+                # 超星API成功时code=1，失败时code!=1
+                if result.get("code") != 1:
+                    print(f"获取第 {page_num} 页失败: {result.get('message', '未知错误')}")
+                    break
                 
-                with connection.cursor() as cursor:
-                    while True:
-                        print(f"\n正在获取第 {page_num} 页数据...")
-                        result = self.get_student_list(page_num, max_page_size)
-                        
-                        # 超星API成功时code=1，失败时code!=1
-                        if result.get("code") != 1:
-                            print(f"获取第 {page_num} 页失败: {result.get('message', '未知错误')}")
-                            break
-                        
-                        # 获取分页信息（从extendData中）
-                        extend_data = result.get("extendData", {})
-                        if total_pages is None and extend_data:
-                            total_pages = int(extend_data.get("pages", 0))
-                            total_count = int(extend_data.get("count", 0))
-                            print(f"总页数: {total_pages}, 总条数: {total_count}")
-                        
-                        data = result.get("data")
-                        if not data:
-                            print("没有更多数据")
-                            break
-                        
-                        # 处理返回的数据
-                        items = []
-                        if isinstance(data, list):
-                            items = data
-                        elif isinstance(data, dict):
-                            items = data.get("list", data.get("records", []))
-                        
-                        if not items:
-                            break
-                        
-                        print(f"第 {page_num} 页获取 {len(items)} 条数据")
-                        
-                        # 逐条插入数据
+                # 获取分页信息（从extendData中）
+                extend_data = result.get("extendData", {})
+                if total_pages is None and extend_data:
+                    total_pages = int(extend_data.get("pages", 0))
+                    total_count = int(extend_data.get("count", 0))
+                    print(f"总页数: {total_pages}, 总条数: {total_count}")
+                
+                data = result.get("data")
+                if not data:
+                    print("没有更多数据")
+                    break
+                
+                # 处理返回的数据
+                items = []
+                if isinstance(data, list):
+                    items = data
+                elif isinstance(data, dict):
+                    items = data.get("list", data.get("records", []))
+                
+                if not items:
+                    break
+                
+                print(f"第 {page_num} 页获取 {len(items)} 条数据")
+                
+                # 使用重试机制插入数据
+                def insert_batch(connection):
+                    nonlocal total_saved
+                    with connection.cursor() as cursor:
                         for item in items:
                             try:
                                 cursor.execute(insert_sql, (
@@ -469,28 +569,33 @@ class ChaoxingStudentAPI:
                             except Exception as e:
                                 print(f"插入单条数据失败: {e}, 数据: {item}")
                         
-                        # 如果知道总页数，检查是否已获取完所有页
-                        if total_pages and page_num >= total_pages:
-                            break
-                        # 如果返回的数据少于page_size，说明已经是最后一页
-                        if len(items) < max_page_size:
-                            break
-                        
-                        page_num += 1
+                        # 提交当前批次剩余数据
+                        if total_saved % batch_size != 0:
+                            connection.commit()
                     
-                    # 提交剩余数据
-                    if total_saved % batch_size != 0:
-                        connection.commit()
-                        remaining = total_saved % batch_size
-                        print(f"✓ 分批提交剩余 {remaining} 条数据到表 {table_name}，累计 {total_saved} 条")
-                    
-                    print(f"\n✓ 总共成功保存 {total_saved} 条数据到表 {table_name}")
                     return total_saved
-            finally:
-                connection.close()
+                
+                # 执行插入操作（带重试）
+                self._execute_with_retry(insert_batch)
+                
+                # 如果知道总页数，检查是否已获取完所有页
+                if total_pages and page_num >= total_pages:
+                    break
+                # 如果返回的数据少于page_size，说明已经是最后一页
+                if len(items) < max_page_size:
+                    break
+                
+                page_num += 1
+            
+            print(f"\n✓ 总共成功保存 {total_saved} 条数据到表 {table_name}")
+            return total_saved
         except Exception as e:
             print(f"✗ 获取并保存数据失败: {e}")
-            return 0
+            import traceback
+            traceback.print_exc()
+            return total_saved  # 返回已保存的数量
+        finally:
+            self._close_db_connection()
 
     # --------- 新增的方法：同步到xg_student_msg ---------
 
@@ -502,28 +607,33 @@ class ChaoxingStudentAPI:
             成功同步的记录数
         """
         try:
-            connection = pymysql.connect(**DB_CONFIG, cursorclass=pymysql.cursors.DictCursor)
-            try:
+            def do_sync(connection):
                 source_rows = self._fetch_source_students(connection)
                 if not source_rows:
                     print("cx_xssj表暂无可同步数据")
-                    return 0
+                    return 0, 0, 0, 0
 
                 print(f"待同步数据量: {len(source_rows)} 条")
 
                 success, user_inserted, user_skipped, role_inserted = self._insert_into_target(connection, source_rows)
 
                 connection.commit()
-                print(f"✓ 已同步 {success}/{len(source_rows)} 条数据至xg_student_msg")
-                if success:
-                    print(f"✓ 新增 {user_inserted} 个sys_user账号，跳过 {user_skipped} 个已存在账号")
-                    print(f"✓ 新增 {role_inserted} 条学生角色绑定 (role_id={self.STUDENT_ROLE_ID})")
-                return success
-            finally:
-                connection.close()
+                return success, user_inserted, user_skipped, role_inserted
+            
+            success, user_inserted, user_skipped, role_inserted = self._execute_with_retry(do_sync)
+            
+            print(f"✓ 已同步 {success} 条数据至xg_student_msg")
+            if success:
+                print(f"✓ 新增 {user_inserted} 个sys_user账号，跳过 {user_skipped} 个已存在账号")
+                print(f"✓ 新增 {role_inserted} 条学生角色绑定 (role_id={self.STUDENT_ROLE_ID})")
+            return success
         except Exception as exc:
             print(f"✗ 同步到xg_student_msg失败: {exc}")
+            import traceback
+            traceback.print_exc()
             return 0
+        finally:
+            self._close_db_connection()
 
     def _fetch_source_students(self, connection) -> List[Dict[str, Any]]:
         """
@@ -600,7 +710,7 @@ class ChaoxingStudentAPI:
 
                 unique_id = self._generate_student_id(timestamp_str, sequence)
                 gender = self._map_gender(row.get("xsjbxx_xb"))
-                status = self._map_student_status(row.get("xsjbxx_xjzt"))
+                status = self._map_student_status(row.get("xsjbxx_xsdqzt"))
 
                 class_id, class_name = self._find_class_dept(
                     connection,
@@ -669,6 +779,11 @@ class ChaoxingStudentAPI:
                         role_inserted_total += self._ensure_user_role(connection, student_profile["user_id"], self.STUDENT_ROLE_ID)
 
                     if batch_inserted >= 100:
+                        # 批量提交前ping一下连接
+                        try:
+                            connection.ping(reconnect=True)
+                        except Exception as e:
+                            print(f"警告: 连接ping失败: {e}")
                         connection.commit()
                         print("✓ xg_student_msg分批提交100条记录")
                         batch_inserted = 0
@@ -709,13 +824,28 @@ class ChaoxingStudentAPI:
 
     def _map_student_status(self, status_text: Optional[str]) -> Optional[int]:
         """
-        将学籍状态文本映射为目标表字段
+        将学生当前状态（xsjbxx_xsdqzt）文本映射为目标表字段
         """
-        if status_text == "在籍":
+        if not status_text:
             return 1
-        if status_text == "不在籍":
-            return 6
-        return None
+
+        mapping = {
+            "毕业": 5,
+            "退学": 4,
+            "结业": 7,
+            "在读": 1,
+            "停学": 8,
+            "分流": 9,
+            "休学": 2,
+            "复学": 1,
+        }
+
+        normalized = str(status_text).strip()
+        mapped_value = mapping.get(normalized)
+        if mapped_value is None:
+            print(f"警告: 未配置的学生当前状态 -> raw='{status_text}'，默认使用1(在读)")
+            return 1
+        return mapped_value
 
     def _find_class_dept(
         self,
@@ -889,6 +1019,8 @@ class ChaoxingStudentAPI:
             return 1, 0
         except Exception as exc:
             print(f"插入sys_user失败: {exc}, user_id={user_id}, student_no={student['student_no']}")
+            import traceback
+            traceback.print_exc()
             return 0, 1
 
     def _ensure_user_role(self, connection, user_id: int, role_id: int) -> int:

超星数据移植与定时同步/请求系部信息API.py

@@ -334,7 +334,7 @@ class DepartmentSync:
         self.parent_ancestors = '0,100,101'  # 父节点的ancestors
         self.tenant_id = '000000'
         # 排除的部门名称列表（不同步到sys_dept表）
-        self.excluded_dept_names = ['教务处', '基础部', '外聘']
+        self.excluded_dept_names = ['教务处', '外聘']
     
     def get_valid_departments(self) -> List[Dict[str, Any]]:
         """