120 lines
4.0 KiB
Python
120 lines
4.0 KiB
Python
#!/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()
|
||
|
||
|