855 lines
37 KiB
Python
855 lines
37 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
06 - 班级数据入库
|
||
|
||
从 his_xssj 表将班级数据导入学工表：
|
||
- sys_dept
|
||
- xg_class_msg
|
||
- xg_class_teacher_counselor_rel
|
||
|
||
脚本特性：
|
||
- 列名映射可配置（见 COLUMN_MAP）
|
||
- 学院编号映射（见 COLLEGE_CODE_TO_NAME）
|
||
- 对班级存在性检测：使用 sys_dept(dept_name, dept_category)
|
||
- 使用时间戳+序号生成 dept_id 与 class id（与现有脚本保持一致）
|
||
- 对班级中学生状态判定 class_status 与 is_graduation
|
||
- 对班主任通过 sys_user.nick_name 进行唯一性解析
|
||
|
||
使用方法：
|
||
- 在运行前确认 `db_config.py` 中 `DB_CONFIG` 可用
|
||
- 如需先预览请设置 DRY_RUN = True
|
||
"""
|
||
from __future__ import annotations
|
||
import time
|
||
from datetime import datetime
|
||
from typing import Dict, Any, List, Optional, Tuple
|
||
import pymysql
|
||
from db_config import DB_CONFIG
|
||
import os
|
||
import json
|
||
|
||
# 可根据 his_xssj 表模型调整映射（使用真实中文列名）
|
||
COLUMN_MAP: Dict[str, str] = {
|
||
"class_code": "班级编号", # 班级编号
|
||
"class_name": "班级名称", # 班级名称
|
||
"major_code": "专业编号", # 专业编号
|
||
"major_name": "专业名称", # 专业名称
|
||
"college_code": "学院编号", # 学院编号
|
||
"enrollment_year": "入学年份", # 入学年份（文本，需要转换为 int）
|
||
"grade": "所属年级", # 所属年级
|
||
"school_system": "学制编码", # 学制编码（文本，需要转换为 int）
|
||
"class_type": "层次编号", # 层次编号
|
||
"training_type": "学生类别", # 学生类别/培养方式
|
||
"class_teacher_name": "班主任", # 班主任姓名
|
||
"student_status": "学生状态", # 学生状态（判定是否为'在读'）
|
||
"enrollment_date": "入学日期" # 入学日期（可为空）
|
||
}
|
||
|
||
# 学院编号对应关系（题主给定）
|
||
COLLEGE_CODE_TO_NAME: Dict[str, str] = {
|
||
"14": "智能控制学院",
|
||
"15": "智能制造学院",
|
||
"19": "人文学院",
|
||
"18": "餐饮休闲学院",
|
||
"28": "数字信息学院",
|
||
"16": "医药健康学院",
|
||
"17": "生态园林学院"
|
||
}
|
||
|
||
def extract_distinct_classes_to_his_bjsj(connection, dry_run: bool = False) -> int:
|
||
"""
|
||
（已移除写入逻辑）该函数保留为占位：如果需要从脚本内写入 his_bjsj，请使用外部迁移或重新启用写入逻辑。
|
||
当前实现仅返回 0 或在 dry_run 模式下打印示例，不执行对 his_bjsj 的 INSERT/DELETE 操作。
|
||
"""
|
||
select_sql = """
|
||
SELECT DISTINCT
|
||
`入学年份` AS enrollment_year,
|
||
`入学日期` AS enrollment_date,
|
||
`教学系部` AS teaching_dept,
|
||
`班级名称` AS class_name,
|
||
`班主任` AS class_teacher,
|
||
`学生类别` AS student_type,
|
||
`学院编号` AS college_code,
|
||
`专业编号` AS major_code,
|
||
`专业名称` AS major_name,
|
||
`层次编号` AS level_code,
|
||
`报读层次` AS level_name,
|
||
`所属年级` AS grade,
|
||
`班级编号` AS class_code,
|
||
`学制编码` AS school_system
|
||
FROM his_xssj
|
||
"""
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(select_sql)
|
||
rows = cursor.fetchall()
|
||
if dry_run:
|
||
print(f"[DRY_RUN] 将从 his_xssj 提取 {len(rows)} 条去重班级记录（示例前 5 条）")
|
||
for r in rows[:5]:
|
||
print(r)
|
||
return 0
|
||
|
||
def get_connection():
|
||
return pymysql.connect(cursorclass=pymysql.cursors.DictCursor, **DB_CONFIG)
|
||
|
||
|
||
def fetch_distinct_classes(connection) -> List[Dict[str, Any]]:
|
||
"""
|
||
从 his_xssj 中取出按班级分组的基础信息（去重后的班级列表）
|
||
"""
|
||
cols = COLUMN_MAP
|
||
sql = f"""
|
||
SELECT
|
||
`{cols['class_code']}` AS class_code,
|
||
`{cols['class_name']}` AS class_name,
|
||
`{cols['major_code']}` AS major_code,
|
||
`{cols['major_name']}` AS major_name,
|
||
`{cols['college_code']}` AS college_code,
|
||
`{cols['enrollment_year']}` AS enrollment_year,
|
||
`{cols['grade']}` AS grade,
|
||
`{cols['school_system']}` AS school_system,
|
||
`{cols['class_type']}` AS class_type,
|
||
`{cols['training_type']}` AS training_type,
|
||
`{cols['class_teacher_name']}` AS class_teacher_name,
|
||
`{cols['enrollment_date']}` AS enrollment_date
|
||
FROM his_xssj
|
||
GROUP BY class_code, class_name, major_code, major_name, college_code, enrollment_year, grade, school_system, class_type, training_type, class_teacher_name, enrollment_date
|
||
ORDER BY enrollment_year ASC, class_name ASC
|
||
"""
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql)
|
||
return cursor.fetchall()
|
||
|
||
|
||
def fetch_classes_from_his_bjsj(connection) -> List[Dict[str, Any]]:
|
||
"""
|
||
从 his_bjsj 中读取班级数据，作为后续插入学工表的唯一数据源。
|
||
返回的字段与原来从 his_xssj 聚合出的字段结构一致（class_code, class_name, major_code, major_name, college_code, enrollment_year, grade, school_system, class_type, training_type, class_teacher_name, enrollment_date）。
|
||
注意：his_bjsj 表需事先存在且已由 extract_distinct_classes_to_his_bjsj 填充。
|
||
"""
|
||
sql = """
|
||
SELECT
|
||
class_code AS class_code,
|
||
class_name AS class_name,
|
||
major_code AS major_code,
|
||
major_name AS major_name,
|
||
college_code AS college_code,
|
||
enrollment_year AS enrollment_year,
|
||
grade AS grade,
|
||
school_system AS school_system,
|
||
level_code AS class_type,
|
||
student_type AS training_type,
|
||
class_teacher AS class_teacher_name,
|
||
enrollment_date AS enrollment_date
|
||
FROM his_bjsj
|
||
ORDER BY enrollment_year ASC, class_name ASC
|
||
"""
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql)
|
||
return cursor.fetchall()
|
||
|
||
|
||
def class_has_active_student(connection, class_code: str) -> bool:
|
||
"""
|
||
判断班级内是否存在至少一名学生状态为 '在读'
|
||
"""
|
||
cols = COLUMN_MAP
|
||
sql = f"""
|
||
SELECT COUNT(1) AS cnt FROM his_xssj
|
||
WHERE `{cols['class_code']}` = %s AND `{cols['student_status']}` = '在读'
|
||
"""
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql, (class_code,))
|
||
row = cursor.fetchone()
|
||
return (row and row.get("cnt", 0) > 0)
|
||
|
||
|
||
def resolve_major_dept(connection, major_name: str, major_code: str, college_code: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
根据规则查找专业部门，优先尝试精确匹配 (dept_category = 'zy-'+major_code AND dept_name = major_name)。
|
||
若未命中且提供了 college_code，则：
|
||
1) 通过 dept_category = 'xy-'+college_code 在 sys_dept 查到学院 parent_id；
|
||
2) 在该学院的子节点中（parent_id = 学院dept_id，且 dept_category LIKE 'zy-%'）按 dept_name 精确匹配 major_name。
|
||
返回 dict 包含 dept_id, ancestors, parent_id 等或 None。
|
||
"""
|
||
# 1) 精确匹配（保留原有逻辑）
|
||
if major_name is not None and major_code is not None and str(major_code).strip() != "":
|
||
sql = """
|
||
SELECT dept_id, parent_id, ancestors, dept_name, dept_category
|
||
FROM sys_dept
|
||
WHERE dept_name = %s AND dept_category = %s
|
||
LIMIT 1
|
||
"""
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql, (major_name, f"zy-{major_code}"))
|
||
row = cursor.fetchone()
|
||
if row:
|
||
return row
|
||
|
||
# 2) 若未匹配到，且提供 college_code，则根据 his_bjsj 的 college_code 关联学院，再在学院下按名称查找专业
|
||
if college_code:
|
||
try:
|
||
sql_college = "SELECT dept_id FROM sys_dept WHERE dept_category = %s LIMIT 1"
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql_college, (f"xy-{college_code}",))
|
||
college_row = cursor.fetchone()
|
||
if college_row and college_row.get("dept_id") is not None:
|
||
college_dept_id = college_row["dept_id"]
|
||
sql_child = """
|
||
SELECT dept_id, parent_id, ancestors, dept_name, dept_category
|
||
FROM sys_dept
|
||
WHERE parent_id = %s AND dept_category LIKE %s AND dept_name = %s
|
||
LIMIT 1
|
||
"""
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql_child, (college_dept_id, "zy-%", major_name))
|
||
child = cursor.fetchone()
|
||
if child:
|
||
return child
|
||
except Exception:
|
||
# 不阻塞主流程，继续回退为 None
|
||
pass
|
||
|
||
return None
|
||
|
||
|
||
def resolve_college_dept(connection, college_code: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
通过学院编号查找学院在 sys_dept 中的 dept_id（dept_category = 'xy-' + code）
|
||
"""
|
||
sql = """
|
||
SELECT dept_id, dept_name
|
||
FROM sys_dept
|
||
WHERE dept_category = %s
|
||
LIMIT 1
|
||
"""
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql, (f"xy-{college_code}",))
|
||
return cursor.fetchone()
|
||
|
||
|
||
def resolve_teacher_user(connection, teacher_name: str) -> Optional[int]:
|
||
"""
|
||
通过 sys_user.nick_name 查找教师 user_id
|
||
- 唯一命中返回 user_id
|
||
- 多于或少于 1 条返回 None
|
||
"""
|
||
if not teacher_name:
|
||
return None
|
||
sql = """
|
||
SELECT user_id FROM sys_user WHERE nick_name = %s
|
||
"""
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql, (teacher_name,))
|
||
rows = cursor.fetchall()
|
||
if len(rows) == 1:
|
||
return rows[0].get("user_id")
|
||
return None
|
||
|
||
|
||
def find_similar_majors(connection, major_name: str, limit: int = 5) -> List[Dict[str, Any]]:
|
||
"""
|
||
在 sys_dept 中查找与 major_name 相似的专业（dept_category LIKE 'zy-%'）作为候选建议。
|
||
返回最多 limit 条候选，元素包含 dept_id 与 dept_name。
|
||
"""
|
||
if not major_name:
|
||
return []
|
||
sql = """
|
||
SELECT dept_id, dept_name, dept_category
|
||
FROM sys_dept
|
||
WHERE dept_category LIKE %s AND dept_name LIKE %s
|
||
LIMIT %s
|
||
"""
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql, ('zy-%', f"%{major_name}%", limit))
|
||
return cursor.fetchall()
|
||
|
||
|
||
LOG_DIR = "logs"
|
||
SKIP_LOG_FILE = os.path.join(LOG_DIR, "06_insert_classes_skipped_majors.jsonl")
|
||
|
||
|
||
def ensure_log_dir():
|
||
if not os.path.exists(LOG_DIR):
|
||
os.makedirs(LOG_DIR, exist_ok=True)
|
||
|
||
|
||
def log_skipped_major(connection, row: Dict[str, Any]):
|
||
"""
|
||
将未找到父专业（major）导致跳过的班级信息记录到日志文件（JSONL），并返回候选建议列表。
|
||
"""
|
||
ensure_log_dir()
|
||
suggestions = find_similar_majors(connection, row.get("major_name") or "")
|
||
# 尝试获取 his_xssj 中该班级的所有原始记录（完整行），以便人工核对
|
||
his_rows: List[Dict[str, Any]] = []
|
||
class_code = row.get("class_code")
|
||
his_class_col = COLUMN_MAP.get("class_code")
|
||
try:
|
||
with connection.cursor() as c:
|
||
if his_class_col and class_code is not None:
|
||
sql = f"SELECT * FROM his_xssj WHERE `{his_class_col}` = %s"
|
||
c.execute(sql, (class_code,))
|
||
his_rows = c.fetchall()
|
||
except Exception:
|
||
his_rows = []
|
||
|
||
entry = {
|
||
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"class_code": class_code,
|
||
"class_name": row.get("class_name"),
|
||
"major_code": row.get("major_code"),
|
||
"major_name": row.get("major_name"),
|
||
"college_code": row.get("college_code"),
|
||
"row": row,
|
||
"his_rows_count": len(his_rows),
|
||
"his_rows": his_rows,
|
||
"suggestions": suggestions
|
||
}
|
||
try:
|
||
with open(SKIP_LOG_FILE, "a", encoding="utf-8") as fh:
|
||
fh.write(json.dumps(entry, ensure_ascii=False, default=str) + "\n")
|
||
except Exception:
|
||
# 写日志失败不应阻塞主流程
|
||
pass
|
||
return suggestions
|
||
|
||
|
||
def get_parent_ancestors(connection, parent_id: int) -> str:
|
||
sql = "SELECT ancestors FROM sys_dept WHERE dept_id = %s LIMIT 1"
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql, (parent_id,))
|
||
row = cursor.fetchone()
|
||
if row and row.get("ancestors"):
|
||
return f"{row['ancestors']},{parent_id}"
|
||
return f"0,100,101,{parent_id}"
|
||
|
||
|
||
def get_max_order_num(connection, parent_id: int) -> int:
|
||
sql = "SELECT MAX(order_num) AS max_order FROM sys_dept WHERE parent_id = %s"
|
||
with connection.cursor() as cursor:
|
||
cursor.execute(sql, (parent_id,))
|
||
row = cursor.fetchone()
|
||
return int(row["max_order"]) if row and row["max_order"] is not None else 0
|
||
|
||
|
||
def generate_id(base_timestamp: str, index: int) -> int:
|
||
"""
|
||
生成 dept_id / class id： timestamp + 四位序号（与原脚本保持一致）
|
||
"""
|
||
return int(f"{base_timestamp}{str(index).zfill(4)}")
|
||
|
||
|
||
def parse_date_to_mysql(value: Any) -> Optional[str]:
|
||
"""
|
||
将各种可能的日期/时间值规范化为 MySQL DATETIME 字符串 'YYYY-MM-DD HH:MM:SS'。
|
||
- 支持 datetime 对象直接格式化
|
||
- 支持常见字符串格式，如 'YYYY-MM-DD HH:MM:SS', 'YYYY/MM/DD HH:MM:SS', 'M/D/YYYY H:M:S', 'YYYY-MM-DD', 'M/D/YYYY'
|
||
- 若无法解析或为 None，返回 None
|
||
"""
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, datetime):
|
||
return value.strftime("%Y-%m-%d %H:%M:%S")
|
||
if not isinstance(value, str):
|
||
try:
|
||
# 尝试将数字型视为时间戳（秒）
|
||
ts = float(value)
|
||
dt = datetime.fromtimestamp(ts)
|
||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||
except Exception:
|
||
return None
|
||
|
||
s = value.strip()
|
||
if s == "":
|
||
return None
|
||
|
||
patterns = [
|
||
"%Y-%m-%d %H:%M:%S",
|
||
"%Y/%m/%d %H:%M:%S",
|
||
"%m/%d/%Y %H:%M:%S",
|
||
"%m/%d/%y %H:%M:%S",
|
||
"%Y-%m-%d",
|
||
"%Y/%m/%d",
|
||
"%m/%d/%Y",
|
||
"%m/%d/%y",
|
||
]
|
||
for p in patterns:
|
||
try:
|
||
dt = datetime.strptime(s, p)
|
||
# 如果没有时分秒，补 00:00:00
|
||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||
except Exception:
|
||
continue
|
||
# 兜底：尝试解析含有空格并用 '/' 或 '-' 的混合格式（例如 '9/1/2009 0:0:0' 已被覆盖上面，但保留此处以防）
|
||
try:
|
||
parts = s.split()
|
||
date_part = parts[0]
|
||
time_part = parts[1] if len(parts) > 1 else "0:0:0"
|
||
md = date_part.replace("-", "/")
|
||
combined = f"{md} {time_part}"
|
||
dt = datetime.strptime(combined, "%m/%d/%Y %H:%M:%S")
|
||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def insert_sys_dept(cursor, dept_id: int, tenant_id: str, parent_id: int, ancestors: str, dept_name: str, dept_category: str, order_num: int, current_time: datetime):
|
||
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)
|
||
"""
|
||
params = (
|
||
dept_id, tenant_id, parent_id, ancestors, dept_name, dept_category,
|
||
order_num, None, None, None, '0', '0', 0, 1, current_time, 1, current_time
|
||
)
|
||
cursor.execute(sql, params)
|
||
|
||
|
||
def insert_xg_class_msg(cursor, class_data: Dict[str, Any]):
|
||
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
|
||
)
|
||
"""
|
||
params = (
|
||
class_data.get("id"),
|
||
class_data.get("tenant_id", "000000"),
|
||
class_data.get("create_dept", 0),
|
||
class_data.get("create_by", 1),
|
||
class_data.get("create_time"),
|
||
class_data.get("update_by", 1),
|
||
class_data.get("update_time"),
|
||
class_data.get("deleted", 0),
|
||
class_data.get("college_id"),
|
||
class_data.get("college_name"),
|
||
class_data.get("major_id"),
|
||
class_data.get("major_name"),
|
||
class_data.get("class_code"),
|
||
class_data.get("class_name"),
|
||
class_data.get("class_type"),
|
||
class_data.get("grade"),
|
||
class_data.get("enrollment_year"),
|
||
class_data.get("graduation_year"),
|
||
class_data.get("training_type"),
|
||
None, # degree_type
|
||
None, None,
|
||
class_data.get("class_teacher_id"),
|
||
class_data.get("class_teacher_name"),
|
||
None, None, None, None, None, None,
|
||
class_data.get("class_status"),
|
||
class_data.get("is_graduation"),
|
||
class_data.get("graduation_date"),
|
||
class_data.get("school_system")
|
||
)
|
||
cursor.execute(sql, params)
|
||
|
||
|
||
def insert_teacher_rel(cursor, rel_data: Dict[str, Any]):
|
||
sql = """
|
||
INSERT INTO xg_class_teacher_counselor_rel (
|
||
id, tenant_id, create_dept, create_by, create_time,
|
||
update_by, update_time, deleted, class_id, class_name,
|
||
teacher_id, teacher_name, teacher_type, start_date, end_date, remark
|
||
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
"""
|
||
params = (
|
||
rel_data.get("id"),
|
||
rel_data.get("tenant_id", "000000"),
|
||
rel_data.get("create_dept", 0),
|
||
rel_data.get("create_by", 1),
|
||
rel_data.get("create_time"),
|
||
rel_data.get("update_by", 1),
|
||
rel_data.get("update_time"),
|
||
rel_data.get("deleted", 0),
|
||
rel_data.get("class_id"),
|
||
rel_data.get("class_name"),
|
||
rel_data.get("teacher_id"),
|
||
rel_data.get("teacher_name"),
|
||
rel_data.get("teacher_type"),
|
||
rel_data.get("start_date"),
|
||
rel_data.get("end_date"),
|
||
rel_data.get("remark")
|
||
)
|
||
cursor.execute(sql, params)
|
||
|
||
|
||
def main():
|
||
print("开始：06 - 班级数据入库")
|
||
base_timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
|
||
seq_counter = 0
|
||
|
||
connection = get_connection()
|
||
try:
|
||
# 在开始时询问是否删除 sys_dept 中 dept_category 以 bj- 开头的记录
|
||
def prompt_delete_sys_dept_bj(conn) -> bool:
|
||
try:
|
||
ans = input("是否先删除 `sys_dept` 中 dept_category 以 'bj-' 开头的记录？ [y/N]: ").strip().lower()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print("\\n未检测到交互输入，默认不删除 sys_dept 的 bj- 记录。")
|
||
return False
|
||
if ans not in ("y", "yes"):
|
||
return False
|
||
try:
|
||
with conn.cursor() as c:
|
||
c.execute("SELECT COUNT(1) AS cnt FROM sys_dept WHERE dept_category LIKE %s", (f"bj-%",))
|
||
cnt = int(c.fetchone().get("cnt", 0))
|
||
print(f"将删除 sys_dept 中 {cnt} 条 dept_category LIKE 'bj-%' 的记录。")
|
||
try:
|
||
confirm = input("确认删除并继续？此操作不可恢复。输入 y 确认： ").strip().lower()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print("\\n未检测到交互输入，取消删除。")
|
||
return False
|
||
if confirm not in ("y", "yes"):
|
||
print("已取消删除操作，继续执行导入（不删除）。")
|
||
return False
|
||
with conn.cursor() as c:
|
||
c.execute("DELETE FROM sys_dept WHERE dept_category LIKE %s", (f"bj-%",))
|
||
conn.commit()
|
||
print("已删除并提交：sys_dept 中 dept_category LIKE 'bj-%' 的记录。")
|
||
return True
|
||
except Exception as e:
|
||
print("删除 sys_dept 过程中发生异常，跳过删除：", e)
|
||
return False
|
||
|
||
try:
|
||
deleted_sys_dept_bj = prompt_delete_sys_dept_bj(connection)
|
||
if deleted_sys_dept_bj:
|
||
print("用户已确认并删除 sys_dept 中 bj- 开头的记录，继续后续流程。")
|
||
else:
|
||
print("未删除 sys_dept 的 bj- 记录，继续执行导入流程。")
|
||
except Exception as e:
|
||
print("询问或删除 sys_dept(bj-) 过程中发生异常，继续执行导入：", e)
|
||
|
||
# 在正式执行导入前，询问用户是否要先删除历史数据
|
||
def prompt_delete_tables(conn) -> bool:
|
||
"""
|
||
交互式询问并在确认后删除 xg_class_msg 与 xg_class_teacher_counselor_rel 中的数据。
|
||
返回 True 表示已执行删除并提交，False 表示未删除。
|
||
"""
|
||
try:
|
||
ans = input("是否先删除 `xg_class_msg` 与 `xg_class_teacher_counselor_rel` 中的现有数据？ [y/N]: ").strip().lower()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print("\\n未检测到交互输入，默认不删除。")
|
||
return False
|
||
if ans not in ("y", "yes"):
|
||
return False
|
||
|
||
with conn.cursor() as c:
|
||
c.execute("SELECT COUNT(1) AS cnt FROM xg_class_msg")
|
||
cnt_msg = int(c.fetchone().get("cnt", 0))
|
||
c.execute("SELECT COUNT(1) AS cnt FROM xg_class_teacher_counselor_rel")
|
||
cnt_rel = int(c.fetchone().get("cnt", 0))
|
||
print(f"将删除 xg_class_msg 中 {cnt_msg} 条，xg_class_teacher_counselor_rel 中 {cnt_rel} 条。")
|
||
try:
|
||
confirm = input("确认删除并继续？此操作不可恢复。输入 y 确认： ").strip().lower()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print("\\n未检测到交互输入，取消删除。")
|
||
return False
|
||
if confirm not in ("y", "yes"):
|
||
print("已取消删除操作，继续执行导入（不删除）。")
|
||
return False
|
||
|
||
with conn.cursor() as c:
|
||
# 先删除关系表，再删除班级表
|
||
c.execute("DELETE FROM xg_class_teacher_counselor_rel")
|
||
c.execute("DELETE FROM xg_class_msg")
|
||
conn.commit()
|
||
print("已删除并提交：xg_class_msg 与 xg_class_teacher_counselor_rel。")
|
||
return True
|
||
|
||
# 在继续前询问是否删除历史数据（仅在交互式环境有意义）
|
||
try:
|
||
deleted = prompt_delete_tables(connection)
|
||
if deleted:
|
||
print("用户已确认并删除历史数据，继续执行后续导入流程。")
|
||
else:
|
||
print("未删除历史数据，继续执行导入流程。")
|
||
except Exception as e:
|
||
print("询问或删除过程中发生异常，继续执行导入：", e)
|
||
|
||
# 注意：脚本已移除对 his_bjsj 的写入/删除操作。请确保 his_bjsj 表由外部迁移或手动创建并填充。
|
||
print("提示：脚本不会对 his_bjsj 执行删除或插入操作，后续将直接从 his_bjsj 读取数据。")
|
||
|
||
# 从 his_bjsj 中读取班级数据作为唯一数据源（his_bjsj 应由前面的抽取步骤或外部迁移提前填充）
|
||
class_rows = fetch_classes_from_his_bjsj(connection)
|
||
print(f"共发现 {len(class_rows)} 个去重后的班级记录（将按专业分组并插入）")
|
||
|
||
# 预分组：按 major(parent) 收集要插入的班级以便计算 order_num
|
||
by_parent: Dict[int, List[Dict[str, Any]]] = {}
|
||
prepared_classes: List[Dict[str, Any]] = []
|
||
|
||
for r in class_rows:
|
||
class_code = r.get("class_code")
|
||
class_name = r.get("class_name")
|
||
major_code = str(r.get("major_code") or "")
|
||
major_name = r.get("major_name")
|
||
|
||
major_dept = resolve_major_dept(connection, major_name, major_code, str(r.get("college_code") or ""))
|
||
# 若未命中，则尝试从 his_bjsj 补全班级信息后再次匹配
|
||
if not major_dept:
|
||
try:
|
||
with connection.cursor() as c:
|
||
c.execute("SELECT major_code, major_name, college_code, enrollment_year, enrollment_date FROM his_bjsj WHERE class_code = %s LIMIT 1", (class_code,))
|
||
bj = c.fetchone()
|
||
if bj:
|
||
# 用 his_bjsj 的值覆盖 row 中的同类字段以便后续匹配
|
||
major_code = str(bj.get("major_code") or "")
|
||
major_name = bj.get("major_name") or major_name
|
||
# 也更新原始 row，便于日志和后续处理
|
||
r["major_code"] = major_code
|
||
r["major_name"] = major_name
|
||
r["college_code"] = bj.get("college_code") or r.get("college_code")
|
||
r["enrollment_year"] = bj.get("enrollment_year") or r.get("enrollment_year")
|
||
r["enrollment_date"] = bj.get("enrollment_date") or r.get("enrollment_date")
|
||
major_dept = resolve_major_dept(connection, major_name, major_code, str(r.get("college_code") or ""))
|
||
except Exception:
|
||
major_dept = None
|
||
|
||
if not major_dept:
|
||
# 记录并打印更详细的信息（包含专业编号与名称），同时输出候选建议
|
||
suggestions = log_skipped_major(connection, r)
|
||
print(f"跳过班级（未找到父专业）：{class_code} {class_name} 专业({major_code} - {major_name})")
|
||
if suggestions:
|
||
cand_str = "; ".join([f"{s.get('dept_id')}:{s.get('dept_name')}" for s in suggestions])
|
||
print(f" 建议候选专业：{cand_str}")
|
||
else:
|
||
print(" 无相似候选专业。已写入 logs/06_insert_classes_skipped_majors.jsonl 以便人工核对。")
|
||
continue
|
||
|
||
parent_id = major_dept["dept_id"]
|
||
prepared = {
|
||
"row": r,
|
||
"parent_id": parent_id,
|
||
"major_dept": major_dept
|
||
}
|
||
by_parent.setdefault(parent_id, []).append(prepared)
|
||
prepared_classes.append(prepared)
|
||
|
||
# 在每个 parent_id 下按 enrollment_year, class_name 排序并计算 order_num 起点
|
||
insertion_plan: List[Tuple[int, Dict[str, Any]]] = [] # list of (order_num, prepared)
|
||
for parent_id, class_list in by_parent.items():
|
||
max_order = get_max_order_num(connection, parent_id)
|
||
# sort by enrollment_year (统一转为 int，发生异常或为空时视为 0) 然后按 class_name
|
||
def _safe_year_key(p: Dict[str, Any]) -> int:
|
||
val = p["row"].get("enrollment_year")
|
||
if val is None or val == "":
|
||
return 0
|
||
try:
|
||
return int(val)
|
||
except Exception:
|
||
return 0
|
||
|
||
class_list.sort(key=lambda p: (_safe_year_key(p), p["row"].get("class_name") or ""))
|
||
for idx, item in enumerate(class_list, start=1):
|
||
item_order = max_order + idx
|
||
item["order_num"] = item_order
|
||
insertion_plan.append((item_order, item))
|
||
|
||
# 按插入顺序执行插入（可调整为按parent分批提交）
|
||
inserted_count = 0
|
||
class_msg_inserted = 0
|
||
teacher_rel_inserted = 0
|
||
|
||
# 使用单个连接事务（最后 commit）
|
||
with connection.cursor() as cursor:
|
||
for item in sorted(insertion_plan, key=lambda x: (x[1]["parent_id"], x[0])):
|
||
prepared = item[1]
|
||
row = prepared["row"]
|
||
parent_id = prepared["parent_id"]
|
||
order_num = prepared["order_num"]
|
||
|
||
class_code = row.get("class_code")
|
||
class_name = row.get("class_name") or ""
|
||
enrollment_year = row.get("enrollment_year")
|
||
grade = row.get("grade") or str(enrollment_year or "")
|
||
school_system_raw = row.get("school_system")
|
||
try:
|
||
school_system = int(school_system_raw) if school_system_raw not in (None, "") else None
|
||
except Exception:
|
||
school_system = None
|
||
class_type = row.get("class_type")
|
||
training_type = row.get("training_type")
|
||
teacher_name = row.get("class_teacher_name")
|
||
college_code = str(row.get("college_code") or "")
|
||
|
||
# 判断班级是否已存在（sys_dept: dept_category = 'bj-' + class_code）
|
||
dept_category = f"bj-{class_code}"
|
||
cursor.execute("SELECT COUNT(1) AS cnt FROM sys_dept WHERE dept_name = %s AND dept_category = %s", (class_name, dept_category))
|
||
if cursor.fetchone().get("cnt", 0) > 0:
|
||
print(f"已存在 sys_dept - 跳过: {class_name} ({dept_category})")
|
||
continue
|
||
|
||
# 生成 dept_id
|
||
seq_counter += 1
|
||
dept_id = generate_id(base_timestamp, seq_counter)
|
||
ancestors = get_parent_ancestors(connection, parent_id)
|
||
current_time = datetime.now()
|
||
|
||
# 插入 sys_dept
|
||
insert_sys_dept(cursor, dept_id, "000000", parent_id, ancestors, class_name[:30], dept_category, order_num, current_time)
|
||
inserted_count += 1
|
||
print(f"插入 sys_dept: dept_id={dept_id}, parent_id={parent_id}, name={class_name}, order_num={order_num}")
|
||
|
||
# ========== 构建并插入 xg_class_msg ==========
|
||
# major info
|
||
major_info = prepared["major_dept"]
|
||
major_id = major_info.get("dept_id")
|
||
major_name = major_info.get("dept_name", row.get("major_name"))
|
||
|
||
# college info（通过 college_code -> name -> 在 sys_dept 查找）
|
||
college_name_guess = COLLEGE_CODE_TO_NAME.get(college_code)
|
||
college_id = None
|
||
college_name = None
|
||
if college_name_guess:
|
||
# 尝试在 sys_dept 中查找学院 dept_id
|
||
cursor.execute("SELECT dept_id, dept_name FROM sys_dept WHERE dept_category = %s LIMIT 1", (f"xy-{college_code}",))
|
||
crow = cursor.fetchone()
|
||
if crow:
|
||
college_id = crow.get("dept_id")
|
||
college_name = crow.get("dept_name")
|
||
else:
|
||
# 退回到 mapping 的名称
|
||
college_name = college_name_guess
|
||
else:
|
||
# 尝试直接用 college_code 在 sys_dept 查找
|
||
col_dept = resolve_college_dept(connection, college_code)
|
||
if col_dept:
|
||
college_id = col_dept.get("dept_id")
|
||
college_name = col_dept.get("dept_name")
|
||
|
||
# 如果仍未解析到 college_id，尝试通过学院名称在 sys_dept 中查找（fallback）
|
||
if not college_id:
|
||
# 优先使用映射名称，其次使用已有 college_name
|
||
search_name = COLLEGE_CODE_TO_NAME.get(college_code) or college_name
|
||
if search_name:
|
||
try:
|
||
cursor.execute("SELECT dept_id, dept_name FROM sys_dept WHERE dept_name = %s AND dept_category LIKE %s LIMIT 1", (search_name, "xy-%"))
|
||
crow2 = cursor.fetchone()
|
||
if crow2:
|
||
college_id = crow2.get("dept_id")
|
||
college_name = crow2.get("dept_name")
|
||
except Exception:
|
||
# 不阻塞主流程，保留 college_id 为 None
|
||
pass
|
||
|
||
# 班级在读/毕业判断
|
||
has_active = class_has_active_student(connection, class_code)
|
||
class_status = '1' if has_active else '3'
|
||
is_graduation = 0 if has_active else 1
|
||
|
||
# 计算毕业年份
|
||
try:
|
||
enrollment_year_int = int(enrollment_year) if enrollment_year else None
|
||
except Exception:
|
||
enrollment_year_int = None
|
||
graduation_year = None
|
||
if enrollment_year_int is not None and school_system is not None:
|
||
try:
|
||
graduation_year = enrollment_year_int + school_system
|
||
except Exception:
|
||
graduation_year = None
|
||
|
||
# 尝试解析班主任 user_id，并填充到 class_msg.class_teacher_id
|
||
class_teacher_id = resolve_teacher_user(connection, teacher_name)
|
||
|
||
class_msg = {
|
||
"id": dept_id,
|
||
"tenant_id": "000000",
|
||
"create_time": current_time,
|
||
"update_time": current_time,
|
||
"college_id": college_id,
|
||
"college_name": college_name,
|
||
"major_id": major_id,
|
||
"major_name": major_name,
|
||
"class_code": class_code,
|
||
"class_name": class_name,
|
||
"class_type": class_type,
|
||
"grade": grade,
|
||
"enrollment_year": enrollment_year_int,
|
||
"graduation_year": graduation_year,
|
||
"training_type": training_type,
|
||
"class_teacher_id": class_teacher_id,
|
||
"class_teacher_name": teacher_name,
|
||
"class_status": class_status,
|
||
"is_graduation": is_graduation,
|
||
"graduation_date": None,
|
||
"school_system": school_system
|
||
}
|
||
|
||
insert_xg_class_msg(cursor, class_msg)
|
||
class_msg_inserted += 1
|
||
print(f"插入 xg_class_msg: id={dept_id}, class_code={class_code}, class_name={class_name}")
|
||
|
||
# ========== 插入班主任关系表（如能唯一解析到 teacher_id） ==========
|
||
teacher_id = resolve_teacher_user(connection, teacher_name)
|
||
if teacher_id:
|
||
seq_counter += 1
|
||
rel_id = generate_id(base_timestamp, seq_counter)
|
||
# 处理 start_date：优先使用 enrollment_date 的解析结果；若为空且有 enrollment_year，则使用该年 9 月 1 日
|
||
parsed_enroll_date = parse_date_to_mysql(row.get("enrollment_date"))
|
||
start_date_value = parsed_enroll_date
|
||
try:
|
||
if not start_date_value:
|
||
# enrollment_year_int 在前面已计算
|
||
if enrollment_year_int:
|
||
start_date_value = f"{enrollment_year_int}-09-01 00:00:00"
|
||
except Exception:
|
||
start_date_value = parsed_enroll_date
|
||
|
||
rel = {
|
||
"id": rel_id,
|
||
"tenant_id": "000000",
|
||
"create_time": current_time,
|
||
"update_time": current_time,
|
||
"class_id": dept_id,
|
||
"class_name": class_name,
|
||
"teacher_id": teacher_id,
|
||
"teacher_name": teacher_name,
|
||
"teacher_type": "class_teacher",
|
||
"start_date": start_date_value,
|
||
"end_date": None,
|
||
"remark": None
|
||
}
|
||
insert_teacher_rel(cursor, rel)
|
||
teacher_rel_inserted += 1
|
||
print(f"插入班主任关系: rel_id={rel_id}, class_id={dept_id}, teacher_id={teacher_id}")
|
||
else:
|
||
if teacher_name:
|
||
print(f"无法唯一解析班主任 '{teacher_name}' 到 sys_user.user_id，跳过关系插入")
|
||
|
||
# 提交事务
|
||
connection.commit()
|
||
print("✓ 事务已提交")
|
||
|
||
# 结束统计
|
||
print("=" * 60)
|
||
print(f"sys_dept 插入: {inserted_count}")
|
||
print(f"xg_class_msg 插入: {class_msg_inserted}")
|
||
print(f"xg_class_teacher_counselor_rel 插入: {teacher_rel_inserted}")
|
||
print("=" * 60)
|
||
|
||
except Exception as e:
|
||
print("✗ 处理过程中发生异常:", e)
|
||
finally:
|
||
connection.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|
||
|