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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
超星系统 - 系部信息完整同步脚本
功能：
1. 从超星API获取系部信息（API路径: xbsjpi_64522/list）
2. 保存到cx_xbsj表
3. 从cx_xbsj表同步到sys_dept表（学工系统部门表）
同步规则：
- 筛选条件：yxsj_sfkkyx='是' AND yxsj_sfskyx='是' AND yxsj_sfqy='是'
- 排除部门：教务处、基础部、外聘（不同步到sys_dept表）
- 判断是否已存在：dept_category='xy-' + yxsj_yxdm AND dept_name=yxsj_yxmc
- 插入到 parent_id=101（二级学院）下
"""

import hashlib
import base64
import json
import requests
import pymysql
import time
from datetime import datetime
from typing import Dict, Any, Optional, List
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
from db_config import DB_CONFIG, TABLE_MAPPING


class ChaoxingAPI:
    """超星API请求类"""
    
    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 = "xbsjpi_64522"
        # 生成AES解密密钥：appId去除'-'后取前16位
        self.aes_key = app_id.replace("-", "")[:16].encode('utf-8')
    
    def generate_enc(self, page_num: int, page_size: int) -> str:
        """
        生成enc加密验证串
        
        enc加密方式：32位大写的MD5字符串
        格式：[apiPath=xbsjpi_64522#$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_department_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_departments(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_department_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 clear_table(self) -> bool:
        """
        清空系部数据表
        
        Returns:
            是否成功
        """
        try:
            connection = pymysql.connect(**DB_CONFIG)
            try:
                with connection.cursor() as cursor:
                    table_name = TABLE_MAPPING["department"]
                    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]) -> int:
        """
        将系部数据保存到数据库
        
        Args:
            data_list: 系部数据列表
            
        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["department"]
                
                with connection.cursor() as cursor:
                    # 构建插入SQL
                    insert_sql = f"""
                    INSERT INTO `{table_name}` (
                        `inserttime`, `updatetime`,
                        `yxsj_yxmc`, `yxsj_yxdm`, `yxsj_sfkkyx`, `yxsj_sfskyx`,
                        `yxsj_sfqy`, `yxsj_yxywmc`, `yxsj_yxjc`, `yxsj_gdjxl`,
                        `yxsj_yxfzr`, `yxsj_kyjxlsx`, `dataMigrateUniqueId`
                    ) VALUES (
                        %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
                    )
                    """
                    
                    success_count = 0
                    for item in data_list:
                        try:
                            cursor.execute(insert_sql, (
                                current_time, current_time,  # inserttime, updatetime
                                item.get('yxsj_yxmc'), item.get('yxsj_yxdm'), item.get('yxsj_sfkkyx'), item.get('yxsj_sfskyx'),
                                item.get('yxsj_sfqy'), item.get('yxsj_yxywmc'), item.get('yxsj_yxjc'), item.get('yxsj_gdjxl'),
                                item.get('yxsj_yxfzr'), item.get('yxsj_kyjxlsx'), item.get('dataMigrateUniqueId')
                            ))
                            success_count += 1
                        except Exception as e:
                            print(f"插入单条数据失败: {e}, 数据: {item}")
                    
                    connection.commit()
                    print(f"✓ 成功插入 {success_count}/{len(data_list)} 条数据到表 {table_name}")
                    return success_count
            finally:
                connection.close()
        except Exception as e:
            print(f"✗ 保存数据到数据库失败: {e}")
            return 0


class DepartmentSync:
    """系部数据同步类 - 从cx_xbsj表同步到sys_dept表"""
    
    def __init__(self):
        """初始化"""
        self.parent_dept_id = 101  # 二级学院的dept_id
        self.parent_ancestors = '0,100,101'  # 父节点的ancestors
        self.tenant_id = '000000'
        # 排除的部门名称列表（不同步到sys_dept表）
        self.excluded_dept_names = ['教务处', '外聘']
    
    def get_valid_departments(self) -> List[Dict[str, Any]]:
        """
        从cx_xbsj表中获取符合条件的数据
        
        Returns:
            符合条件的系部数据列表
        """
        try:
            connection = pymysql.connect(**DB_CONFIG)
            try:
                with connection.cursor(pymysql.cursors.DictCursor) as cursor:
                    # 查询符合条件的数据：三个字段都等于"是"
                    sql = """
                    SELECT 
                        yxsj_yxmc, yxsj_yxdm, yxsj_sfkkyx, yxsj_sfskyx, yxsj_sfqy,
                        yxsj_yxywmc, yxsj_yxjc, yxsj_gdjxl, yxsj_yxfzr, yxsj_kyjxlsx
                    FROM cx_xbsj
                    WHERE yxsj_sfkkyx = '是' 
                      AND yxsj_sfskyx = '是' 
                      AND yxsj_sfqy = '是'
                    ORDER BY inserttime ASC
                    """
                    cursor.execute(sql)
                    results = cursor.fetchall()
                    print(f"✓ 从cx_xbsj表中查询到 {len(results)} 条符合条件的数据")
                    return results
            finally:
                connection.close()
        except Exception as e:
            print(f"✗ 查询cx_xbsj表失败: {e}")
            return []
    
    def check_department_exists(self, dept_name: str, dept_category: str) -> bool:
        """
        检查部门是否已存在
        
        Args:
            dept_name: 部门名称
            dept_category: 部门类别编码
            
        Returns:
            是否存在
        """
        try:
            connection = pymysql.connect(**DB_CONFIG)
            try:
                with connection.cursor() as cursor:
                    sql = """
                    SELECT COUNT(*) as count
                    FROM sys_dept
                    WHERE dept_name = %s AND dept_category = %s
                    """
                    cursor.execute(sql, (dept_name, dept_category))
                    result = cursor.fetchone()
                    return result[0] > 0
            finally:
                connection.close()
        except Exception as e:
            print(f"✗ 检查部门是否存在失败: {e}")
            return False
    
    def get_max_order_num(self, parent_id: int) -> int:
        """
        获取指定父节点下的最大order_num
        
        Args:
            parent_id: 父部门ID
            
        Returns:
            最大order_num，如果没有数据则返回0
        """
        try:
            connection = pymysql.connect(**DB_CONFIG)
            try:
                with connection.cursor() as cursor:
                    sql = """
                    SELECT MAX(order_num) as max_order
                    FROM sys_dept
                    WHERE parent_id = %s
                    """
                    cursor.execute(sql, (parent_id,))
                    result = cursor.fetchone()
                    return result[0] if result[0] is not None else 0
            finally:
                connection.close()
        except Exception as e:
            print(f"✗ 获取最大order_num失败: {e}")
            return 0
    
    def generate_dept_id(self, base_timestamp: str, index: int) -> int:
        """
        生成dept_id
        格式：当前时间戳（%Y%m%d%H%M%S）+ 0001开始递增
        
        Args:
            base_timestamp: 基础时间戳字符串（格式：%Y%m%d%H%M%S）
            index: 序号（从1开始）
            
        Returns:
            dept_id
        """
        # 序号格式化为4位数字（0001, 0002, ...）
        sequence = str(index).zfill(4)
        # 拼接：时间戳 + 序号
        dept_id_str = base_timestamp + sequence
        return int(dept_id_str)
    
    def sync_departments(self) -> Dict[str, Any]:
        """
        同步系部数据到sys_dept表
        
        Returns:
            同步结果统计（包含成功、跳过、失败的系部代码列表）
        """
        # 获取符合条件的数据
        valid_departments = self.get_valid_departments()
        if not valid_departments:
            print("没有符合条件的数据需要同步")
            return {
                "total": 0,
                "inserted": 0,
                "skipped": 0,
                "failed": 0,
                "inserted_yxdm": [],
                "skipped_yxdm": [],
                "failed_yxdm": []
            }
        
        # 生成基础时间戳（用于生成dept_id）
        base_timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
        
        # 获取当前最大order_num
        max_order_num = self.get_max_order_num(self.parent_dept_id)
        current_order_num = max_order_num
        
        # 统计信息
        stats = {
            "total": len(valid_departments),
            "inserted": 0,
            "skipped": 0,
            "failed": 0,
            "inserted_yxdm": [],  # 成功插入的系部代码列表
            "skipped_yxdm": [],   # 跳过的系部代码列表
            "failed_yxdm": []     # 失败的系部代码列表
        }
        
        # 当前时间
        current_time = datetime.now()
        
        try:
            connection = pymysql.connect(**DB_CONFIG)
            try:
                with connection.cursor() as cursor:
                    # 准备插入SQL
                    insert_sql = """
                    INSERT INTO sys_dept (
                        dept_id, tenant_id, parent_id, ancestors, dept_name, dept_category,
                        order_num, leader, phone, email, status, del_flag,
                        create_dept, create_by, create_time, update_by, update_time
                    ) VALUES (
                        %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
                    )
                    """
                    
                    # 用于生成dept_id的序号计数器（只对实际插入的数据计数）
                    insert_index = 0
                    
                    # 遍历数据，逐个处理
                    for dept in valid_departments:
                        dept_name = dept.get('yxsj_yxmc')
                        yxsj_yxdm = dept.get('yxsj_yxdm')
                        
                        # 验证数据
                        if not dept_name:
                            print(f"⚠ 跳过：部门名称为空 (yxsj_yxdm: {yxsj_yxdm})")
                            stats["skipped"] += 1
                            if yxsj_yxdm:
                                stats["skipped_yxdm"].append(yxsj_yxdm)
                            continue
                        
                        if not yxsj_yxdm:
                            print(f"⚠ 跳过：系部代码为空 (dept_name: {dept_name})")
                            stats["skipped"] += 1
                            # 系部代码为空，无法记录
                            continue
                        
                        # 检查是否在排除列表中
                        if dept_name in self.excluded_dept_names:
                            print(f"⏭  跳过排除的部门: {dept_name}")
                            stats["skipped"] += 1
                            stats["skipped_yxdm"].append(yxsj_yxdm)
                            continue
                        
                        # 检查dept_name长度（sys_dept表中dept_name是varchar(30)）
                        if len(dept_name) > 30:
                            print(f"⚠ 警告：部门名称超过30字符，将被截断: {dept_name} (长度: {len(dept_name)})")
                            dept_name = dept_name[:30]
                        
                        # 构建dept_category
                        dept_category = f"xy-{yxsj_yxdm}"
                        
                        # 检查是否已存在（使用处理后的dept_name）
                        if self.check_department_exists(dept_name, dept_category):
                            print(f"⏭  跳过已存在的部门: {dept_name} (dept_category: {dept_category})")
                            stats["skipped"] += 1
                            stats["skipped_yxdm"].append(yxsj_yxdm)
                            continue
                        
                        # 递增插入序号（只对实际插入的数据计数）
                        insert_index += 1
                        
                        # 生成dept_id
                        dept_id = self.generate_dept_id(base_timestamp, insert_index)
                        
                        # 递增order_num
                        current_order_num += 1
                        
                        # 准备插入数据
                        try:
                            cursor.execute(insert_sql, (
                                dept_id,                           # dept_id
                                self.tenant_id,                    # tenant_id
                                self.parent_dept_id,               # parent_id
                                self.parent_ancestors,             # ancestors
                                dept_name,                         # dept_name
                                dept_category,                     # dept_category
                                current_order_num,                 # order_num
                                None,                              # leader
                                None,                              # phone
                                None,                              # email
                                '0',                               # status
                                '0',                               # del_flag
                                0,                                 # create_dept
                                1,                                 # create_by
                                current_time,                      # create_time
                                1,                                 # update_by
                                current_time                       # update_time
                            ))
                            stats["inserted"] += 1
                            stats["inserted_yxdm"].append(yxsj_yxdm)
                            print(f"✓ 插入部门: {dept_name} (dept_id: {dept_id}, dept_category: {dept_category}, order_num: {current_order_num})")
                        except Exception as e:
                            stats["failed"] += 1
                            stats["failed_yxdm"].append(yxsj_yxdm)
                            print(f"✗ 插入部门失败: {dept_name}, 错误: {e}")
                    
                    # 提交事务
                    connection.commit()
                    print(f"\n✓ 事务提交成功")
                    
            finally:
                connection.close()
        except Exception as e:
            print(f"✗ 同步过程发生错误: {e}")
            stats["failed"] = stats["total"] - stats["inserted"] - stats["skipped"]
        
        return stats
    
    def print_statistics(self, stats: Dict[str, Any]):
        """
        打印统计信息
        
        Args:
            stats: 统计信息字典（包含系部代码列表）
        """
        print("\n" + "=" * 60)
        print("同步统计信息")
        print("=" * 60)
        print(f"总数据量: {stats['total']}")
        print(f"成功插入: {stats['inserted']}")
        print(f"跳过（已存在）: {stats['skipped']}")
        print(f"失败: {stats['failed']}")
        print("=" * 60)
        
        # 输出系部代码列表（逗号分隔）
        if stats.get('inserted_yxdm'):
            inserted_yxdm_str = ','.join(stats['inserted_yxdm'])
            print(f"\n成功插入的系部代码: {inserted_yxdm_str}")
        
        if stats.get('skipped_yxdm'):
            skipped_yxdm_str = ','.join(stats['skipped_yxdm'])
            print(f"跳过的系部代码: {skipped_yxdm_str}")
        
        if stats.get('failed_yxdm'):
            failed_yxdm_str = ','.join(stats['failed_yxdm'])
            print(f"失败的系部代码: {failed_yxdm_str}")
        
        print("=" * 60)


def main():
    """主函数 - 完整同步流程"""
    # API配置信息
    APP_ID = "eb68f9d4-b833-4a09-aae8-28e970de1461"
    APP_KEY = "dyjs_xbxx"
    FID = "19628"  # 单位ID
    
    print("=" * 60)
    print("超星系部信息完整同步流程")
    print("=" * 60)
    print("流程：超星API -> cx_xbsj表 -> sys_dept表")
    print("=" * 60)
    
    # ========== 第一阶段：从超星API获取数据并保存到cx_xbsj表 ==========
    print("\n" + "=" * 60)
    print("第一阶段：从超星API获取系部信息")
    print("=" * 60)
    
    # 创建API客户端
    api = ChaoxingAPI(app_id=APP_ID, app_key=APP_KEY, fid=FID)
    
    # 步骤1: 清空表
    print("\n步骤1: 清空系部数据表（cx_xbsj）...")
    api.clear_table()
    
    # 步骤2: 获取所有数据
    print("\n步骤2: 获取所有系部信息（自动分页）...")
    all_departments = api.get_all_departments(max_page_size=500)
    
    # 步骤3: 保存到数据库
    if all_departments:
        print(f"\n步骤3: 保存数据到cx_xbsj表...")
        saved_count = api.save_to_database(all_departments)
        print(f"\n✓ 第一阶段完成！共获取 {len(all_departments)} 条数据，成功保存 {saved_count} 条到cx_xbsj表")
    else:
        print("\n✗ 未获取到数据，终止流程")
        return
    
    # ========== 第二阶段：从cx_xbsj表同步到sys_dept表 ==========
    print("\n" + "=" * 60)
    print("第二阶段：从cx_xbsj表同步到sys_dept表")
    print("=" * 60)
    print(f"目标父节点: dept_id=101 (二级学院)")
    print(f"筛选条件: yxsj_sfkkyx='是' AND yxsj_sfskyx='是' AND yxsj_sfqy='是'")
    print(f"排除部门: 教务处、基础部、外聘")
    print("=" * 60)
    
    # 创建同步实例
    sync = DepartmentSync()
    
    # 执行同步
    stats = sync.sync_departments()
    
    # 打印统计信息
    sync.print_statistics(stats)
    
    # ========== 总结 ==========
    print("\n" + "=" * 60)
    print("完整同步流程总结")
    print("=" * 60)
    print(f"✓ 从超星API获取: {len(all_departments)} 条数据")
    print(f"✓ 保存到cx_xbsj表: {saved_count} 条数据")
    print(f"✓ 同步到sys_dept表: 插入 {stats['inserted']} 条，跳过 {stats['skipped']} 条，失败 {stats['failed']} 条")
    print("=" * 60)
    
    if stats["inserted"] > 0:
        print("\n✓ 完整同步流程执行成功！")
    elif stats["skipped"] == stats["total"]:
        print("\n✓ 所有数据已存在，无需同步")
    else:
        print("\n⚠ 同步过程中出现问题，请检查日志")


if __name__ == "__main__":
    main()