all-project/学校历史数据迁移/07-reorder_classes_by_grade.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
07 - 按年级重排序 sys_dept 中专业下的班级 (修改 order_num)

说明：
- 针对 sys_dept 中 dept_category 以 'zy-' 开头的专业，查找 parent_id 为该专业的子部门（dept_category LIKE 'bj-%'）
- 使用 xg_class_msg.enrollment_year 或 sys_dept.dept_name 进行排序（优先 enrollment_year，空值视为 0），然后按 dept_name
- 为每个父专业从 1 开始重新分配 order_num（可修改为起始值为当前最大值 + 1）
- 脚本执行前会交互确认（避免误删/误改）；支持 DRY_RUN 模式，仅打印将要更新的改动

使用方法：
- 运行前确认 `db_config.DB_CONFIG` 正确，执行： python 07-reorder_classes_by_grade.py
"""
from __future__ import annotations
from typing import List, Dict, Any, Optional, Tuple
from datetime import datetime
import pymysql
from db_config import DB_CONFIG

DRY_RUN = False


def get_connection():
    return pymysql.connect(cursorclass=pymysql.cursors.DictCursor, **DB_CONFIG)


def fetch_all_majors(conn) -> List[Dict[str, Any]]:
    sql = "SELECT dept_id, dept_name FROM sys_dept WHERE dept_category LIKE %s ORDER BY dept_name"
    with conn.cursor() as c:
        c.execute(sql, ("zy-%",))
        return c.fetchall()


def fetch_classes_under_major(conn, parent_id: int) -> List[Dict[str, Any]]:
    """
    返回列表包含 sys_dept.dept_id, dept_name, xg_class_msg.enrollment_year
    """
    sql = """
    SELECT sd.dept_id, sd.dept_name, xc.enrollment_year
    FROM sys_dept sd
    LEFT JOIN xg_class_msg xc ON sd.dept_id = xc.id
    WHERE sd.parent_id = %s AND sd.dept_category LIKE %s
    """
    with conn.cursor() as c:
        c.execute(sql, (parent_id, "bj-%"))
        return c.fetchall()


def normalize_year(val: Any) -> int:
    if val is None or val == "":
        return 0
    try:
        return int(val)
    except Exception:
        return 0


def update_order_nums(conn, updates: List[Tuple[int, int]]):
    """
    updates: list of (order_num, dept_id)
    """
    with conn.cursor() as c:
        for order_num, dept_id in updates:
            if DRY_RUN:
                print(f"[DRY_RUN] UPDATE sys_dept SET order_num={order_num} WHERE dept_id={dept_id}")
            else:
                c.execute("UPDATE sys_dept SET order_num = %s WHERE dept_id = %s", (order_num, dept_id))
        if not DRY_RUN:
            conn.commit()


def prompt_confirm() -> bool:
    try:
        ans = input("确认按年级重排序 sys_dept 中专业下的班级并修改 order_num 吗？[y/N]: ").strip().lower()
    except (EOFError, KeyboardInterrupt):
        print("\n未检测到交互输入，取消操作。")
        return False
    return ans in ("y", "yes")


def main():
    print("开始：07 - 按年级重排序 sys_dept 专业下的班级（修改 order_num）")
    if not prompt_confirm():
        print("已取消操作。")
        return

    conn = get_connection()
    try:
        majors = fetch_all_majors(conn)
        print(f"发现 {len(majors)} 个专业，将逐个处理...")
        total_changed = 0
        for m in majors:
            parent_id = m.get("dept_id")
            major_name = m.get("dept_name")
            classes = fetch_classes_under_major(conn, parent_id)
            if not classes:
                continue
            # sort by enrollment_year(int, 0) asc, then dept_name asc
            classes.sort(key=lambda r: (normalize_year(r.get("enrollment_year")), r.get("dept_name") or ""))
            updates: List[Tuple[int, int]] = []
            for idx, row in enumerate(classes, start=1):
                updates.append((idx, row.get("dept_id")))
            if updates:
                print(f"专业 {major_name}({parent_id}) 将更新 {len(updates)} 条 order_num。")
                update_order_nums(conn, updates)
                total_changed += len(updates)

        print(f"完成：共更新 order_num {total_changed} 条记录。")
    except Exception as e:
        print("发生异常:", e)
    finally:
        conn.close()


if __name__ == "__main__":
    main()