326 lines
13 KiB
Python
326 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
更新 sys_dept 表：
|
||
1) 打印 sys_dept 表结构供学习
|
||
2) 可删除 sys_dept 中除 dept_id 为 100/101/102 外的所有记录（需确认或使用 --yes）
|
||
3) 从 teacher.xlsx 中读取部门列，按规则插入到 sys_dept （部分学院放到 parent_id=101，其余放到 102）
|
||
|
||
用法：
|
||
python update_sys_dept.py # 交互式执行（会提示确认删除）
|
||
python update_sys_dept.py --yes # 自动确认删除
|
||
python update_sys_dept.py --teacher-file /path/to/teacher.xlsx
|
||
|
||
注意：
|
||
- 本脚本会尝试读取 `学校历史数据迁移/db_config.py` 中的 DB_CONFIG 连接数据库
|
||
- 插入时会尽量推断并填充常用字段（ancestors, parent_id, dept_name, order_num, status, del_flag, create_time）
|
||
"""
|
||
from __future__ import annotations
|
||
import os
|
||
import sys
|
||
import argparse
|
||
import datetime
|
||
from typing import List, Dict, Any, Optional
|
||
import pandas as pd
|
||
from sqlalchemy import create_engine, text
|
||
import importlib.util
|
||
|
||
|
||
def load_db_config_from_file(config_path: str) -> dict:
|
||
spec = importlib.util.spec_from_file_location("db_config", config_path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module) # type: ignore
|
||
if not hasattr(module, "DB_CONFIG"):
|
||
raise RuntimeError(f"No DB_CONFIG found in {config_path}")
|
||
return getattr(module, "DB_CONFIG")
|
||
|
||
|
||
def build_engine_from_config(db_config: dict):
|
||
user = db_config.get("user")
|
||
password = db_config.get("password")
|
||
host = db_config.get("host", "localhost")
|
||
port = db_config.get("port", 3306)
|
||
database = db_config.get("database")
|
||
charset = db_config.get("charset", "utf8mb4")
|
||
uri = f"mysql+pymysql://{user}:{password}@{host}:{port}/{database}?charset={charset}"
|
||
engine = create_engine(uri, pool_pre_ping=True)
|
||
return engine
|
||
|
||
|
||
def inspect_table(engine, table_name: str) -> List[Dict[str, Any]]:
|
||
q = text(
|
||
"SELECT COLUMN_NAME, IS_NULLABLE, COLUMN_DEFAULT, DATA_TYPE "
|
||
"FROM information_schema.columns "
|
||
"WHERE table_schema = :schema AND table_name = :table ORDER BY ORDINAL_POSITION"
|
||
)
|
||
schema = engine.url.database
|
||
with engine.connect() as conn:
|
||
result = conn.execute(q, {"schema": schema, "table": table_name})
|
||
# Use mappings() so each row is a mapping of column_name->value
|
||
rows = result.mappings().all()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
def find_dept_column(df: pd.DataFrame) -> Optional[str]:
|
||
candidates = []
|
||
for col in df.columns:
|
||
name = str(col).strip()
|
||
low = name.lower()
|
||
if any(k in low for k in ("部门", "学院", "系", "组织", "院", "department", "dept", "office", "unit")):
|
||
candidates.append(name)
|
||
if candidates:
|
||
return candidates[0]
|
||
# fallback: ask user by listing columns
|
||
return None
|
||
|
||
|
||
def parent_has_children(engine, parent_id: int) -> bool:
|
||
"""
|
||
检查 parent_id 下是否已有子部门记录（用于决定是否跳过该 parent 的批量插入）
|
||
"""
|
||
q = text("SELECT 1 FROM sys_dept WHERE parent_id = :pid LIMIT 1")
|
||
with engine.connect() as conn:
|
||
r = conn.execute(q, {"pid": parent_id}).fetchone()
|
||
return bool(r)
|
||
|
||
|
||
def read_teacher_departments_from_db(engine, table_name: str = "his_lssj") -> List[str]:
|
||
"""
|
||
从数据库表（默认 his_lssj）读取去重的部门名称（列名 '部门'）。
|
||
返回去重、排序后的部门名称列表。
|
||
"""
|
||
print(f"从数据库表 `{table_name}` 读取部门列（列名假定为 '部门'）")
|
||
q = text(f"SELECT DISTINCT `部门` AS dept_name FROM `{table_name}` WHERE `部门` IS NOT NULL AND TRIM(`部门`) != ''")
|
||
with engine.connect() as conn:
|
||
rows = conn.execute(q).mappings().all()
|
||
depts = sorted({str(r.get("dept_name")).strip() for r in rows if r.get("dept_name") is not None})
|
||
print(f"发现 {len(depts)} 个去重部门名（示例前20）:")
|
||
for item in depts[:20]:
|
||
print(" -", item)
|
||
return depts
|
||
|
||
|
||
def get_parent_info(engine, parent_ids: List[int]) -> Dict[int, Dict[str, Any]]:
|
||
q = text("SELECT * FROM sys_dept WHERE dept_id IN :ids")
|
||
with engine.connect() as conn:
|
||
result = conn.execute(q, {"ids": tuple(parent_ids)})
|
||
rows = result.mappings().all()
|
||
out: Dict[int, Dict[str, Any]] = {}
|
||
for r in rows:
|
||
d = dict(r)
|
||
out[int(d.get("dept_id"))] = d
|
||
return out
|
||
|
||
|
||
def delete_except(engine, keep_ids: List[int], yes: bool):
|
||
if not yes:
|
||
resp = input(f"将删除 sys_dept 表中除 dept_id IN {keep_ids} 外的所有记录，确认继续？(yes/no): ").strip().lower()
|
||
if resp not in ("y", "yes"):
|
||
print("取消操作。")
|
||
return False
|
||
q = text("DELETE FROM sys_dept WHERE dept_id NOT IN :keep")
|
||
with engine.begin() as conn:
|
||
conn.execute(q, {"keep": tuple(keep_ids)})
|
||
print("删除完成。")
|
||
return True
|
||
|
||
|
||
def insert_departments(engine, dept_names: List[str], parent_mapping: Dict[str, int], parent_info: Dict[int, Dict[str, Any]]):
|
||
"""
|
||
插入时遵循以下规则以保持与现有数据一致性：
|
||
- dept_id: 若表中已有较大数值 id（如 2025xxxxxxxxxxxx），则以 max(dept_id)+1 继续；否则以当前时间 `%Y%m%d%H%M%S` + 4位序号生成
|
||
- tenant_id: 写为 '000000'
|
||
- order_num: 在同一 parent_id 下取 max(order_num)+1
|
||
- ancestors: 从 parent_info 推断，若父项无 ancestors 则使用 "0,{parent_id}"
|
||
- 其他字段填充 create_by/update_by/create_dept 等默认值以匹配样例
|
||
"""
|
||
inserted = 0
|
||
skipped = 0
|
||
now_dt = datetime.datetime.now()
|
||
now_str = now_dt.strftime("%Y-%m-%d %H:%M:%S")
|
||
timestamp_prefix = now_dt.strftime("%Y%m%d%H%M%S")
|
||
|
||
# helper to generate unique dept_id
|
||
def generate_dept_id(conn, local_counter_start=1):
|
||
# get current max dept_id
|
||
r = conn.execute(text("SELECT MAX(dept_id) as m FROM sys_dept")).fetchone()
|
||
max_id = int(r[0]) if r and r[0] is not None else 0
|
||
if max_id > 10**14:
|
||
return max_id + 1
|
||
# else use timestamp-based id plus 4-digit counter
|
||
counter = local_counter_start
|
||
while True:
|
||
candidate = int(f"{timestamp_prefix}{counter:04d}")
|
||
exists = conn.execute(text("SELECT 1 FROM sys_dept WHERE dept_id = :did LIMIT 1"), {"did": candidate}).fetchone()
|
||
if not exists:
|
||
return candidate
|
||
counter += 1
|
||
|
||
# Cache per-parent decision to avoid repeated checks: 'skip' or 'insert'
|
||
parent_decision: Dict[int, str] = {}
|
||
with engine.begin() as conn:
|
||
# fetch column info once
|
||
col_info = inspect_table(engine, "sys_dept")
|
||
col_names = [c["COLUMN_NAME"] for c in col_info]
|
||
# prepare xy counter for parent_id = 101 based on existing dept_category values
|
||
xy_counter = 1
|
||
try:
|
||
rows = conn.execute(text("SELECT dept_category FROM sys_dept WHERE parent_id = :pid AND dept_category LIKE 'xy-%'"), {"pid": 101}).mappings().all()
|
||
nums = []
|
||
for r in rows:
|
||
val = r.get("dept_category")
|
||
if not val:
|
||
continue
|
||
parts = str(val).split("-")
|
||
if len(parts) >= 2:
|
||
try:
|
||
nums.append(int(parts[-1]))
|
||
except Exception:
|
||
continue
|
||
if nums:
|
||
xy_counter = max(nums) + 1
|
||
except Exception:
|
||
# ignore and keep default
|
||
xy_counter = 1
|
||
|
||
for name in dept_names:
|
||
pname = name.strip()
|
||
if not pname:
|
||
continue
|
||
parent_id = parent_mapping.get(pname, None)
|
||
if parent_id is None:
|
||
parent_id = 102
|
||
# If we've decided to skip inserting for this parent, continue
|
||
if parent_id in parent_decision and parent_decision[parent_id] == "skip":
|
||
skipped += 1
|
||
continue
|
||
|
||
# First time encountering this parent: if it already has children, skip inserting any under it
|
||
if parent_id not in parent_decision:
|
||
try:
|
||
has_children = conn.execute(text("SELECT 1 FROM sys_dept WHERE parent_id = :pid LIMIT 1"), {"pid": parent_id}).fetchone()
|
||
if has_children:
|
||
parent_decision[parent_id] = "skip"
|
||
logger = None # no logger in this module-level function; just mark skip
|
||
skipped += 1
|
||
continue
|
||
else:
|
||
parent_decision[parent_id] = "insert"
|
||
except Exception:
|
||
parent_decision[parent_id] = "insert"
|
||
# skip if same dept_name and parent exists
|
||
exists_q = text("SELECT count(1) as c FROM sys_dept WHERE dept_name = :name AND parent_id = :pid")
|
||
res = conn.execute(exists_q, {"name": pname, "pid": parent_id}).fetchone()
|
||
if res and int(res[0]) > 0:
|
||
skipped += 1
|
||
continue
|
||
|
||
parent = parent_info.get(parent_id, {})
|
||
parent_ancestors = parent.get("ancestors") if parent else None
|
||
if parent_ancestors:
|
||
ancestors = f"{parent_ancestors},{parent_id}"
|
||
else:
|
||
# ensure root '0' present
|
||
if parent_id == 0:
|
||
ancestors = "0"
|
||
else:
|
||
ancestors = f"0,{parent_id}"
|
||
|
||
# compute next order_num under this parent
|
||
r = conn.execute(text("SELECT COALESCE(MAX(order_num),0) as m FROM sys_dept WHERE parent_id = :pid"), {"pid": parent_id}).fetchone()
|
||
next_order = int(r[0]) + 1 if r and r[0] is not None else 1
|
||
|
||
# generate unique dept_id
|
||
new_dept_id = generate_dept_id(conn)
|
||
|
||
# prepare insert dict with default fields matching provided samples
|
||
# set dept_category to 'xy-序号' for entries under parent_id == 101
|
||
dept_category_val = None
|
||
if parent_id == 101:
|
||
dept_category_val = f"xy-{xy_counter}"
|
||
xy_counter += 1
|
||
|
||
insert_dict: Dict[str, Any] = {
|
||
"dept_id": new_dept_id,
|
||
"tenant_id": "000000",
|
||
"parent_id": parent_id,
|
||
"ancestors": ancestors,
|
||
"dept_name": pname,
|
||
"dept_category": dept_category_val,
|
||
"order_num": next_order,
|
||
"leader": None,
|
||
"phone": None,
|
||
"email": None,
|
||
"status": "0",
|
||
"del_flag": "0",
|
||
"create_dept": 0,
|
||
"create_by": 1,
|
||
"create_time": now_str,
|
||
"update_by": 1,
|
||
"update_time": now_str,
|
||
}
|
||
|
||
# only keep keys that exist in the actual table
|
||
use_cols = {k: v for k, v in insert_dict.items() if k in col_names}
|
||
if "dept_name" not in use_cols:
|
||
raise RuntimeError("目标表不包含 dept_name 列，无法继续")
|
||
|
||
cols_part = ", ".join(f"`{k}`" for k in use_cols.keys())
|
||
vals_part = ", ".join(f":{k}" for k in use_cols.keys())
|
||
ins_q = text(f"INSERT INTO sys_dept ({cols_part}) VALUES ({vals_part})")
|
||
conn.execute(ins_q, use_cols)
|
||
inserted += 1
|
||
|
||
print(f"插入完成：新增 {inserted} 条，跳过已存在 {skipped} 条。")
|
||
|
||
|
||
def main():
|
||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||
config_path = os.path.join(base_dir, "db_config.py")
|
||
if not os.path.exists(config_path):
|
||
print(f"找不到配置文件: {config_path}")
|
||
sys.exit(1)
|
||
db_config = load_db_config_from_file(config_path)
|
||
engine = build_engine_from_config(db_config)
|
||
|
||
# 1. 学习表结构
|
||
print("正在读取 sys_dept 表结构...")
|
||
cols = inspect_table(engine, "sys_dept")
|
||
for c in cols:
|
||
print(f" - {c['COLUMN_NAME']} type={c['DATA_TYPE']} nullable={c['IS_NULLABLE']} default={c['COLUMN_DEFAULT']}")
|
||
|
||
# 2. 删除除指定 dept_id 外的记录
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--yes", action="store_true", help="自动确认删除")
|
||
parser.add_argument("--teacher-file", type=str, default=os.path.join(base_dir, "teacher.xlsx"))
|
||
args = parser.parse_args()
|
||
|
||
keep_ids = [100, 101, 102]
|
||
ok = delete_except(engine, keep_ids, args.yes)
|
||
if not ok:
|
||
print("用户取消，退出。")
|
||
return
|
||
|
||
# 3. 从 his_lssj 表导入部门（更快）
|
||
dept_names = read_teacher_departments_from_db(engine, table_name="his_lssj")
|
||
|
||
# mapping: specific colleges -> parent 101, others -> 102
|
||
top_level_101 = {
|
||
"人文学院", "餐饮休闲学院", "生态园林学院", "数字信息学院",
|
||
"医药健康学院", "智能控制学院", "智能制造学院"
|
||
}
|
||
# build mapping for exact names
|
||
parent_mapping = {name: 101 for name in top_level_101}
|
||
|
||
# get parent info
|
||
parent_info = get_parent_info(engine, [101, 102])
|
||
if 101 not in parent_info or 102 not in parent_info:
|
||
print("警告：未能在 sys_dept 中找到 parent_id 为 101 或 102 的记录，请确认这两个父组织是否存在。")
|
||
insert_departments(engine, dept_names, parent_mapping, parent_info)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|
||
|