368 lines
13 KiB
Python
368 lines
13 KiB
Python
#!/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()
|
|
|