ee2a4d9e22
All checks were successful
技能自动化发布 / release (push) Successful in 15s
基于 skill-template 脚手架开发，包含： - 5 张业务表（教职工、请假政策、考勤日历、请假记录、假期余额） - 请假核算引擎（扣薪计算、审批层级判定、年假抵扣） - CLI 入口（import-employees / calculate-leave / render-html） - HTML 审批与扣薪报告生成 - 示例数据与完整测试套件（295 tests passed）
180 lines
5.8 KiB
Python
180 lines
5.8 KiB
Python
"""教职工导入：从 input/教职工库 读取 JSON，并可选播种示例政策与假日。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
from typing import Any
|
||
|
||
from db import leave_repository as repo
|
||
from util.logging_config import get_skill_logger
|
||
from util.runtime_paths import get_input_dir, get_skill_root, resolve_input_path
|
||
|
||
IMPORT_SUBDIR = "教职工库"
|
||
|
||
|
||
def _log():
|
||
try:
|
||
return get_skill_logger()
|
||
except RuntimeError:
|
||
return None
|
||
|
||
|
||
def get_builtin_samples_dir() -> str:
|
||
return os.path.join(get_skill_root(), "assets", "samples")
|
||
|
||
|
||
def get_employees_input_dir(*, create: bool = True) -> str:
|
||
path = os.path.join(get_input_dir(create=create), IMPORT_SUBDIR)
|
||
if create:
|
||
os.makedirs(path, exist_ok=True)
|
||
return path
|
||
|
||
|
||
def ensure_builtin_samples(*, overwrite: bool = False) -> list[str]:
|
||
"""将 assets/samples/示例教职工.json 复制到 input/教职工库（若不存在）。"""
|
||
src_dir = get_builtin_samples_dir()
|
||
dst_dir = get_employees_input_dir(create=True)
|
||
copied: list[str] = []
|
||
employee_src = os.path.join(src_dir, "示例教职工.json")
|
||
employee_dst = os.path.join(dst_dir, "示例教职工.json")
|
||
if os.path.isfile(employee_src) and (overwrite or not os.path.isfile(employee_dst)):
|
||
shutil.copy2(employee_src, employee_dst)
|
||
copied.append("示例教职工.json")
|
||
|
||
seed_sample_policies_if_empty()
|
||
seed_sample_holidays_if_empty()
|
||
|
||
log = _log()
|
||
if log and copied:
|
||
log.info("employee_samples_synced count=%s", len(copied))
|
||
return copied
|
||
|
||
|
||
def seed_sample_policies_if_empty() -> int:
|
||
if repo.count_policies() > 0:
|
||
return 0
|
||
path = os.path.join(get_builtin_samples_dir(), "示例政策.json")
|
||
if not os.path.isfile(path):
|
||
return 0
|
||
with open(path, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
count = 0
|
||
for policy in data.get("policies") or []:
|
||
repo.upsert_policy(policy)
|
||
count += 1
|
||
log = _log()
|
||
if log:
|
||
log.info("sample_policies_seeded count=%s", count)
|
||
return count
|
||
|
||
|
||
def seed_sample_holidays_if_empty() -> int:
|
||
if repo.count_holidays() > 0:
|
||
return 0
|
||
path = os.path.join(get_builtin_samples_dir(), "示例政策.json")
|
||
if not os.path.isfile(path):
|
||
return 0
|
||
with open(path, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
count = 0
|
||
for holiday in data.get("holidays") or []:
|
||
repo.upsert_holiday(holiday)
|
||
count += 1
|
||
log = _log()
|
||
if log:
|
||
log.info("sample_holidays_seeded count=%s", count)
|
||
return count
|
||
|
||
|
||
def resolve_employees_file(raw: str) -> str:
|
||
text = (raw or "").strip().strip('"').strip("'")
|
||
if not text:
|
||
raise ValueError("请选择教职工文件（input/教职工库/ 下的 .json）。")
|
||
if "{{" in text or "}}" in text:
|
||
raise ValueError("未选择有效文件。请选择 input/教职工库/ 目录下的 JSON。")
|
||
if os.path.isabs(text) and os.path.isfile(text):
|
||
return os.path.abspath(text)
|
||
candidate = resolve_input_path(text, create_parent=False)
|
||
if os.path.isfile(candidate):
|
||
return candidate
|
||
packs = get_employees_input_dir(create=True)
|
||
by_name = os.path.abspath(os.path.join(packs, os.path.basename(text)))
|
||
if os.path.isfile(by_name):
|
||
return by_name
|
||
under = os.path.abspath(os.path.join(packs, text.replace("\\", "/").lstrip("/")))
|
||
if os.path.isfile(under):
|
||
return under
|
||
raise ValueError(
|
||
f"找不到教职工文件：{text}。请放到「输入目录/{IMPORT_SUBDIR}/」后重新选择。"
|
||
)
|
||
|
||
|
||
def _normalize_employee(item: dict[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
"name": str(item.get("name") or "").strip(),
|
||
"employee_no": str(item.get("employee_no") or "").strip() or None,
|
||
"department": str(item.get("department") or "").strip() or None,
|
||
"position": str(item.get("position") or "").strip() or None,
|
||
"position_category": str(item.get("position_category") or "").strip() or None,
|
||
"join_date": str(item.get("join_date") or "").strip() or None,
|
||
"monthly_salary": float(item.get("monthly_salary", 0) or 0),
|
||
"annual_leave_remaining": float(item.get("annual_leave_remaining", 0) or 0),
|
||
"is_active": item.get("is_active", 1),
|
||
"notes": str(item.get("notes") or "").strip() or None,
|
||
}
|
||
|
||
|
||
def import_employees_json(file_path: str) -> dict[str, Any]:
|
||
path = resolve_employees_file(file_path)
|
||
with open(path, encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
if isinstance(data, dict):
|
||
items = data.get("employees") or data.get("items") or []
|
||
if not items and data.get("name"):
|
||
items = [data]
|
||
elif isinstance(data, list):
|
||
items = data
|
||
else:
|
||
raise ValueError("教职工 JSON 格式无效：应为数组或含 employees 字段的对象。")
|
||
|
||
created = 0
|
||
updated = 0
|
||
skipped = 0
|
||
for raw in items:
|
||
if not isinstance(raw, dict):
|
||
skipped += 1
|
||
continue
|
||
payload = _normalize_employee(raw)
|
||
if not payload["name"]:
|
||
skipped += 1
|
||
continue
|
||
before = repo.find_employee_by_name(payload["name"])
|
||
repo.upsert_employee(payload)
|
||
if before:
|
||
updated += 1
|
||
else:
|
||
created += 1
|
||
|
||
ensure_builtin_samples(overwrite=False)
|
||
result = {
|
||
"ok": True,
|
||
"file": path,
|
||
"created": created,
|
||
"updated": updated,
|
||
"skipped": skipped,
|
||
"total": created + updated,
|
||
"message": f"已导入教职工：新增 {created}，更新 {updated}，跳过 {skipped}。",
|
||
}
|
||
log = _log()
|
||
if log:
|
||
log.info(
|
||
"employees_imported created=%s updated=%s skipped=%s file=%s",
|
||
created,
|
||
updated,
|
||
skipped,
|
||
path,
|
||
)
|
||
return result
|