#!/usr/bin/env python # -*- coding: utf-8 -*- """ 超星系统 - 学生学籍异动信息API请求脚本 API路径: xjydsjpi_48018/list """ import hashlib import base64 import json import requests import pymysql import time 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 ChaoxingStudentChangeAPI: """超星学生学籍异动信息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 = "xjydsjpi_48018" # 生成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=xjydsjpi_48018#$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_change_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 } # 构建请求头（appId放在请求头中） 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_student_changes(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_change_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["student_change"] 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["student_change"] with connection.cursor() as cursor: # 构建插入SQL insert_sql = f""" INSERT INTO `{table_name}` ( `inserttime`, `updatetime`, `xjyd_yjsqsj`, `xjyd_ydsxxq`, `xjydsp_ydlx`, `xjydsp_sqrxh`, `xjydsp_xm`, `xjydsp_ydqssxq`, `xjydsp_ydqsznj`, `xjydsp_ydqszyx`, `xjydsp_ydqszzy`, `xjydsp_ydqszbj`, `xjydsp_ydqxz`, `xjydsp_ydhszbj`, `xjydsp_ydhsznj`, `xjydsp_ydhszyx`, `xjydsp_ydhszzy`, `xjydsp_ydhssxq`, `xjydsp_ydhxz`, `xjydsp_ydhbzrgh`, `xjyd_name`, `xjydsp_sqyy`, `xjydsp_ydhbzr`, `xjydsp_sxqm`, `xjydsp_zmfj`, `xjyd_shzt`, `xjydsp_ydhyxbh`, `xjydsp_ydhzybh`, `xjydsp_ydhbjbh` ) 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 ) """ success_count = 0 for item in data_list: try: cursor.execute(insert_sql, ( current_time, current_time, # inserttime, updatetime item.get('xjyd_yjsqsj'), item.get('xjyd_ydsxxq'), item.get('xjydsp_ydlx'), item.get('xjydsp_sqrxh'), item.get('xjydsp_xm'), item.get('xjydsp_ydqssxq'), item.get('xjydsp_ydqsznj'), item.get('xjydsp_ydqszyx'), item.get('xjydsp_ydqszzy'), item.get('xjydsp_ydqszbj'), item.get('xjydsp_ydqxz'), item.get('xjydsp_ydhszbj'), item.get('xjydsp_ydhsznj'), item.get('xjydsp_ydhszyx'), item.get('xjydsp_ydhszzy'), item.get('xjydsp_ydhssxq'), item.get('xjydsp_ydhxz'), item.get('xjydsp_ydhbzrgh'), item.get('xjyd_name'), item.get('xjydsp_sqyy'), item.get('xjydsp_ydhbzr'), item.get('xjydsp_sxqm'), item.get('xjydsp_zmfj'), item.get('xjyd_shzt'), item.get('xjydsp_ydhyxbh'), item.get('xjydsp_ydhzybh'), item.get('xjydsp_ydhbjbh') )) 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 def main(): """主函数 - 示例用法""" # API配置信息 APP_ID = "4083b823-cc45-4b4b-8eb4-a789f791b23c" APP_KEY = "dyjs_xjyd" FID = "19628" # 单位ID # 创建API客户端 api = ChaoxingStudentChangeAPI(app_id=APP_ID, app_key=APP_KEY, fid=FID) # 获取所有数据并保存到数据库 print("=" * 60) print("获取所有学籍异动信息并保存到数据库") print("=" * 60) # 步骤1: 清空表 print("\n步骤1: 清空学籍异动数据表...") api.clear_table() # 步骤2: 获取所有数据 print("\n步骤2: 获取所有学籍异动信息（自动分页）...") all_changes = api.get_all_student_changes(max_page_size=500) # 步骤3: 保存到数据库 if all_changes: print(f"\n步骤3: 保存数据到数据库...") saved_count = api.save_to_database(all_changes) print(f"\n✓ 完成！共获取 {len(all_changes)} 条数据，成功保存 {saved_count} 条") print(f"\n前3条数据示例:") for i, change in enumerate(all_changes[:3], 1): print(f"{i}. {change}") else: print("\n✗ 未获取到数据") if __name__ == "__main__": main()