1120 lines
44 KiB
Python
1120 lines
44 KiB
Python
#!/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 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')
|
|
|
|
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 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:
|
|
connection = pymysql.connect(**DB_CONFIG)
|
|
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
|
|
)
|
|
"""
|
|
|
|
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)} 条数据")
|
|
|
|
# 逐条插入数据
|
|
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_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()
|
|
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()
|
|
except Exception as e:
|
|
print(f"✗ 获取并保存数据失败: {e}")
|
|
return 0
|
|
|
|
# --------- 新增的方法：同步到xg_student_msg ---------
|
|
|
|
def sync_student_msg(self) -> int:
|
|
"""
|
|
将cx_xssj表中的学生数据同步到xg_student_msg表
|
|
|
|
Returns:
|
|
成功同步的记录数
|
|
"""
|
|
try:
|
|
connection = pymysql.connect(**DB_CONFIG, cursorclass=pymysql.cursors.DictCursor)
|
|
try:
|
|
source_rows = self._fetch_source_students(connection)
|
|
if not source_rows:
|
|
print("cx_xssj表暂无可同步数据")
|
|
return 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()
|
|
except Exception as exc:
|
|
print(f"✗ 同步到xg_student_msg失败: {exc}")
|
|
return 0
|
|
|
|
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_xjzt"))
|
|
|
|
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:
|
|
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]:
|
|
"""
|
|
将学籍状态文本映射为目标表字段
|
|
"""
|
|
if status_text == "在籍":
|
|
return 1
|
|
if status_text == "不在籍":
|
|
return 6
|
|
return None
|
|
|
|
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']}")
|
|
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()
|
|
|