all-project/学校历史数据迁移/06-insert_classes.py

#!/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()