1248 lines
54 KiB
Python
1248 lines
54 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
超星系统 - 班级信息完整同步脚本
|
||
功能：
|
||
1. 从超星API获取班级信息（API路径: bjsjpi_53146/list）
|
||
2. 保存到cx_bjsj表
|
||
3. 从cx_bjsj表同步到sys_dept表（学工系统部门表）
|
||
4. 从cx_bjsj表同步到xg_class_msg表（学工系统班级信息表）
|
||
|
||
同步规则（从cx_bjsj表同步到sys_dept表）：
|
||
- 筛选条件：bjxx_sfqy='是' 的数据才可以同步
|
||
- 判断是否已存在：dept_category='bj-' + bjxx_bjbh AND dept_name=bjxx_bjmc
|
||
- 父部门查找：通过 dept_category='zy-' + bjxx_zybh AND dept_name=bjxx_zy 查找父部门ID
|
||
- dept_id：当前时间戳（%Y%m%d%H%M%S）+ 0001开始递增
|
||
- parent_id：通过专业信息查找得到的dept_id
|
||
- ancestors：根据parent_id计算
|
||
- order_num：按照返回数据的顺序，每个父部门下独立递增
|
||
|
||
同步规则（从cx_bjsj表同步到xg_class_msg表）：
|
||
- 筛选条件：与sys_dept表同步相同的筛选条件（bjxx_sfqy='是'）
|
||
- 判断是否已存在：id = sys_dept表的dept_id
|
||
- id：与sys_dept表的dept_id一致
|
||
- major_id和major_name：通过 dept_category='zy-' + bjxx_zybh AND dept_name=bjxx_zy 查找
|
||
- college_id和college_name：通过major_id查找parent_id，然后检查dept_category是否以"xy-"开头
|
||
- school_system：通过bjxx_zybh关联cx_zysj表的zysj_zybh，获取zysj_xzm
|
||
- grade和enrollment_year：使用bjxx_rxnf，如果为空则使用当前年
|
||
- 其他字段：按照需求设置（大部分为空或默认值）
|
||
"""
|
||
|
||
import hashlib
|
||
import base64
|
||
import json
|
||
import requests
|
||
import pymysql
|
||
import time
|
||
import re
|
||
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 ChaoxingClassAPI:
|
||
"""超星班级信息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 = "bjsjpi_53146"
|
||
# 生成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=bjsjpi_53146#$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_class_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_classes(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_class_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["class"]
|
||
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:
|
||
"""
|
||
将班级数据保存到数据库
|
||
只保留近5年的数据（根据bjxx_rxnf字段，即班级入学年份）
|
||
|
||
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["class"]
|
||
|
||
with connection.cursor() as cursor:
|
||
# 构建插入SQL
|
||
insert_sql = f"""
|
||
INSERT INTO `{table_name}` (
|
||
`inserttime`, `updatetime`,
|
||
`bjxx_bjmc`, `bjxx_bjbh`, `bjxx_ssyx`, `bjxx_yxbm`,
|
||
`bjxx_zy`, `bjxx_zybh`, `bjxx_rxnf`, `bjxx_bzr`,
|
||
`bjxx_bzrgh`, `bjxx_szxq`, `bjxx_xqbm`, `bjxx_njrs`,
|
||
`bjxx_bjzdrs`, `bjxx_ssjys`, `bjxx_jysbh`, `bjxx_bynf`,
|
||
`bjxx_jbny`, `bjxx_byxnxq`, `bjxx_byny`, `bjxx_gdjs`,
|
||
`bjxx_gdjsbh`, `bjxx_bjjc`, `bjxx_sfqy`, `bjxx_sznf`,
|
||
`dataMigrateUniqueId`
|
||
) 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
|
||
)
|
||
"""
|
||
|
||
success_count = 0
|
||
|
||
for item in data_list:
|
||
try:
|
||
cursor.execute(insert_sql, (
|
||
current_time, current_time, # inserttime, updatetime
|
||
item.get('bjxx_bjmc'), item.get('bjxx_bjbh'), item.get('bjxx_ssyx'), item.get('bjxx_yxbm'),
|
||
item.get('bjxx_zy'), item.get('bjxx_zybh'), item.get('bjxx_rxnf'), item.get('bjxx_bzr'),
|
||
item.get('bjxx_bzrgh'), item.get('bjxx_szxq'), item.get('bjxx_xqbm'), item.get('bjxx_njrs'),
|
||
item.get('bjxx_bjzdrs'), item.get('bjxx_ssjys'), item.get('bjxx_jysbh'), item.get('bjxx_bynf'),
|
||
item.get('bjxx_jbny'), item.get('bjxx_byxnxq'), item.get('bjxx_byny'), item.get('bjxx_gdjs'),
|
||
item.get('bjxx_gdjsbh'), item.get('bjxx_bjjc'), item.get('bjxx_sfqy'), item.get('bjxx_sznf'),
|
||
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 ClassSync:
|
||
"""班级数据同步类 - 从cx_bjsj表同步到sys_dept表"""
|
||
|
||
def __init__(self):
|
||
"""初始化"""
|
||
self.tenant_id = '000000'
|
||
|
||
def get_valid_classes(self) -> List[Dict[str, Any]]:
|
||
"""
|
||
从cx_bjsj表中获取符合条件的数据（bjxx_sfqy = "是"）
|
||
|
||
Returns:
|
||
符合条件的班级数据列表
|
||
"""
|
||
try:
|
||
connection = pymysql.connect(**DB_CONFIG)
|
||
try:
|
||
with connection.cursor(pymysql.cursors.DictCursor) as cursor:
|
||
# 查询符合条件的数据：bjxx_sfqy = "是"
|
||
sql = """
|
||
SELECT
|
||
bjxx_bjmc, bjxx_bjbh, bjxx_zy, bjxx_zybh, bjxx_rxnf
|
||
FROM cx_bjsj
|
||
WHERE bjxx_sfqy = '是'
|
||
ORDER BY inserttime ASC
|
||
"""
|
||
cursor.execute(sql)
|
||
results = cursor.fetchall()
|
||
print(f"✓ 从cx_bjsj表中查询到 {len(results)} 条符合条件的数据（bjxx_sfqy='是'）")
|
||
return results
|
||
finally:
|
||
connection.close()
|
||
except Exception as e:
|
||
print(f"✗ 查询cx_bjsj表失败: {e}")
|
||
return []
|
||
|
||
def check_class_exists(self, dept_name: str, dept_category: str) -> bool:
|
||
"""
|
||
检查班级是否已存在
|
||
|
||
Args:
|
||
dept_name: 部门名称（班级名称）
|
||
dept_category: 部门类别编码（"bj-" + bjxx_bjbh）
|
||
|
||
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 find_parent_dept_id(self, zy_name: str, zybh: str) -> Optional[int]:
|
||
"""
|
||
查找父部门ID（专业）
|
||
|
||
通过 dept_category = "zy-" + bjxx_zybh 和 dept_name = bjxx_zy 查找
|
||
|
||
Args:
|
||
zy_name: 专业名称
|
||
zybh: 专业编号
|
||
|
||
Returns:
|
||
父部门ID，如果未找到则返回None
|
||
"""
|
||
if not zybh:
|
||
return None
|
||
|
||
try:
|
||
connection = pymysql.connect(**DB_CONFIG)
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
dept_category = f"zy-{zybh}"
|
||
sql = """
|
||
SELECT dept_id, ancestors
|
||
FROM sys_dept
|
||
WHERE dept_name = %s AND dept_category = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql, (zy_name, dept_category))
|
||
result = cursor.fetchone()
|
||
if result:
|
||
return result[0] # 返回dept_id
|
||
return None
|
||
finally:
|
||
connection.close()
|
||
except Exception as e:
|
||
print(f"✗ 查找父部门失败: {e}")
|
||
return None
|
||
|
||
def find_major_info(self, zy_name: str, zybh: str, cursor=None) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
查找专业信息（major_id和major_name）
|
||
|
||
通过 dept_category = "zy-" + bjxx_zybh 和 dept_name = bjxx_zy 查找
|
||
|
||
Args:
|
||
zy_name: 专业名称
|
||
zybh: 专业编号
|
||
cursor: 数据库游标对象，如果提供则使用该游标，否则创建新连接
|
||
|
||
Returns:
|
||
专业信息字典（包含major_id和major_name），如果未找到则返回None
|
||
"""
|
||
if not zybh:
|
||
return None
|
||
|
||
try:
|
||
if cursor is not None:
|
||
# 使用提供的游标
|
||
dept_category = f"zy-{zybh}"
|
||
sql = """
|
||
SELECT dept_id as major_id, dept_name as major_name
|
||
FROM sys_dept
|
||
WHERE dept_name = %s AND dept_category = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql, (zy_name, dept_category))
|
||
result = cursor.fetchone()
|
||
if result:
|
||
return {
|
||
'major_id': result[0] if isinstance(result, tuple) else result.get('major_id'),
|
||
'major_name': result[1] if isinstance(result, tuple) else result.get('major_name')
|
||
}
|
||
return None
|
||
else:
|
||
# 创建新连接
|
||
connection = pymysql.connect(**DB_CONFIG)
|
||
try:
|
||
with connection.cursor(pymysql.cursors.DictCursor) as cursor:
|
||
dept_category = f"zy-{zybh}"
|
||
sql = """
|
||
SELECT dept_id as major_id, dept_name as major_name
|
||
FROM sys_dept
|
||
WHERE dept_name = %s AND dept_category = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql, (zy_name, dept_category))
|
||
result = cursor.fetchone()
|
||
return result
|
||
finally:
|
||
connection.close()
|
||
except Exception as e:
|
||
print(f"✗ 查找专业信息失败: {e}")
|
||
return None
|
||
|
||
def find_college_info(self, major_id: int, cursor=None) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
查找学院信息（college_id和college_name）
|
||
|
||
通过major_id找到parent_id，然后检查dept_category是否以"xy-"开头
|
||
|
||
Args:
|
||
major_id: 专业ID
|
||
cursor: 数据库游标对象，如果提供则使用该游标，否则创建新连接
|
||
|
||
Returns:
|
||
学院信息字典（包含college_id和college_name），如果未找到则返回None
|
||
"""
|
||
if not major_id:
|
||
return None
|
||
|
||
try:
|
||
if cursor is not None:
|
||
# 使用提供的游标（普通游标，返回元组）
|
||
# 第一步：通过major_id查找parent_id
|
||
sql1 = """
|
||
SELECT parent_id
|
||
FROM sys_dept
|
||
WHERE dept_id = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql1, (major_id,))
|
||
result1 = cursor.fetchone()
|
||
if not result1 or not result1[0]:
|
||
return None
|
||
|
||
# 普通游标返回元组，取第一个元素（parent_id）
|
||
parent_id = result1[0]
|
||
|
||
# 第二步：查询parent_id对应的部门信息，检查dept_category是否以"xy-"开头
|
||
# 根据专业信息API的代码，专业的parent_id应该就是学院的dept_id
|
||
sql2 = """
|
||
SELECT dept_id, dept_name, dept_category
|
||
FROM sys_dept
|
||
WHERE dept_id = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql2, (parent_id,))
|
||
result2 = cursor.fetchone()
|
||
if result2:
|
||
# 普通游标返回元组 (dept_id, dept_name, dept_category)
|
||
dept_id = result2[0]
|
||
dept_name = result2[1]
|
||
dept_category = result2[2] if len(result2) > 2 else None
|
||
|
||
# 检查dept_category是否以"xy-"开头（学院标识）
|
||
if dept_category and dept_category.startswith('xy-'):
|
||
return {
|
||
'college_id': dept_id,
|
||
'college_name': dept_name
|
||
}
|
||
|
||
return None
|
||
else:
|
||
# 创建新连接（使用DictCursor）
|
||
connection = pymysql.connect(**DB_CONFIG)
|
||
try:
|
||
with connection.cursor(pymysql.cursors.DictCursor) as cursor:
|
||
# 第一步：通过major_id查找parent_id
|
||
sql1 = """
|
||
SELECT parent_id
|
||
FROM sys_dept
|
||
WHERE dept_id = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql1, (major_id,))
|
||
result1 = cursor.fetchone()
|
||
if not result1 or not result1.get('parent_id'):
|
||
return None
|
||
|
||
parent_id = result1.get('parent_id')
|
||
|
||
# 第二步：查询parent_id对应的部门信息，检查dept_category是否以"xy-"开头
|
||
sql2 = """
|
||
SELECT dept_id, dept_name, dept_category
|
||
FROM sys_dept
|
||
WHERE dept_id = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql2, (parent_id,))
|
||
result2 = cursor.fetchone()
|
||
if result2:
|
||
dept_category = result2.get('dept_category')
|
||
# 检查dept_category是否以"xy-"开头（学院标识）
|
||
if dept_category and dept_category.startswith('xy-'):
|
||
return {
|
||
'college_id': result2.get('dept_id'),
|
||
'college_name': result2.get('dept_name')
|
||
}
|
||
|
||
return None
|
||
finally:
|
||
connection.close()
|
||
except Exception as e:
|
||
print(f"✗ 查找学院信息失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return None
|
||
|
||
def find_school_system(self, zybh: str, cursor=None) -> Optional[int]:
|
||
"""
|
||
查找学制（school_system）
|
||
|
||
通过bjxx_zybh关联cx_zysj表的zysj_zybh，获取cx_zysj表的zysj_xzm
|
||
|
||
Args:
|
||
zybh: 专业编号
|
||
cursor: 数据库游标对象，如果提供则使用该游标，否则创建新连接
|
||
|
||
Returns:
|
||
学制（年），如果未找到则返回None
|
||
"""
|
||
if not zybh:
|
||
return None
|
||
|
||
try:
|
||
if cursor is not None:
|
||
# 使用提供的游标
|
||
sql = """
|
||
SELECT zysj_xzm
|
||
FROM cx_zysj
|
||
WHERE zysj_zybh = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql, (zybh,))
|
||
result = cursor.fetchone()
|
||
if result and result[0]:
|
||
# 尝试将zysj_xzm转换为整数
|
||
try:
|
||
school_system = int(result[0])
|
||
return school_system
|
||
except (ValueError, TypeError):
|
||
# 如果无法转换为整数，尝试提取数字
|
||
match = re.search(r'\d+', str(result[0]))
|
||
if match:
|
||
return int(match.group())
|
||
return None
|
||
return None
|
||
else:
|
||
# 创建新连接
|
||
connection = pymysql.connect(**DB_CONFIG)
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
sql = """
|
||
SELECT zysj_xzm
|
||
FROM cx_zysj
|
||
WHERE zysj_zybh = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql, (zybh,))
|
||
result = cursor.fetchone()
|
||
if result and result[0]:
|
||
# 尝试将zysj_xzm转换为整数
|
||
try:
|
||
school_system = int(result[0])
|
||
return school_system
|
||
except (ValueError, TypeError):
|
||
# 如果无法转换为整数，尝试提取数字
|
||
import re
|
||
match = re.search(r'\d+', str(result[0]))
|
||
if match:
|
||
return int(match.group())
|
||
return None
|
||
return None
|
||
finally:
|
||
connection.close()
|
||
except Exception as e:
|
||
print(f"✗ 查找学制失败: {e}")
|
||
return None
|
||
|
||
def check_class_msg_exists(self, class_id: int, cursor=None) -> bool:
|
||
"""
|
||
检查班级信息是否已存在于xg_class_msg表
|
||
|
||
Args:
|
||
class_id: 班级ID（与sys_dept表的dept_id一致）
|
||
cursor: 数据库游标对象，如果提供则使用该游标，否则创建新连接
|
||
|
||
Returns:
|
||
是否存在
|
||
"""
|
||
try:
|
||
if cursor is not None:
|
||
# 使用提供的游标
|
||
sql = """
|
||
SELECT COUNT(*) as count
|
||
FROM xg_class_msg
|
||
WHERE id = %s
|
||
"""
|
||
cursor.execute(sql, (class_id,))
|
||
result = cursor.fetchone()
|
||
return result[0] > 0 if result else False
|
||
else:
|
||
# 创建新连接
|
||
connection = pymysql.connect(**DB_CONFIG)
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
sql = """
|
||
SELECT COUNT(*) as count
|
||
FROM xg_class_msg
|
||
WHERE id = %s
|
||
"""
|
||
cursor.execute(sql, (class_id,))
|
||
result = cursor.fetchone()
|
||
return result[0] > 0 if result else False
|
||
finally:
|
||
connection.close()
|
||
except Exception as e:
|
||
print(f"✗ 检查班级信息是否存在失败: {e}")
|
||
return False
|
||
|
||
def insert_class_msg(self, class_data: Dict[str, Any], connection) -> bool:
|
||
"""
|
||
插入班级信息到xg_class_msg表
|
||
|
||
Args:
|
||
class_data: 班级数据字典
|
||
connection: 数据库连接对象
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
insert_sql = """
|
||
INSERT INTO xg_class_msg (
|
||
id, tenant_id, create_dept, create_by, create_time,
|
||
update_by, update_time, deleted,
|
||
college_id, college_name, major_id, major_name,
|
||
class_code, class_name, class_type, grade, enrollment_year,
|
||
graduation_year, training_type, degree_type,
|
||
counselor_id, counselor_name, class_teacher_id, class_teacher_name,
|
||
class_advisor_id, class_advisor_name,
|
||
class_monitor_id, class_monitor_name,
|
||
deputy_monitor_id, deputy_monitor_name,
|
||
class_status, is_graduation, graduation_date, school_system
|
||
) 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
|
||
)
|
||
"""
|
||
cursor.execute(insert_sql, (
|
||
class_data.get('id'), # id
|
||
class_data.get('tenant_id'), # tenant_id
|
||
class_data.get('create_dept'), # create_dept
|
||
class_data.get('create_by'), # create_by
|
||
class_data.get('create_time'), # create_time
|
||
class_data.get('update_by'), # update_by
|
||
class_data.get('update_time'), # update_time
|
||
class_data.get('deleted'), # deleted
|
||
class_data.get('college_id'), # college_id
|
||
class_data.get('college_name'), # college_name
|
||
class_data.get('major_id'), # major_id
|
||
class_data.get('major_name'), # major_name
|
||
class_data.get('class_code'), # class_code
|
||
class_data.get('class_name'), # class_name
|
||
class_data.get('class_type'), # class_type
|
||
class_data.get('grade'), # grade
|
||
class_data.get('enrollment_year'), # enrollment_year
|
||
class_data.get('graduation_year'), # graduation_year
|
||
class_data.get('training_type'), # training_type
|
||
class_data.get('degree_type'), # degree_type
|
||
class_data.get('counselor_id'), # counselor_id
|
||
class_data.get('counselor_name'), # counselor_name
|
||
class_data.get('class_teacher_id'), # class_teacher_id
|
||
class_data.get('class_teacher_name'), # class_teacher_name
|
||
class_data.get('class_advisor_id'), # class_advisor_id
|
||
class_data.get('class_advisor_name'), # class_advisor_name
|
||
class_data.get('class_monitor_id'), # class_monitor_id
|
||
class_data.get('class_monitor_name'), # class_monitor_name
|
||
class_data.get('deputy_monitor_id'), # deputy_monitor_id
|
||
class_data.get('deputy_monitor_name'), # deputy_monitor_name
|
||
class_data.get('class_status'), # class_status
|
||
class_data.get('is_graduation'), # is_graduation
|
||
class_data.get('graduation_date'), # graduation_date
|
||
class_data.get('school_system') # school_system
|
||
))
|
||
return True
|
||
except Exception as e:
|
||
print(f"✗ 插入班级信息到xg_class_msg表失败: {e}")
|
||
return False
|
||
|
||
def get_parent_ancestors(self, parent_id: int) -> str:
|
||
"""
|
||
获取父部门的ancestors
|
||
|
||
Args:
|
||
parent_id: 父部门ID
|
||
|
||
Returns:
|
||
ancestors字符串（格式：0,100,101,parent_id）
|
||
"""
|
||
try:
|
||
connection = pymysql.connect(**DB_CONFIG)
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
sql = """
|
||
SELECT ancestors
|
||
FROM sys_dept
|
||
WHERE dept_id = %s
|
||
LIMIT 1
|
||
"""
|
||
cursor.execute(sql, (parent_id,))
|
||
result = cursor.fetchone()
|
||
if result and result[0]:
|
||
# 父部门的ancestors + 父部门ID
|
||
return f"{result[0]},{parent_id}"
|
||
else:
|
||
# 如果找不到父部门，返回默认值
|
||
return f"0,100,101,{parent_id}"
|
||
finally:
|
||
connection.close()
|
||
except Exception as e:
|
||
print(f"✗ 获取父部门ancestors失败: {e}")
|
||
return f"0,100,101,{parent_id}"
|
||
|
||
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_classes(self) -> Dict[str, Any]:
|
||
"""
|
||
同步班级数据到sys_dept表和xg_class_msg表
|
||
|
||
Returns:
|
||
同步结果统计（包含成功、跳过、失败的班级编号列表）
|
||
"""
|
||
# 获取符合条件的数据
|
||
valid_classes = self.get_valid_classes()
|
||
if not valid_classes:
|
||
print("没有符合条件的数据需要同步")
|
||
return {
|
||
"total": 0,
|
||
"inserted": 0,
|
||
"skipped": 0,
|
||
"failed": 0,
|
||
"inserted_bjbh": [],
|
||
"skipped_bjbh": [],
|
||
"failed_bjbh": [],
|
||
"class_msg_inserted": 0, # xg_class_msg表插入成功数
|
||
"class_msg_failed": 0 # xg_class_msg表插入失败数
|
||
}
|
||
|
||
# 生成基础时间戳（用于生成dept_id）
|
||
base_timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
|
||
|
||
# 统计信息
|
||
stats = {
|
||
"total": len(valid_classes),
|
||
"inserted": 0,
|
||
"skipped": 0,
|
||
"failed": 0,
|
||
"inserted_bjbh": [], # 成功插入的班级编号列表
|
||
"skipped_bjbh": [], # 跳过的班级编号列表
|
||
"failed_bjbh": [], # 失败的班级编号列表
|
||
"class_msg_inserted": 0, # xg_class_msg表插入成功数
|
||
"class_msg_failed": 0 # xg_class_msg表插入失败数
|
||
}
|
||
|
||
# 当前时间
|
||
current_time = datetime.now()
|
||
current_year = current_time.year
|
||
|
||
# 用于记录每个parent_id对应的最大order_num和当前order_num
|
||
parent_order_map = {}
|
||
|
||
try:
|
||
connection = pymysql.connect(**DB_CONFIG)
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
# 准备插入sys_dept的SQL
|
||
insert_dept_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 class_info in valid_classes:
|
||
bjxx_bjmc = class_info.get('bjxx_bjmc')
|
||
bjxx_bjbh = class_info.get('bjxx_bjbh')
|
||
bjxx_zy = class_info.get('bjxx_zy')
|
||
bjxx_zybh = class_info.get('bjxx_zybh')
|
||
bjxx_rxnf = class_info.get('bjxx_rxnf') # 入学年份
|
||
|
||
# 验证数据
|
||
if not bjxx_bjmc:
|
||
print(f"⚠ 跳过：班级名称为空 (bjxx_bjbh: {bjxx_bjbh})")
|
||
stats["skipped"] += 1
|
||
if bjxx_bjbh:
|
||
stats["skipped_bjbh"].append(bjxx_bjbh)
|
||
continue
|
||
|
||
if not bjxx_bjbh:
|
||
print(f"⚠ 跳过：班级编号为空 (bjxx_bjmc: {bjxx_bjmc})")
|
||
stats["skipped"] += 1
|
||
# 班级编号为空，无法记录
|
||
continue
|
||
|
||
# 检查dept_name长度（sys_dept表中dept_name是varchar(30)）
|
||
dept_name = bjxx_bjmc
|
||
if len(dept_name) > 30:
|
||
print(f"⚠ 警告：班级名称超过30字符，将被截断: {dept_name} (长度: {len(dept_name)})")
|
||
dept_name = dept_name[:30]
|
||
|
||
# 构建dept_category
|
||
dept_category = f"bj-{bjxx_bjbh}"
|
||
|
||
# 检查是否已存在
|
||
if self.check_class_exists(dept_name, dept_category):
|
||
print(f"⏭ 跳过已存在的班级: {dept_name} (dept_category: {dept_category})")
|
||
stats["skipped"] += 1
|
||
stats["skipped_bjbh"].append(bjxx_bjbh)
|
||
continue
|
||
|
||
# 查找父部门ID（专业）
|
||
parent_id = self.find_parent_dept_id(bjxx_zy, bjxx_zybh)
|
||
if parent_id is None:
|
||
print(f"⚠ 跳过：未找到父部门（专业）: {bjxx_zy} (bjxx_zybh: {bjxx_zybh}, bjxx_bjbh: {bjxx_bjbh})")
|
||
stats["skipped"] += 1
|
||
stats["skipped_bjbh"].append(bjxx_bjbh)
|
||
continue
|
||
|
||
# 获取或初始化该parent_id的order_num
|
||
if parent_id not in parent_order_map:
|
||
max_order = self.get_max_order_num(parent_id)
|
||
parent_order_map[parent_id] = max_order
|
||
parent_order_map[parent_id] += 1
|
||
current_order_num = parent_order_map[parent_id]
|
||
|
||
# 获取父部门的ancestors
|
||
ancestors = self.get_parent_ancestors(parent_id)
|
||
|
||
# 递增插入序号（只对实际插入的数据计数）
|
||
insert_index += 1
|
||
|
||
# 生成dept_id
|
||
dept_id = self.generate_dept_id(base_timestamp, insert_index)
|
||
|
||
# 准备插入sys_dept数据
|
||
try:
|
||
cursor.execute(insert_dept_sql, (
|
||
dept_id, # dept_id
|
||
self.tenant_id, # tenant_id
|
||
parent_id, # parent_id
|
||
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_bjbh"].append(bjxx_bjbh)
|
||
print(f"✓ 插入班级到sys_dept: {dept_name} (dept_id: {dept_id}, dept_category: {dept_category}, parent_id: {parent_id})")
|
||
|
||
# ========== 插入到xg_class_msg表 ==========
|
||
# 检查xg_class_msg表是否已存在
|
||
if not self.check_class_msg_exists(dept_id, cursor):
|
||
# 查找专业信息
|
||
major_info = self.find_major_info(bjxx_zy, bjxx_zybh, cursor)
|
||
if not major_info:
|
||
print(f"⚠ 警告：未找到专业信息，跳过xg_class_msg插入: {bjxx_zy} (bjxx_zybh: {bjxx_zybh})")
|
||
stats["class_msg_failed"] += 1
|
||
else:
|
||
major_id = major_info.get('major_id')
|
||
major_name = major_info.get('major_name')
|
||
|
||
# 查找学院信息
|
||
college_info = self.find_college_info(major_id, cursor)
|
||
if college_info:
|
||
college_id = college_info.get('college_id')
|
||
college_name = college_info.get('college_name')
|
||
else:
|
||
college_id = None
|
||
college_name = None
|
||
print(f" ⚠ 警告：未找到学院信息 (major_id: {major_id}, 专业: {major_name})")
|
||
|
||
# 查找学制
|
||
school_system = self.find_school_system(bjxx_zybh, cursor)
|
||
|
||
# 处理入学年份
|
||
enrollment_year = current_year
|
||
grade = str(current_year)
|
||
if bjxx_rxnf:
|
||
try:
|
||
enrollment_year = int(str(bjxx_rxnf).strip())
|
||
grade = str(enrollment_year)
|
||
except (ValueError, TypeError):
|
||
# 如果无法转换为整数，使用当前年
|
||
enrollment_year = current_year
|
||
grade = str(current_year)
|
||
|
||
# 构建xg_class_msg表的数据
|
||
class_msg_data = {
|
||
'id': dept_id, # id（与sys_dept表的dept_id一致）
|
||
'tenant_id': self.tenant_id, # tenant_id
|
||
'create_dept': 0, # create_dept
|
||
'create_by': 1, # create_by
|
||
'create_time': current_time, # create_time
|
||
'update_by': 1, # update_by
|
||
'update_time': current_time, # update_time
|
||
'deleted': 0, # deleted
|
||
'college_id': college_id, # college_id
|
||
'college_name': college_name, # college_name
|
||
'major_id': major_id, # major_id
|
||
'major_name': major_name, # major_name
|
||
'class_code': bjxx_bjbh, # class_code
|
||
'class_name': bjxx_bjmc, # class_name
|
||
'class_type': None, # class_type（为空）
|
||
'grade': grade, # grade
|
||
'enrollment_year': enrollment_year, # enrollment_year
|
||
'graduation_year': None, # graduation_year（为空）
|
||
'training_type': None, # training_type（为空）
|
||
'degree_type': None, # degree_type（为空）
|
||
'counselor_id': None, # counselor_id（为空）
|
||
'counselor_name': None, # counselor_name（为空）
|
||
'class_teacher_id': None, # class_teacher_id（为空）
|
||
'class_teacher_name': None, # class_teacher_name（为空）
|
||
'class_advisor_id': None, # class_advisor_id（为空）
|
||
'class_advisor_name': None, # class_advisor_name（为空）
|
||
'class_monitor_id': None, # class_monitor_id（为空）
|
||
'class_monitor_name': None, # class_monitor_name（为空）
|
||
'deputy_monitor_id': None, # deputy_monitor_id（为空）
|
||
'deputy_monitor_name': None, # deputy_monitor_name（为空）
|
||
'class_status': '1', # class_status（1）
|
||
'is_graduation': 0, # is_graduation（0）
|
||
'graduation_date': None, # graduation_date（为空）
|
||
'school_system': school_system # school_system
|
||
}
|
||
|
||
# 插入到xg_class_msg表
|
||
if self.insert_class_msg(class_msg_data, connection):
|
||
stats["class_msg_inserted"] += 1
|
||
print(f" ✓ 插入班级到xg_class_msg: {bjxx_bjmc} (id: {dept_id}, major_id: {major_id}, college_id: {college_id})")
|
||
else:
|
||
stats["class_msg_failed"] += 1
|
||
print(f" ✗ 插入班级到xg_class_msg失败: {bjxx_bjmc}")
|
||
else:
|
||
print(f" ⏭ 跳过已存在的xg_class_msg记录: {bjxx_bjmc} (id: {dept_id})")
|
||
|
||
except Exception as e:
|
||
stats["failed"] += 1
|
||
stats["failed_bjbh"].append(bjxx_bjbh)
|
||
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"sys_dept表 - 成功插入: {stats['inserted']}")
|
||
print(f"sys_dept表 - 跳过（已存在/未找到父部门）: {stats['skipped']}")
|
||
print(f"sys_dept表 - 失败: {stats['failed']}")
|
||
print(f"xg_class_msg表 - 成功插入: {stats.get('class_msg_inserted', 0)}")
|
||
print(f"xg_class_msg表 - 失败: {stats.get('class_msg_failed', 0)}")
|
||
print("=" * 60)
|
||
|
||
# 输出班级编号列表（逗号分隔）
|
||
if stats.get('inserted_bjbh'):
|
||
inserted_bjbh_str = ','.join(stats['inserted_bjbh'])
|
||
print(f"\n成功插入的班级编号: {inserted_bjbh_str}")
|
||
|
||
if stats.get('skipped_bjbh'):
|
||
skipped_bjbh_str = ','.join(stats['skipped_bjbh'])
|
||
print(f"跳过的班级编号: {skipped_bjbh_str}")
|
||
|
||
if stats.get('failed_bjbh'):
|
||
failed_bjbh_str = ','.join(stats['failed_bjbh'])
|
||
print(f"失败的班级编号: {failed_bjbh_str}")
|
||
|
||
print("=" * 60)
|
||
|
||
|
||
def main():
|
||
"""主函数 - 完整同步流程"""
|
||
# API配置信息
|
||
APP_ID = "73462569-d0da-4a05-bd6a-a9077c723dba"
|
||
APP_KEY = "dyjs_bjxx"
|
||
FID = "19628" # 单位ID
|
||
|
||
print("=" * 60)
|
||
print("超星班级信息完整同步流程")
|
||
print("=" * 60)
|
||
print("流程：超星API -> cx_bjsj表 -> sys_dept表")
|
||
print("=" * 60)
|
||
|
||
# ========== 第一阶段：从超星API获取数据并保存到cx_bjsj表 ==========
|
||
print("\n" + "=" * 60)
|
||
print("第一阶段：从超星API获取班级信息")
|
||
print("=" * 60)
|
||
|
||
# 创建API客户端
|
||
api = ChaoxingClassAPI(app_id=APP_ID, app_key=APP_KEY, fid=FID)
|
||
|
||
# 步骤1: 清空表
|
||
print("\n步骤1: 清空班级数据表（cx_bjsj）...")
|
||
api.clear_table()
|
||
|
||
# 步骤2: 获取所有数据
|
||
print("\n步骤2: 获取所有班级信息（自动分页）...")
|
||
all_classes = api.get_all_classes(max_page_size=500)
|
||
|
||
# 步骤3: 保存到数据库
|
||
if all_classes:
|
||
print(f"\n步骤3: 保存数据到cx_bjsj表...")
|
||
saved_count = api.save_to_database(all_classes)
|
||
print(f"\n✓ 第一阶段完成！共获取 {len(all_classes)} 条数据，成功保存 {saved_count} 条到cx_bjsj表")
|
||
else:
|
||
print("\n✗ 未获取到数据，终止流程")
|
||
return
|
||
|
||
# ========== 第二阶段：从cx_bjsj表同步到sys_dept表 ==========
|
||
print("\n" + "=" * 60)
|
||
print("第二阶段：从cx_bjsj表同步到sys_dept表")
|
||
print("=" * 60)
|
||
print(f"筛选条件: bjxx_sfqy='是'")
|
||
print(f"判断存在: dept_category='bj-' + bjxx_bjbh AND dept_name=bjxx_bjmc")
|
||
print(f"父部门查找: dept_category='zy-' + bjxx_zybh AND dept_name=bjxx_zy")
|
||
print("=" * 60)
|
||
|
||
# 创建同步实例
|
||
sync = ClassSync()
|
||
|
||
# 执行同步
|
||
stats = sync.sync_classes()
|
||
|
||
# 打印统计信息
|
||
sync.print_statistics(stats)
|
||
|
||
# ========== 总结 ==========
|
||
print("\n" + "=" * 60)
|
||
print("完整同步流程总结")
|
||
print("=" * 60)
|
||
print(f"✓ 从超星API获取: {len(all_classes)} 条数据")
|
||
print(f"✓ 保存到cx_bjsj表: {saved_count} 条数据")
|
||
print(f"✓ 同步到sys_dept表: 插入 {stats['inserted']} 条，跳过 {stats['skipped']} 条，失败 {stats['failed']} 条")
|
||
print(f"✓ 同步到xg_class_msg表: 插入 {stats.get('class_msg_inserted', 0)} 条，失败 {stats.get('class_msg_failed', 0)} 条")
|
||
print("=" * 60)
|
||
|
||
if stats["inserted"] > 0:
|
||
print("\n✓ 完整同步流程执行成功！")
|
||
elif stats["skipped"] == stats["total"]:
|
||
print("\n✓ 所有数据已存在，无需同步")
|
||
else:
|
||
print("\n⚠ 同步过程中出现问题，请检查日志")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|