feat: 初始化高校人事请假考勤技能 (process-leave-attendance-default)
All checks were successful
技能自动化发布 / release (push) Successful in 15s

基于 skill-template 脚手架开发,包含:
- 5 张业务表(教职工、请假政策、考勤日历、请假记录、假期余额)
- 请假核算引擎(扣薪计算、审批层级判定、年假抵扣)
- CLI 入口(import-employees / calculate-leave / render-html)
- HTML 审批与扣薪报告生成
- 示例数据与完整测试套件(295 tests passed)
This commit is contained in:
chenfuqiang
2026-09-07 16:59:24 +08:00
commit ee2a4d9e22
206 changed files with 29510 additions and 0 deletions

View File

@@ -0,0 +1,179 @@
"""教职工导入:从 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