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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
超星系统 - 学生基本信息API请求脚本
API路径: xsjbxxpi_66247/list
"""

import hashlib
import base64
import json
import time
from datetime import datetime
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

from db_config import DB_CONFIG, TABLE_MAPPING


class ChaoxingStudentAPI:
    """超星学生基本信息API请求类"""

    DEFAULT_TENANT_ID = "000000"
    DEFAULT_CREATE_BY = 1
    DEFAULT_UPDATE_BY = 1
    DEFAULT_ACCOUNT_TYPE = "2"
    DEFAULT_USER_TYPE = "sys_user"
    DEFAULT_PASSWORD = "Hzdyjs123456!@#"
    STUDENT_ROLE_ID = 1938120227459309569
    SYS_USER_REMARK = "学生信息系统自动创建(超星同步)"
    SYS_USER_STATUS = "0"
    SYS_USER_DEL_FLAG = "0"
    QUERY_BATCH_SIZE = 1000
    
    def __init__(self, app_id: str, app_key: str, fid: str = "19628"):
        """
        初始化API客户端
        
        Args:
            app_id: API的appId
            app_key: API的appKey
            fid: 单位ID，默认19628
        """
        self.app_id = app_id
        self.app_key = app_key
        self.fid = fid
        self.base_url = "https://appswh.chaoxing.com/biapi/api"
        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:
        """
        生成enc加密验证串
        
        enc加密方式：32位大写的MD5字符串
        格式：[apiPath=xsjbxxpi_66247#$pageNum=页码#$pageSize=单页条数#$fid=19628#$appKey=授权appKey]
        
        Args:
            page_num: 页码
            page_size: 单页条数
            
        Returns:
            32位大写MD5字符串
        """
        # 构建加密字符串
        enc_string = f"[apiPath={self.api_path}#$pageNum={page_num}#$pageSize={page_size}#$fid={self.fid}#$appKey={self.app_key}]"
        
        # MD5加密并转为大写
        md5_hash = hashlib.md5(enc_string.encode('utf-8')).hexdigest().upper()
        
        return md5_hash
    
    def decrypt_data(self, ciphertext: str) -> Any:
        """
        AES解密data字段
        
        解密算法：AES/ECB/PKCS5Padding
        密钥：appId去除'-'后前16位字符
        
        Args:
            ciphertext: Base64编码的密文
            
        Returns:
            解密后的数据（可能是JSON字符串，需要进一步解析）
        """
        try:
            # Base64解码
            encrypted_bytes = base64.b64decode(ciphertext)
            
            # AES解密
            cipher = AES.new(self.aes_key, AES.MODE_ECB)
            decrypted_bytes = cipher.decrypt(encrypted_bytes)
            
            # PKCS5Padding去填充
            decrypted_bytes = unpad(decrypted_bytes, AES.block_size)
            
            # 转为字符串
            decrypted_text = decrypted_bytes.decode('utf-8')
            
            # 尝试解析为JSON
            try:
                return json.loads(decrypted_text)
            except json.JSONDecodeError:
                # 如果不是JSON，直接返回字符串
                return decrypted_text
                
        except Exception as e:
            print(f"解密失败: {e}")
            return None
    
    def get_student_list(self, page_num: int = 1, page_size: int = 100) -> Dict[str, Any]:
        """
        获取学生基本信息列表
        
        Args:
            page_num: 页码，默认1
            page_size: 页条数，默认100，最大1000
            
        Returns:
            API响应结果
        """
        # 验证page_size
        if page_size > 1000:
            page_size = 1000
            print("警告: pageSize超过最大限制1000，已自动调整为1000")
        
        # 生成enc加密串
        enc = self.generate_enc(page_num, page_size)
        
        # 构建请求URL
        url = f"{self.base_url}/{self.api_path}/list"
        
        # 构建请求参数
        params = {
            "pageNum": page_num,
            "pageSize": page_size,
            "fid": self.fid,
            "enc": enc
        }
        
        # 构建请求头
        headers = {
            "appId": self.app_id,
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        }
        
        try:
            # 发送GET请求
            print(f"正在请求: {url}")
            print(f"参数: {params}")
            print(f"请求头appId: {self.app_id}")
            
            response = requests.get(url, params=params, headers=headers, timeout=30)
            
            # 检查HTTP状态码
            response.raise_for_status()
            
            # 解析JSON响应
            result = response.json()
            
            # 解密data字段
            if result.get("data") and isinstance(result.get("data"), str):
                decrypted_data = self.decrypt_data(result["data"])
                if decrypted_data is not None:
                    result["data"] = decrypted_data
                else:
                    print("警告: data字段解密失败，保持原始值")
            
            return result
            
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            return {
                "code": -1,
                "message": f"请求异常: {str(e)}",
                "data": None
            }
        except ValueError as e:
            print(f"JSON解析失败: {e}")
            return {
                "code": -1,
                "message": f"响应解析失败: {str(e)}",
                "data": None
            }
    
    def get_all_students(self, max_page_size: int = 500) -> list:
        """
        获取所有学生基本信息（自动分页）
        
        Args:
            max_page_size: 每页最大条数，默认500
            
        Returns:
            所有学生信息列表
        """
        all_data = []
        page_num = 1
        total_pages = None  # 总页数，从第一页响应中获取
        
        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
            
            # 如果是列表，直接添加
            if isinstance(data, list):
                all_data.extend(data)
                print(f"第 {page_num} 页获取 {len(data)} 条数据，累计 {len(all_data)} 条")
                
                # 如果知道总页数，检查是否已获取完所有页
                if total_pages and page_num >= total_pages:
                    break
                # 如果返回的数据少于page_size，说明已经是最后一页
                if len(data) < max_page_size:
                    break
            # 如果是字典，可能包含分页信息
            elif isinstance(data, dict):
                items = data.get("list", data.get("records", []))
                if items:
                    all_data.extend(items)
                    print(f"第 {page_num} 页获取 {len(items)} 条数据，累计 {len(all_data)} 条")
                    # 检查是否还有下一页
                    total = data.get("total", 0)
                    if total and len(all_data) >= total:
                        break
                else:
                    break
            else:
                break
            
            page_num += 1
        
        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:
        """
        清空学生数据表
        
        Returns:
            是否成功
        """
        try:
            connection = pymysql.connect(**DB_CONFIG)
            try:
                with connection.cursor() as cursor:
                    table_name = TABLE_MAPPING["student"]
                    cursor.execute(f"TRUNCATE TABLE `{table_name}`")
                    connection.commit()
                    print(f"✓ 已清空表 {table_name}")
                    return True
            finally:
                connection.close()
        except Exception as e:
            print(f"✗ 清空表失败: {e}")
            return False
    
    def save_to_database(self, data_list: List[Dict], batch_size: int = 100) -> int:
        """
        将学生数据保存到数据库（支持分批提交）
        
        Args:
            data_list: 学生数据列表
            batch_size: 每批提交的记录数，默认100
            
        Returns:
            成功插入的记录数
        """
        if not data_list:
            print("数据列表为空，跳过保存")
            return 0
        
        try:
            connection = pymysql.connect(**DB_CONFIG)
            try:
                current_time = int(time.time() * 1000)  # 毫秒时间戳
                table_name = TABLE_MAPPING["student"]
                
                with connection.cursor() as cursor:
                    # 构建插入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
                    )
                    """
                    
                    success_count = 0
                    batch_count = 0
                    for item in data_list:
                        try:
                            cursor.execute(insert_sql, (
                                current_time, current_time,  # inserttime, updatetime
                                item.get('xsjbxx_xm'), item.get('xsjbxx_xh'), item.get('xsjbxx_lxdh'), item.get('xsjbxx_sfzjlx'),
                                item.get('xsjbxx_sfzjh'), item.get('dataMigrateUniqueId'), item.get('xsjbxx_bjxx'), item.get('xsjbxx_bjbh'),
                                item.get('xsjbxx_szxq'), item.get('xsjbxx_xqdm'), item.get('xsjbxx_sznj'), item.get('xsjbxx_rxnj'),
                                item.get('xsjbxx_yxxx'), item.get('xsjbxx_yxdm'), item.get('xsjbxx_zyxx'), item.get('xsjbxx_zydm'),
                                item.get('xsjbxx_bzrgh'), item.get('xsjbxx_xsdqzt'), item.get('xsjbxx_xssfzx'), item.get('xsjbxx_xjydlx'),
                                item.get('xsjbxx_pycc'), item.get('xsjbxx_xz'), item.get('xsjbxx_xb'), item.get('xsjbxx_csrq'),
                                item.get('xsjbxx_gj'), item.get('xsjbxx_jtzz'), item.get('xsjbxx_jtyzbm'), item.get('xsjbxx_jtlxr'),
                                item.get('xsjbxx_jtdh'), item.get('xsjbxx_fmxm1'), item.get('xsjbxx_fmxm2'), item.get('xsjbxx_cygx1'),
                                item.get('xsjbxx_cygx2'), item.get('xsjbxx_cysfwjhr1'), item.get('xsjbxx_cysfwjhr2'),
                                item.get('xsjbxx_fmsjh1'), item.get('xsjbxx_fmsjh2'), item.get('xsjbxx_fmzjlx1'), item.get('xsjbxx_fmzjlx2'),
                                item.get('xsjbxx_fmzjh1'), item.get('xsjbxx_fmzjh2'), item.get('xsjbxx_xmpy'), item.get('xsjbxx_xjzt')
                            ))
                            success_count += 1
                            batch_count += 1
                            
                            # 达到批次大小时提交
                            if batch_count >= batch_size:
                                connection.commit()
                                print(f"✓ 分批提交 {batch_count} 条数据到表 {table_name}，累计 {success_count} 条")
                                batch_count = 0
                        except Exception as e:
                            print(f"插入单条数据失败: {e}, 数据: {item}")
                    
                    # 提交剩余数据
                    if batch_count > 0:
                        connection.commit()
                        print(f"✓ 分批提交剩余 {batch_count} 条数据到表 {table_name}，累计 {success_count} 条")
                    
                    print(f"✓ 总共成功插入 {success_count}/{len(data_list)} 条数据到表 {table_name}")
                    return success_count
            finally:
                connection.close()
        except Exception as e:
            print(f"✗ 保存数据到数据库失败: {e}")
            return 0
    
    def get_and_save_students_batch(self, max_page_size: int = 500, batch_size: int = 100) -> int:
        """
        获取所有学生信息并分批保存到cx_xssj表
        
        Args:
            max_page_size: API每页最大条数，默认500
            batch_size: 每批提交到数据库的记录数，默认100
            
        Returns:
            成功保存的总记录数
        """
        total_saved = 0
        page_num = 1
        total_pages = None
        
        try:
            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)
                
                # 超星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)} 条数据")
                
                # 使用重试机制插入数据
                def insert_batch(connection):
                    nonlocal total_saved
                    with connection.cursor() as cursor:
                        for item in items:
                            try:
                                cursor.execute(insert_sql, (
                                    current_time, current_time,  # inserttime, updatetime
                                    item.get('xsjbxx_xm'), item.get('xsjbxx_xh'), item.get('xsjbxx_lxdh'), item.get('xsjbxx_sfzjlx'),
                                    item.get('xsjbxx_sfzjh'), item.get('dataMigrateUniqueId'), item.get('xsjbxx_bjxx'), item.get('xsjbxx_bjbh'),
                                    item.get('xsjbxx_szxq'), item.get('xsjbxx_xqdm'), item.get('xsjbxx_sznj'), item.get('xsjbxx_rxnj'),
                                    item.get('xsjbxx_yxxx'), item.get('xsjbxx_yxdm'), item.get('xsjbxx_zyxx'), item.get('xsjbxx_zydm'),
                                    item.get('xsjbxx_bzrgh'), item.get('xsjbxx_xsdqzt'), item.get('xsjbxx_xssfzx'), item.get('xsjbxx_xjydlx'),
                                    item.get('xsjbxx_pycc'), item.get('xsjbxx_xz'), item.get('xsjbxx_xb'), item.get('xsjbxx_csrq'),
                                    item.get('xsjbxx_gj'), item.get('xsjbxx_jtzz'), item.get('xsjbxx_jtyzbm'), item.get('xsjbxx_jtlxr'),
                                    item.get('xsjbxx_jtdh'), item.get('xsjbxx_fmxm1'), item.get('xsjbxx_fmxm2'), item.get('xsjbxx_cygx1'),
                                    item.get('xsjbxx_cygx2'), item.get('xsjbxx_cysfwjhr1'), item.get('xsjbxx_cysfwjhr2'),
                                    item.get('xsjbxx_fmsjh1'), item.get('xsjbxx_fmsjh2'), item.get('xsjbxx_fmzjlx1'), item.get('xsjbxx_fmzjlx2'),
                                    item.get('xsjbxx_fmzjh1'), item.get('xsjbxx_fmzjh2'), item.get('xsjbxx_xmpy'), item.get('xsjbxx_xjzt')
                                ))
                                total_saved += 1
                                
                                # 达到批次大小时提交
                                if total_saved % batch_size == 0:
                                    connection.commit()
                                    print(f"✓ 分批提交 {batch_size} 条数据到表 {table_name}，累计 {total_saved} 条")
                            except Exception as e:
                                print(f"插入单条数据失败: {e}, 数据: {item}")
                        
                        # 提交当前批次剩余数据
                        if total_saved % batch_size != 0:
                            connection.commit()
                    
                    return total_saved
                
                # 执行插入操作（带重试）
                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}")
            import traceback
            traceback.print_exc()
            return total_saved  # 返回已保存的数量
        finally:
            self._close_db_connection()

    # --------- 新增的方法：同步到xg_student_msg ---------

    def sync_student_msg(self) -> int:
        """
        将cx_xssj表中的学生数据同步到xg_student_msg表

        Returns:
            成功同步的记录数
        """
        try:
            def do_sync(connection):
                source_rows = self._fetch_source_students(connection)
                if not source_rows:
                    print("cx_xssj表暂无可同步数据")
                    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()
                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]]:
        """
        从cx_xssj表读取需要同步的学生数据
        """
        select_sql = """
        SELECT
            xsjbxx_xm,
            xsjbxx_xh,
            xsjbxx_lxdh,
            xsjbxx_sfzjh,
            xsjbxx_xb,
            xsjbxx_csrq,
            xsjbxx_xjzt,
            xsjbxx_rxnj,
            xsjbxx_sznj,
            xsjbxx_bjbh,
            xsjbxx_bjxx
        FROM `{table_name}`
        """.format(table_name=TABLE_MAPPING["student"])

        with connection.cursor() as cursor:
            cursor.execute(select_sql)
            rows = cursor.fetchall()
            print(f"从{TABLE_MAPPING['student']}取回 {len(rows)} 条记录")
            return rows

    def _insert_into_target(self, connection, rows: List[Dict[str, Any]]) -> Tuple[int, List[Dict[str, Any]]]:
        """
        将数据写入xg_student_msg表
        """
        insert_sql = """
        INSERT INTO `xg_student_msg` (
            `id`, `tenant_id`, `create_dept`, `create_by`, `create_time`,
            `update_by`, `update_time`, `deleted`, `student_no`, `id_card`,
            `name`, `gender`, `birth_date`, `student_status`, `enrollment_year`,
            `data_source`, `grade`, `class_id`, `class_name`, `major_id`,
            `major_name`, `college_id`, `college_name`
        ) VALUES (
            %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s,
            %s, %s, %s
        )
        """

        base_datetime = datetime.now()
        timestamp_str = base_datetime.strftime("%Y%m%d%H%M%S")
        success_count = 0
        sequence = 1
        batch_inserted = 0
        user_inserted_total = 0
        user_skipped_total = 0
        role_inserted_total = 0

        existing_student_nos = self._load_existing_student_nos(connection)

        with connection.cursor() as cursor:
            for row in rows:
                student_no = row.get("xsjbxx_xh")
                name = row.get("xsjbxx_xm")
                if not student_no or not name:
                    print(f"跳过无效数据: {row}")
                    continue

                if student_no in existing_student_nos:
                    print(f"跳过已存在学生: student_no={student_no}")
                    continue

                if sequence > 99999:
                    print("同步条目超过单次支持的最大值99999，停止写入以避免ID溢出")
                    break

                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_xsdqzt"))

                class_id, class_name = self._find_class_dept(
                    connection,
                    row.get("xsjbxx_bjbh"),
                    row.get("xsjbxx_bjxx")
                )

                major_id, major_name, college_id, college_name = self._find_major_info(connection, class_id)

                birth_date = self._normalize_birth_date(row.get("xsjbxx_csrq"))

                enrollment_year = self._normalize_enrollment_year(row.get("xsjbxx_rxnj"))

                params = (
                    unique_id,
                    self.DEFAULT_TENANT_ID,
                    0,  # create_dept
                    self.DEFAULT_CREATE_BY,
                    base_datetime,
                    self.DEFAULT_UPDATE_BY,
                    base_datetime,
                    0,
                    student_no,
                    row.get("xsjbxx_sfzjh"),
                    name,
                    gender,
                    birth_date,
                    status,
                    enrollment_year,
                    "sync",               # data_source
                    row.get("xsjbxx_sznj"),
                    class_id,
                    class_name,
                    major_id,
                    major_name,
                    college_id,
                    college_name
                )

                try:
                    cursor.execute(insert_sql, params)
                    success_count += 1
                    sequence += 1
                    batch_inserted += 1
                    existing_student_nos.add(student_no)
                    student_profile = {
                        "user_id": int(unique_id),
                        "student_no": student_no,
                        "name": name,
                        "gender": gender,
                        "phone": self._normalize_phone(row.get("xsjbxx_lxdh")),
                        "email": None,
                        "class_id": int(class_id) if class_id is not None else None,
                        "tenant_id": self.DEFAULT_TENANT_ID,
                        "create_time": base_datetime,
                        "update_time": base_datetime,
                    }

                    inserted_flag, skipped_flag = self._create_sys_user_for_student(connection, student_profile)
                    user_inserted_total += inserted_flag
                    user_skipped_total += skipped_flag
                    if inserted_flag:
                        role_inserted_total += self._ensure_user_role(connection, student_profile["user_id"], self.STUDENT_ROLE_ID)
                    else:
                        # 即使用户已存在，仍尝试补齐角色
                        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
                except Exception as exc:
                    print(f"插入xg_student_msg失败: {exc}, 数据: {params}")

        if batch_inserted > 0:
            connection.commit()
            print(f"✓ xg_student_msg分批提交剩余 {batch_inserted} 条记录")

        return success_count, user_inserted_total, user_skipped_total, role_inserted_total

    def _load_existing_student_nos(self, connection) -> set:
        """
        获取xg_student_msg表中已存在的学号集合
        """
        query = "SELECT student_no FROM `xg_student_msg`"
        with connection.cursor() as cursor:
            cursor.execute(query)
            rows = cursor.fetchall()
            return {row[0] if isinstance(row, tuple) else row.get("student_no") for row in rows}

    def _generate_student_id(self, timestamp_str: str, sequence: int) -> str:
        """
        生成符合要求的学生ID
        """
        return f"{timestamp_str}{sequence:05d}"

    def _map_gender(self, gender_text: Optional[str]) -> Optional[int]:
        """
        将性别文本映射为目标表字段
        """
        if gender_text == "男":
            return 0
        if gender_text == "女":
            return 1
        return 0

    def _map_student_status(self, status_text: Optional[str]) -> Optional[int]:
        """
        将学生当前状态（xsjbxx_xsdqzt）文本映射为目标表字段
        """
        if not status_text:
            return 1

        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,
        connection,
        class_code: Optional[str],
        class_name: Optional[str]
    ) -> Tuple[Optional[int], Optional[str]]:
        """
        根据班级信息查找sys_dept中的班级部门
        """
        if not class_code or not class_name:
            return None, None

        dept_category = f"bj-{class_code}"
        query = """
        SELECT dept_id, dept_name
        FROM `sys_dept`
        WHERE dept_category = %s AND dept_name = %s
        LIMIT 1
        """

        with connection.cursor() as cursor:
            cursor.execute(query, (dept_category, class_name))
            result = cursor.fetchone()
            if result:
                return result["dept_id"], result["dept_name"]

        print(f"警告: 未在sys_dept中找到班级 -> dept_category={dept_category}, dept_name={class_name}")
        return None, None

    def _find_major_info(
        self,
        connection,
        class_id: Optional[int]
    ) -> Tuple[Optional[int], Optional[str], Optional[int], Optional[str]]:
        """
        根据班级ID从xg_class_msg中查询专业与学院信息
        """
        if not class_id:
            return None, None, None, None

        query = """
        SELECT major_id, major_name, college_id, college_name
        FROM `xg_class_msg`
        WHERE id = %s
        LIMIT 1
        """

        with connection.cursor() as cursor:
            cursor.execute(query, (class_id,))
            result = cursor.fetchone()
            if result:
                return (
                    result.get("major_id"),
                    result.get("major_name"),
                    result.get("college_id"),
                    result.get("college_name")
                )

        print(f"警告: 未在xg_class_msg中找到班级信息 -> class_id={class_id}")
        return None, None, None, None

    def _normalize_birth_date(self, raw_value: Optional[str]) -> Optional[str]:
        """
        将出生日期转换为YYYY-MM-DD格式, 无效值返回None
        """
        if not raw_value:
            return None

        text = str(raw_value).strip()
        if not text:
            return None

        for fmt in ("%Y-%m-%d", "%Y%m%d", "%Y/%m/%d"):
            try:
                parsed = datetime.strptime(text, fmt)
                return parsed.strftime("%Y-%m-%d")
            except ValueError:
                continue

        print(f"警告: 无法解析出生日期 -> raw='{raw_value}'")
        return None

    def _normalize_enrollment_year(self, raw_value: Optional[Any]) -> Optional[int]:
        """
        清洗入学年份字段，返回有效的年份整数或None
        """
        if raw_value is None:
            return None

        text = str(raw_value).strip()
        if not text:
            return None

        if text.isdigit():
            return int(text)

        # 处理可能的日期格式，如2025-09-01
        try:
            parsed = datetime.strptime(text[:10], "%Y-%m-%d")
            return parsed.year
        except Exception:
            print(f"警告: 无法解析入学年份 -> raw='{raw_value}'")
            return None

    def _normalize_phone(self, raw_value: Optional[str]) -> Optional[str]:
        """
        清洗手机号字段
        """
        if not raw_value:
            return None

        text = str(raw_value).strip()
        return text if text else None

    def _create_sys_user_for_student(self, connection, student: Dict[str, Any]) -> Tuple[int, int]:
        """
        为学生创建sys_user账号

        Returns:
            (新增账号数量, 已存在账号数量)
        """
        user_id = student["user_id"]
        with connection.cursor() as cursor:
            cursor.execute("SELECT 1 FROM `sys_user` WHERE user_id = %s LIMIT 1", (user_id,))
            if cursor.fetchone():
                return 0, 1

        password_hash = bcrypt.hashpw(self.DEFAULT_PASSWORD.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
        insert_sql = """
        INSERT INTO `sys_user` (
            `user_id`, `dept_id`, `user_name`, `nick_name`, `user_type`,
            `email`, `phonenumber`, `sex`, `avatar`, `password`, `status`,
            `del_flag`, `login_ip`, `login_date`, `create_by`, `create_time`,
            `update_by`, `update_time`, `remark`, `tenant_id`, `account_type`
        ) VALUES (
            %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s
        )
        """
        try:
            with connection.cursor() as cursor:
                cursor.execute(
                    insert_sql,
                    (
                        user_id,
                        student.get("class_id"),
                        student["student_no"],
                        student["name"],
                        self.DEFAULT_USER_TYPE,
                        student.get("email"),
                        student.get("phone"),
                        self._map_gender_for_user(student.get("gender")),
                        None,
                        password_hash,
                        self.SYS_USER_STATUS,
                        self.SYS_USER_DEL_FLAG,
                        None,
                        None,
                        self.DEFAULT_CREATE_BY,
                        student["create_time"],
                        self.DEFAULT_UPDATE_BY,
                        student["update_time"],
                        self.SYS_USER_REMARK,
                        student["tenant_id"],
                        self.DEFAULT_ACCOUNT_TYPE,
                    ),
                )
            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:
        """
        确保用户具备指定角色

        Returns:
            新增的角色绑定数量
        """
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT 1 FROM `sys_user_role` WHERE user_id = %s AND role_id = %s LIMIT 1",
                (user_id, role_id),
            )
            if cursor.fetchone():
                return 0
            try:
                cursor.execute(
                    """
                    INSERT INTO `sys_user_role` (`user_id`, `role_id`)
                    VALUES (%s, %s)
                    """,
                    (user_id, role_id),
                )
                return 1
            except Exception as exc:
                print(f"绑定角色失败: {exc}, user_id={user_id}, role_id={role_id}")
                return 0

    def _sync_sys_users(self, connection, students: List[Dict[str, Any]]) -> Tuple[int, int]:
        """
        将新增学生同步到sys_user表
        Returns:
            (新增账号数量, 已存在账号数量)
        """
        if not students:
            return 0, 0

        user_ids = [student["user_id"] for student in students]
        existing_user_ids = self._load_existing_sys_user_ids(connection, user_ids)

        insert_sql = """
        INSERT INTO `sys_user` (
            `user_id`, `dept_id`, `user_name`, `nick_name`, `user_type`,
            `email`, `phonenumber`, `sex`, `avatar`, `password`, `status`,
            `del_flag`, `login_ip`, `login_date`, `create_by`, `create_time`,
            `update_by`, `update_time`, `remark`, `tenant_id`, `account_type`
        ) VALUES (
            %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s,
            %s, %s, %s, %s, %s
        )
        """

        inserted = 0
        skipped = 0

        with connection.cursor() as cursor:
            for student in students:
                user_id = student["user_id"]
                if user_id in existing_user_ids:
                    skipped += 1
                    continue

                password_hash = bcrypt.hashpw(self.DEFAULT_PASSWORD.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
                try:
                    cursor.execute(
                        insert_sql,
                        (
                            user_id,
                            student.get("class_id"),
                            student["student_no"],
                            student["name"],
                            self.DEFAULT_USER_TYPE,
                            student.get("email"),
                            student.get("phone"),
                            self._map_gender_for_user(student.get("gender")),
                            None,
                            password_hash,
                            self.SYS_USER_STATUS,
                            self.SYS_USER_DEL_FLAG,
                            None,
                            None,
                            self.DEFAULT_CREATE_BY,
                            student["create_time"],
                            self.DEFAULT_UPDATE_BY,
                            student["update_time"],
                            self.SYS_USER_REMARK,
                            student["tenant_id"],
                            self.DEFAULT_ACCOUNT_TYPE,
                        ),
                    )
                    inserted += 1
                    existing_user_ids.add(user_id)
                except Exception as exc:
                    skipped += 1
                    print(f"插入sys_user失败: {exc}, user_id={user_id}, student_no={student['student_no']}")

        return inserted, skipped

    def _sync_user_roles(self, connection, user_ids: List[int], role_id: int) -> int:
        """
        为用户绑定指定角色
        Returns:
            新增的角色绑定数量
        """
        if not user_ids:
            return 0

        existing = self._load_existing_user_role_ids(connection, user_ids, role_id)
        insert_sql = """
        INSERT INTO `sys_user_role` (
            `user_id`, `role_id`
        ) VALUES (
            %s, %s
        )
        """

        inserted = 0
        with connection.cursor() as cursor:
            for user_id in user_ids:
                if user_id in existing:
                    continue
                try:
                    cursor.execute(insert_sql, (user_id, role_id))
                    inserted += 1
                except Exception as exc:
                    print(f"绑定角色失败: {exc}, user_id={user_id}, role_id={role_id}")
        return inserted

    def _load_existing_sys_user_ids(self, connection, user_ids: List[int]) -> set:
        """
        批量查询已存在的sys_user用户ID
        """
        if not user_ids:
            return set()

        existing_ids = set()
        with connection.cursor() as cursor:
            for batch in self._split_batches(user_ids):
                placeholders = ",".join(["%s"] * len(batch))
                sql = f"SELECT user_id FROM `sys_user` WHERE user_id IN ({placeholders})"
                cursor.execute(sql, batch)
                rows = cursor.fetchall()
                for row in rows:
                    value = row["user_id"] if isinstance(row, dict) else row[0]
                    existing_ids.add(int(value))
        return existing_ids

    def _load_existing_user_role_ids(self, connection, user_ids: List[int], role_id: int) -> set:
        """
        查询已经存在的用户角色关联
        """
        if not user_ids:
            return set()

        existing = set()
        with connection.cursor() as cursor:
            for batch in self._split_batches(user_ids):
                placeholders = ",".join(["%s"] * len(batch))
                sql = f"""
                SELECT user_id FROM `sys_user_role`
                WHERE role_id = %s AND user_id IN ({placeholders})
                """
                cursor.execute(sql, [role_id, *batch])
                rows = cursor.fetchall()
                for row in rows:
                    value = row["user_id"] if isinstance(row, dict) else row[0]
                    existing.add(int(value))
        return existing

    def _split_batches(self, values: List[int], size: int = QUERY_BATCH_SIZE) -> List[List[int]]:
        """
        将列表按指定大小拆分
        """
        if not values:
            return []
        return [values[i : i + size] for i in range(0, len(values), size)]

    def _map_gender_for_user(self, gender_code: Optional[int]) -> str:
        """
        将性别编码转换为sys_user表所需的字符串
        """
        if gender_code == 0:
            return "0"
        if gender_code == 1:
            return "1"
        return "2"


def main():
    """主函数 - 示例用法"""
    # API配置信息
    APP_ID = "acd4cc28-e8c1-4fb8-a38e-6de8cbaaee34"
    APP_KEY = "dyjs_xsjbxx"
    FID = "19628"  # 单位ID
    
    # 创建API客户端
    api = ChaoxingStudentAPI(app_id=APP_ID, app_key=APP_KEY, fid=FID)
    
    # 获取所有数据并分批保存到数据库
    print("=" * 60)
    print("获取所有学生信息并分批保存到数据库")
    print("=" * 60)
    
    # 步骤1: 清空表
    print("\n步骤1: 清空学生数据表...")
    api.clear_table()
    
    # 步骤2: 获取数据并分批保存到cx_xssj表
    print("\n步骤2: 获取学生信息并分批保存到cx_xssj表（每批100条）...")
    saved_count = api.get_and_save_students_batch(max_page_size=500, batch_size=100)
    
    if saved_count > 0:
        print(f"\n✓ 完成！成功保存 {saved_count} 条数据到cx_xssj表")

        # 步骤3: 同步数据至xg_student_msg
        print("\n步骤3: 同步数据至xg_student_msg...")
        synced_count = api.sync_student_msg()
        print(f"\n✓ 学生信息同步完成，成功写入 {synced_count} 条记录")
    else:
        print("\n✗ 未获取到数据或保存失败")


if __name__ == "__main__":
    main()