ee2a4d9e22
All checks were successful
技能自动化发布 / release (push) Successful in 15s
基于 skill-template 脚手架开发，包含： - 5 张业务表（教职工、请假政策、考勤日历、请假记录、假期余额） - 请假核算引擎（扣薪计算、审批层级判定、年假抵扣） - CLI 入口（import-employees / calculate-leave / render-html） - HTML 审批与扣薪报告生成 - 示例数据与完整测试套件（295 tests passed）
366 lines
13 KiB
Python
366 lines
13 KiB
Python
"""高校人事请假考勤：领域服务（核算引擎与报告生成）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import calendar
|
||
import html
|
||
from datetime import date, datetime
|
||
from typing import Any, Optional
|
||
|
||
from db import leave_repository as repo
|
||
from service.employee_import_service import IMPORT_SUBDIR, ensure_builtin_samples
|
||
from service.html_page_service import (
|
||
HOST_HTML_RENDER_ACTION_ID,
|
||
load_html_template,
|
||
render_html_placeholders,
|
||
write_html_page,
|
||
)
|
||
from util.logging_config import get_skill_logger
|
||
from util.timeutil import now_unix
|
||
|
||
|
||
def _log():
|
||
try:
|
||
return get_skill_logger()
|
||
except RuntimeError:
|
||
return None
|
||
|
||
|
||
def ensure_ready() -> None:
|
||
ensure_builtin_samples(overwrite=False)
|
||
|
||
|
||
def _parse_date(text: str) -> date:
|
||
try:
|
||
return datetime.strptime(text.strip(), "%Y-%m-%d").date()
|
||
except (ValueError, AttributeError) as exc:
|
||
raise ValueError(f"日期格式无效：{text!r}（需 YYYY-MM-DD）") from exc
|
||
|
||
|
||
def _load_holiday_dates() -> set[str]:
|
||
holidays = repo.list_holidays()
|
||
return {h["holiday_date"] for h in holidays if h.get("holiday_date")}
|
||
|
||
|
||
def _count_working_days(start: date, end: date, holiday_dates: set[str]) -> int:
|
||
count = 0
|
||
cur = start
|
||
while cur <= end:
|
||
iso = cur.isoformat()
|
||
if cur.weekday() < 5 and iso not in holiday_dates:
|
||
count += 1
|
||
cur = date.fromordinal(cur.toordinal() + 1)
|
||
return count
|
||
|
||
|
||
def _working_days_in_month(d: date, holiday_dates: set[str]) -> int:
|
||
_, last_day = calendar.monthrange(d.year, d.month)
|
||
count = 0
|
||
for day in range(1, last_day + 1):
|
||
cur = date(d.year, d.month, day)
|
||
iso = cur.isoformat()
|
||
if cur.weekday() < 5 and iso not in holiday_dates:
|
||
count += 1
|
||
return count
|
||
|
||
|
||
def _determine_approval_level(
|
||
working_days: int, policy: dict[str, Any], leave_type: str
|
||
) -> str:
|
||
if leave_type in ("产假", "育假"):
|
||
return "系主任+人事处+校领导"
|
||
threshold_dept = int(policy.get("approval_threshold_dept", 3))
|
||
threshold_hr = int(policy.get("approval_threshold_hr", 10))
|
||
if working_days <= threshold_dept:
|
||
return "系主任"
|
||
if working_days <= threshold_hr:
|
||
return "系主任+人事处"
|
||
return "系主任+人事处+校领导"
|
||
|
||
|
||
def calculate_leave(
|
||
employee_id: int,
|
||
*,
|
||
leave_type: str,
|
||
start_date: str,
|
||
end_date: str,
|
||
has_certificate: bool = False,
|
||
certificate_type: Optional[str] = None,
|
||
) -> dict[str, Any]:
|
||
ensure_ready()
|
||
employee = repo.get_employee(int(employee_id))
|
||
if not employee:
|
||
raise ValueError(f"找不到教职工 id={employee_id}。请先在「教职工」表中新增或导入。")
|
||
|
||
policy = repo.get_policy(leave_type)
|
||
if not policy:
|
||
raise ValueError(
|
||
f"未知请假类型：{leave_type}。请在「请假政策」表中配置该类型后重试。"
|
||
)
|
||
|
||
if int(policy.get("requires_certificate", 0)) and not has_certificate:
|
||
raise ValueError(f"请假类型「{leave_type}」需附证明文件，请提供后重试。")
|
||
|
||
start = _parse_date(start_date)
|
||
end = _parse_date(end_date)
|
||
if end < start:
|
||
raise ValueError("结束日期不能早于开始日期。")
|
||
|
||
natural_days = (end - start).days + 1
|
||
holiday_dates = _load_holiday_dates()
|
||
working_days = _count_working_days(start, end, holiday_dates)
|
||
|
||
day_basis = str(policy.get("day_basis", "workday"))
|
||
billable_days = natural_days if day_basis == "natural" else working_days
|
||
|
||
monthly_salary = float(employee.get("monthly_salary", 0) or 0)
|
||
working_days_in_month = _working_days_in_month(start, holiday_dates)
|
||
if working_days_in_month <= 0:
|
||
working_days_in_month = 21.75
|
||
daily_salary = monthly_salary / working_days_in_month if working_days_in_month else 0.0
|
||
|
||
paid_days = float(policy.get("paid_days", 0) or 0)
|
||
partial_ratio = float(policy.get("partial_pay_ratio", 1.0) or 1.0)
|
||
unpaid_after = policy.get("unpaid_after_days")
|
||
unpaid_after_days = float(unpaid_after) if unpaid_after is not None else None
|
||
|
||
if billable_days <= paid_days:
|
||
deduction_amount = 0.0
|
||
elif unpaid_after_days is not None and billable_days > unpaid_after_days:
|
||
partial_days = max(unpaid_after_days - paid_days, 0)
|
||
unpaid_days = billable_days - unpaid_after_days
|
||
deduction_amount = (
|
||
daily_salary * partial_days * (1 - partial_ratio)
|
||
+ daily_salary * unpaid_days
|
||
)
|
||
else:
|
||
partial_days = billable_days - paid_days
|
||
deduction_amount = daily_salary * partial_days * (1 - partial_ratio)
|
||
|
||
annual_leave_used = 0.0
|
||
remaining_annual_leave = float(employee.get("annual_leave_remaining", 0) or 0)
|
||
if int(policy.get("can_offset_annual", 0)) and remaining_annual_leave > 0:
|
||
offset_days = min(billable_days, remaining_annual_leave)
|
||
deduction_amount = max(deduction_amount - offset_days * daily_salary, 0.0)
|
||
annual_leave_used = offset_days
|
||
remaining_annual_leave = remaining_annual_leave - offset_days
|
||
|
||
approval_level = _determine_approval_level(working_days, policy, leave_type)
|
||
|
||
params = {
|
||
"employee_id": employee_id,
|
||
"leave_type": leave_type,
|
||
"start_date": start_date,
|
||
"end_date": end_date,
|
||
"has_certificate": has_certificate,
|
||
"certificate_type": certificate_type,
|
||
"policy_id": policy.get("id"),
|
||
"day_basis": day_basis,
|
||
}
|
||
|
||
request_id = repo.create_leave_request(
|
||
{
|
||
"employee_id": employee_id,
|
||
"leave_type": leave_type,
|
||
"start_date": start_date,
|
||
"end_date": end_date,
|
||
"has_certificate": 1 if has_certificate else 0,
|
||
"certificate_type": certificate_type,
|
||
"natural_days": natural_days,
|
||
"working_days": working_days,
|
||
"monthly_salary": monthly_salary,
|
||
"working_days_in_month": working_days_in_month,
|
||
"daily_salary": round(daily_salary, 2),
|
||
"deduction_amount": round(deduction_amount, 2),
|
||
"annual_leave_used": annual_leave_used,
|
||
"remaining_annual_leave": remaining_annual_leave,
|
||
"approval_level": approval_level,
|
||
"status": "draft",
|
||
"params": params,
|
||
"summary": f"{employee.get('name')}「{leave_type}」{start_date}~{end_date}（{working_days}工作日）核算中…",
|
||
}
|
||
)
|
||
|
||
try:
|
||
summary = (
|
||
f"{employee.get('name')}「{leave_type}」{start_date}~{end_date}"
|
||
f"（{natural_days}自然日/{working_days}工作日）"
|
||
f"应扣 {round(deduction_amount, 2)} 元，审批：{approval_level}"
|
||
)
|
||
repo.update_leave_request(
|
||
request_id,
|
||
status="completed",
|
||
summary=summary,
|
||
)
|
||
if annual_leave_used > 0:
|
||
repo.update_employee_annual_leave(employee_id, remaining_annual_leave)
|
||
balance_type = _map_balance_type(leave_type)
|
||
existing = repo.get_balance(employee_id, balance_type)
|
||
used = float(existing["used_days"]) if existing else 0.0
|
||
repo.upsert_balance(
|
||
{
|
||
"employee_id": employee_id,
|
||
"balance_type": balance_type,
|
||
"total_days": policy.get("max_days"),
|
||
"used_days": used + billable_days,
|
||
"remaining_days": (
|
||
float(policy.get("max_days", 0) or 0) - used - billable_days
|
||
if policy.get("max_days") is not None
|
||
else None
|
||
),
|
||
}
|
||
)
|
||
except Exception as exc:
|
||
repo.update_leave_request(
|
||
request_id,
|
||
status="failed",
|
||
error_message=str(exc),
|
||
)
|
||
raise
|
||
|
||
payload = {
|
||
"ok": True,
|
||
"request_id": request_id,
|
||
"employee_id": employee_id,
|
||
"employee_name": employee.get("name"),
|
||
"department": employee.get("department"),
|
||
"leave_type": leave_type,
|
||
"start_date": start_date,
|
||
"end_date": end_date,
|
||
"natural_days": natural_days,
|
||
"working_days": working_days,
|
||
"monthly_salary": monthly_salary,
|
||
"working_days_in_month": working_days_in_month,
|
||
"daily_salary": round(daily_salary, 2),
|
||
"deduction_amount": round(deduction_amount, 2),
|
||
"annual_leave_used": annual_leave_used,
|
||
"remaining_annual_leave": remaining_annual_leave,
|
||
"approval_level": approval_level,
|
||
"certificate_type": certificate_type,
|
||
"message": summary,
|
||
}
|
||
log = _log()
|
||
if log:
|
||
log.info(
|
||
"leave_calculated request_id=%s employee_id=%s deduction=%s approval=%s",
|
||
request_id,
|
||
employee_id,
|
||
round(deduction_amount, 2),
|
||
approval_level,
|
||
)
|
||
return payload
|
||
|
||
|
||
def _map_balance_type(leave_type: str) -> str:
|
||
mapping = {
|
||
"病假": "sick",
|
||
"长期病假": "sick",
|
||
"事假": "annual",
|
||
"婚假": "marriage",
|
||
"产假": "maternity",
|
||
"育假": "maternity",
|
||
"丧假": "bereavement",
|
||
}
|
||
return mapping.get(leave_type, "annual")
|
||
|
||
|
||
def build_leave_report_html(request_id: int) -> str:
|
||
req = repo.get_request(int(request_id))
|
||
if not req:
|
||
raise ValueError(f"找不到请假记录 id={request_id}。")
|
||
employee = repo.get_employee(int(req["employee_id"])) if req.get("employee_id") else None
|
||
name = html.escape(str(employee.get("name", "")) if employee else "")
|
||
department = html.escape(str(employee.get("department", "")) if employee else "")
|
||
position = html.escape(str(employee.get("position", "")) if employee else "")
|
||
|
||
rows = [
|
||
("申请人", f"{name}（{department}）"),
|
||
("岗位", position or "—"),
|
||
("请假类型", html.escape(str(req.get("leave_type", "")))),
|
||
("请假日期", f"{req.get('start_date', '')} 至 {req.get('end_date', '')}（共 {req.get('natural_days', 0)} 个自然日）"),
|
||
("实际工作日", f"{req.get('working_days', 0)} 天"),
|
||
("当月应出勤", f"{req.get('working_days_in_month', 0)} 天"),
|
||
("日薪", f"{req.get('daily_salary', 0)} 元"),
|
||
("应扣工资", f"{req.get('deduction_amount', 0)} 元"),
|
||
("已抵年假", f"{req.get('annual_leave_used', 0)} 天"),
|
||
("剩余年假", f"{req.get('remaining_annual_leave', 0)} 天"),
|
||
("审批层级", html.escape(str(req.get("approval_level", "")))),
|
||
("证件类型", html.escape(str(req.get("certificate_type") or "无"))),
|
||
("状态", html.escape(str(req.get("status", "")))),
|
||
]
|
||
rows_html = "".join(
|
||
f"<tr><th>{label}</th><td>{value}</td></tr>" for label, value in rows
|
||
)
|
||
|
||
summary = html.escape(str(req.get("summary", "")))
|
||
approval_hint = (
|
||
"若涉及产假/育假或超过 10 天，需校领导审批并报人事处存档。"
|
||
if "校领导" in str(req.get("approval_level", ""))
|
||
else "请将审批表与证明文件交至对应审批人。"
|
||
)
|
||
|
||
body = f"""
|
||
<section class="card">
|
||
<h2>请假审批与扣薪计算报告</h2>
|
||
<table class="meta-table">{rows_html}</table>
|
||
</section>
|
||
<section class="card">
|
||
<h2>审批流程</h2>
|
||
<p>{approval_hint}</p>
|
||
<p class="muted">已自动生成审批表，需附相关证件原件。</p>
|
||
</section>
|
||
<section class="card">
|
||
<h2>后续处理</h2>
|
||
<ul>
|
||
<li>考勤记录已更新：本次请假 {req.get('working_days', 0)} 个工作日。</li>
|
||
<li>假期余额已同步更新。</li>
|
||
<li>摘要：{summary}</li>
|
||
</ul>
|
||
</section>
|
||
"""
|
||
return body
|
||
|
||
|
||
def render_leave_report(request_id: Optional[int] = None) -> dict[str, Any]:
|
||
ensure_ready()
|
||
rid: Optional[int] = None
|
||
if request_id is not None and str(request_id).strip():
|
||
text = str(request_id).strip()
|
||
if "{{" in text or "}}" in text:
|
||
text = ""
|
||
if text:
|
||
try:
|
||
rid = int(text)
|
||
except ValueError as exc:
|
||
raise ValueError(f"参数 request_id 必须是整数，当前为 {text!r}") from exc
|
||
if rid is None:
|
||
latest = repo.get_latest_completed_request()
|
||
if not latest:
|
||
raise ValueError("暂无已完成的请假核算记录，请先执行「请假核算」。")
|
||
rid = int(latest["id"])
|
||
|
||
body = build_leave_report_html(rid)
|
||
template = load_html_template("assets/html/leave-report.html")
|
||
page = render_html_placeholders(
|
||
template,
|
||
{
|
||
"TITLE": f"请假审批与扣薪报告 #{rid}",
|
||
"GENERATED_AT": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"BODY": body,
|
||
},
|
||
)
|
||
filename = f"leave-report-{rid}-{now_unix()}.html"
|
||
path = write_html_page(filename, page)
|
||
payload = {
|
||
"ok": True,
|
||
"request_id": rid,
|
||
"html_path": path,
|
||
"filename": filename,
|
||
"action_id": HOST_HTML_RENDER_ACTION_ID,
|
||
"message": f"已生成请假报告：{filename}",
|
||
}
|
||
log = _log()
|
||
if log:
|
||
log.info("leave_report_rendered request_id=%s path=%s", rid, path)
|
||
return payload
|