112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
"""导入《软件开发部人员名单.xlsx》→ 人力资源库（staff 表）。
|
||
|
||
用法：
|
||
python scripts/import_staff.py # 用默认路径
|
||
python scripts/import_staff.py <xlsx> # 指定文件
|
||
python scripts/import_staff.py --reset # 清空后重新导入
|
||
python scripts/import_staff.py --dry-run # 只看解析结果，不落库
|
||
|
||
表头按名称匹配：序号 / 姓名 / 归属小组 / 人员归属 / 说明。
|
||
同名人员默认跳过（不动已有记录，避免覆盖页面上的后续编辑）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||
|
||
from app import constants # noqa: E402
|
||
from app.db import Base, engine, SessionLocal # noqa: E402
|
||
from app.models import Staff # noqa: E402
|
||
from app.schema_patch import ensure_columns # noqa: E402
|
||
|
||
DEFAULT_XLSX = "/Users/jiliu/WorkSpace/unissense/部门管理/软件开发部人员名单.xlsx"
|
||
COLUMNS = {"序号": "sort_order", "姓名": "name", "归属小组": "group", "人员归属": "affiliation", "说明": "note"}
|
||
|
||
|
||
def read_rows(path: Path) -> list[dict]:
|
||
import openpyxl
|
||
|
||
wb = openpyxl.load_workbook(path, data_only=True)
|
||
ws = wb.worksheets[0]
|
||
rows = list(ws.iter_rows(values_only=True))
|
||
|
||
head_idx, head = None, {}
|
||
for i, row in enumerate(rows):
|
||
cells = [str(c).strip() if c is not None else "" for c in row]
|
||
if "姓名" in cells:
|
||
head_idx = i
|
||
head = {name: j for j, name in enumerate(cells) if name}
|
||
break
|
||
if head_idx is None:
|
||
raise SystemExit(f"没在 {path} 里找到含「姓名」的表头行")
|
||
|
||
items = []
|
||
for row in rows[head_idx + 1:]:
|
||
rec = {}
|
||
for cn, field in COLUMNS.items():
|
||
j = head.get(cn)
|
||
rec[field] = None if j is None or j >= len(row) else row[j]
|
||
name = str(rec.get("name") or "").strip()
|
||
if not name:
|
||
continue
|
||
try:
|
||
sort_order = int(rec.get("sort_order") or 0)
|
||
except (TypeError, ValueError):
|
||
sort_order = 0
|
||
items.append(
|
||
{
|
||
"name": name,
|
||
"group": constants.normalize_staff_group(str(rec.get("group") or "").strip()),
|
||
"affiliation": constants.normalize_staff_affiliation(str(rec.get("affiliation") or "").strip()),
|
||
"note": str(rec.get("note") or "").strip() or None,
|
||
"sort_order": sort_order,
|
||
}
|
||
)
|
||
return items
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("xlsx", nargs="?", default=DEFAULT_XLSX)
|
||
ap.add_argument("--reset", action="store_true", help="先清空 staff 表")
|
||
ap.add_argument("--dry-run", action="store_true", help="只解析不落库")
|
||
args = ap.parse_args()
|
||
|
||
path = Path(args.xlsx)
|
||
if not path.exists():
|
||
raise SystemExit(f"文件不存在：{path}")
|
||
|
||
Base.metadata.create_all(bind=engine)
|
||
ensure_columns(engine, Base.metadata)
|
||
items = read_rows(path)
|
||
print(f"[i] 解析到 {len(items)} 人：{path}")
|
||
for it in items:
|
||
print(f" {it['sort_order']:>3} {it['name']:<6} {it['group'] or '-':<10} {it['affiliation'] or '-':<4} {it['note'] or ''}")
|
||
if args.dry_run:
|
||
print("[i] dry-run，未写入")
|
||
return
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
if args.reset:
|
||
n = db.query(Staff).delete()
|
||
print(f"[i] 已清空原有人员 {n} 条")
|
||
added = skipped = 0
|
||
for it in items:
|
||
if db.query(Staff).filter(Staff.name == it["name"]).one_or_none():
|
||
skipped += 1
|
||
continue
|
||
db.add(Staff(**it))
|
||
added += 1
|
||
db.commit()
|
||
print(f"[ok] 新增 {added} 人，跳过重名 {skipped} 人，花名册共 {db.query(Staff).count()} 人")
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|