ee2a4d9e22
All checks were successful
技能自动化发布 / release (push) Successful in 15s
基于 skill-template 脚手架开发，包含： - 5 张业务表（教职工、请假政策、考勤日历、请假记录、假期余额） - 请假核算引擎（扣薪计算、审批层级判定、年假抵扣） - CLI 入口（import-employees / calculate-leave / render-html） - HTML 审批与扣薪报告生成 - 示例数据与完整测试套件（295 tests passed）
568 lines
18 KiB
Python
568 lines
18 KiB
Python
"""高校人事请假考勤：业务数据访问。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from typing import Any, Optional
|
|
|
|
from db.connection import get_conn, init_db
|
|
from db.schema import (
|
|
ATTENDANCE_HOLIDAYS_TABLE,
|
|
EMPLOYEES_TABLE,
|
|
LEAVE_BALANCES_TABLE,
|
|
LEAVE_POLICIES_TABLE,
|
|
LEAVE_REQUESTS_TABLE,
|
|
)
|
|
|
|
|
|
def ensure_db() -> None:
|
|
init_db()
|
|
|
|
|
|
def _row_dict(row: sqlite3.Row) -> dict[str, Any]:
|
|
return {k: row[k] for k in row.keys()}
|
|
|
|
|
|
def dumps_json(data: Any) -> str:
|
|
return json.dumps(data, ensure_ascii=False)
|
|
|
|
|
|
def loads_json(raw: Any, default: Any = None) -> Any:
|
|
if raw is None or raw == "":
|
|
return default if default is not None else {}
|
|
if isinstance(raw, (dict, list)):
|
|
return raw
|
|
try:
|
|
return json.loads(str(raw))
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
return default if default is not None else {}
|
|
|
|
|
|
# ---------- employees ----------
|
|
|
|
def get_employee(employee_id: int) -> Optional[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(f"SELECT * FROM {EMPLOYEES_TABLE} WHERE id = ?", (employee_id,))
|
|
row = cur.fetchone()
|
|
return _row_dict(row) if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def find_employee_by_name(name: str) -> Optional[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(f"SELECT * FROM {EMPLOYEES_TABLE} WHERE name = ?", (name,))
|
|
row = cur.fetchone()
|
|
return _row_dict(row) if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def list_active_employees() -> list[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(f"SELECT * FROM {EMPLOYEES_TABLE} WHERE is_active = 1 ORDER BY id")
|
|
return [_row_dict(r) for r in cur.fetchall()]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def count_employees() -> int:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(f"SELECT COUNT(*) FROM {EMPLOYEES_TABLE}")
|
|
return int(cur.fetchone()[0])
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def upsert_employee(payload: dict[str, Any]) -> int:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
name = str(payload.get("name", "")).strip()
|
|
if not name:
|
|
raise ValueError("教职工姓名不能为空")
|
|
employee_no = payload.get("employee_no")
|
|
existing = None
|
|
if employee_no:
|
|
cur.execute(
|
|
f"SELECT id FROM {EMPLOYEES_TABLE} WHERE employee_no = ?",
|
|
(employee_no,),
|
|
)
|
|
existing = cur.fetchone()
|
|
if existing is None:
|
|
cur.execute(
|
|
f"SELECT id FROM {EMPLOYEES_TABLE} WHERE name = ? AND "
|
|
"(employee_no IS NULL OR employee_no = '')",
|
|
(name,),
|
|
)
|
|
existing = cur.fetchone()
|
|
if existing is not None:
|
|
eid = int(existing[0])
|
|
cur.execute(
|
|
f"""UPDATE {EMPLOYEES_TABLE} SET
|
|
name = ?, employee_no = ?, department = ?, position = ?,
|
|
position_category = ?, join_date = ?, monthly_salary = ?,
|
|
annual_leave_remaining = ?, is_active = ?, notes = ?,
|
|
updated_at = unixepoch()
|
|
WHERE id = ?""",
|
|
(
|
|
name,
|
|
employee_no,
|
|
payload.get("department"),
|
|
payload.get("position"),
|
|
payload.get("position_category"),
|
|
payload.get("join_date"),
|
|
float(payload.get("monthly_salary", 0) or 0),
|
|
float(payload.get("annual_leave_remaining", 0) or 0),
|
|
int(payload.get("is_active", 1)),
|
|
payload.get("notes"),
|
|
eid,
|
|
),
|
|
)
|
|
conn.commit()
|
|
return eid
|
|
cur.execute(
|
|
f"""INSERT INTO {EMPLOYEES_TABLE} (
|
|
name, employee_no, department, position, position_category,
|
|
join_date, monthly_salary, annual_leave_remaining, is_active, notes
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
name,
|
|
employee_no,
|
|
payload.get("department"),
|
|
payload.get("position"),
|
|
payload.get("position_category"),
|
|
payload.get("join_date"),
|
|
float(payload.get("monthly_salary", 0) or 0),
|
|
float(payload.get("annual_leave_remaining", 0) or 0),
|
|
int(payload.get("is_active", 1)),
|
|
payload.get("notes"),
|
|
),
|
|
)
|
|
conn.commit()
|
|
return int(cur.lastrowid)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def update_employee_annual_leave(employee_id: int, remaining: float) -> None:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"UPDATE {EMPLOYEES_TABLE} SET annual_leave_remaining = ?, "
|
|
"updated_at = unixepoch() WHERE id = ?",
|
|
(remaining, employee_id),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------- leave_policies ----------
|
|
|
|
def get_policy(leave_type: str) -> Optional[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"SELECT * FROM {LEAVE_POLICIES_TABLE} WHERE leave_type = ?",
|
|
(leave_type,),
|
|
)
|
|
row = cur.fetchone()
|
|
return _row_dict(row) if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def list_policies() -> list[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(f"SELECT * FROM {LEAVE_POLICIES_TABLE} ORDER BY id")
|
|
return [_row_dict(r) for r in cur.fetchall()]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def count_policies() -> int:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(f"SELECT COUNT(*) FROM {LEAVE_POLICIES_TABLE}")
|
|
return int(cur.fetchone()[0])
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def upsert_policy(payload: dict[str, Any]) -> int:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
leave_type = str(payload.get("leave_type", "")).strip()
|
|
if not leave_type:
|
|
raise ValueError("请假类型不能为空")
|
|
cur.execute(
|
|
f"SELECT id FROM {LEAVE_POLICIES_TABLE} WHERE leave_type = ?",
|
|
(leave_type,),
|
|
)
|
|
existing = cur.fetchone()
|
|
if existing is not None:
|
|
pid = int(existing[0])
|
|
cur.execute(
|
|
f"""UPDATE {LEAVE_POLICIES_TABLE} SET
|
|
leave_type = ?, max_days = ?, day_basis = ?, paid_days = ?,
|
|
partial_pay_ratio = ?, unpaid_after_days = ?,
|
|
requires_certificate = ?, can_offset_annual = ?,
|
|
approval_threshold_dept = ?, approval_threshold_hr = ?,
|
|
notes = ?, updated_at = unixepoch()
|
|
WHERE id = ?""",
|
|
(
|
|
leave_type,
|
|
payload.get("max_days"),
|
|
payload.get("day_basis", "workday"),
|
|
float(payload.get("paid_days", 0) or 0),
|
|
float(payload.get("partial_pay_ratio", 1.0) or 1.0),
|
|
payload.get("unpaid_after_days"),
|
|
int(payload.get("requires_certificate", 0)),
|
|
int(payload.get("can_offset_annual", 0)),
|
|
int(payload.get("approval_threshold_dept", 3)),
|
|
int(payload.get("approval_threshold_hr", 10)),
|
|
payload.get("notes"),
|
|
pid,
|
|
),
|
|
)
|
|
conn.commit()
|
|
return pid
|
|
cur.execute(
|
|
f"""INSERT INTO {LEAVE_POLICIES_TABLE} (
|
|
leave_type, max_days, day_basis, paid_days, partial_pay_ratio,
|
|
unpaid_after_days, requires_certificate, can_offset_annual,
|
|
approval_threshold_dept, approval_threshold_hr, notes
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
leave_type,
|
|
payload.get("max_days"),
|
|
payload.get("day_basis", "workday"),
|
|
float(payload.get("paid_days", 0) or 0),
|
|
float(payload.get("partial_pay_ratio", 1.0) or 1.0),
|
|
payload.get("unpaid_after_days"),
|
|
int(payload.get("requires_certificate", 0)),
|
|
int(payload.get("can_offset_annual", 0)),
|
|
int(payload.get("approval_threshold_dept", 3)),
|
|
int(payload.get("approval_threshold_hr", 10)),
|
|
payload.get("notes"),
|
|
),
|
|
)
|
|
conn.commit()
|
|
return int(cur.lastrowid)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------- attendance_holidays ----------
|
|
|
|
def list_holidays() -> list[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"SELECT * FROM {ATTENDANCE_HOLIDAYS_TABLE} ORDER BY holiday_date"
|
|
)
|
|
return [_row_dict(r) for r in cur.fetchall()]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def count_holidays() -> int:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(f"SELECT COUNT(*) FROM {ATTENDANCE_HOLIDAYS_TABLE}")
|
|
return int(cur.fetchone()[0])
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def upsert_holiday(payload: dict[str, Any]) -> int:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
holiday_date = str(payload.get("holiday_date", "")).strip()
|
|
if not holiday_date:
|
|
raise ValueError("假日日期不能为空")
|
|
cur.execute(
|
|
f"SELECT id FROM {ATTENDANCE_HOLIDAYS_TABLE} WHERE holiday_date = ?",
|
|
(holiday_date,),
|
|
)
|
|
existing = cur.fetchone()
|
|
if existing is not None:
|
|
hid = int(existing[0])
|
|
cur.execute(
|
|
f"""UPDATE {ATTENDANCE_HOLIDAYS_TABLE} SET
|
|
holiday_date = ?, holiday_name = ?, holiday_type = ?,
|
|
updated_at = unixepoch()
|
|
WHERE id = ?""",
|
|
(
|
|
holiday_date,
|
|
payload.get("holiday_name"),
|
|
payload.get("holiday_type", "法定假日"),
|
|
hid,
|
|
),
|
|
)
|
|
conn.commit()
|
|
return hid
|
|
cur.execute(
|
|
f"""INSERT INTO {ATTENDANCE_HOLIDAYS_TABLE} (
|
|
holiday_date, holiday_name, holiday_type
|
|
) VALUES (?, ?, ?)""",
|
|
(
|
|
holiday_date,
|
|
payload.get("holiday_name"),
|
|
payload.get("holiday_type", "法定假日"),
|
|
),
|
|
)
|
|
conn.commit()
|
|
return int(cur.lastrowid)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------- leave_requests ----------
|
|
|
|
def create_leave_request(payload: dict[str, Any]) -> int:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"""INSERT INTO {LEAVE_REQUESTS_TABLE} (
|
|
employee_id, leave_type, start_date, end_date,
|
|
has_certificate, certificate_type, natural_days,
|
|
working_days, monthly_salary, working_days_in_month,
|
|
daily_salary, deduction_amount, annual_leave_used,
|
|
remaining_annual_leave, approval_level, status,
|
|
params_json, summary, error_message
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
int(payload.get("employee_id")),
|
|
payload.get("leave_type"),
|
|
payload.get("start_date"),
|
|
payload.get("end_date"),
|
|
int(payload.get("has_certificate", 0)),
|
|
payload.get("certificate_type"),
|
|
payload.get("natural_days"),
|
|
payload.get("working_days"),
|
|
payload.get("monthly_salary"),
|
|
payload.get("working_days_in_month"),
|
|
payload.get("daily_salary"),
|
|
payload.get("deduction_amount"),
|
|
payload.get("annual_leave_used"),
|
|
payload.get("remaining_annual_leave"),
|
|
payload.get("approval_level"),
|
|
payload.get("status", "draft"),
|
|
dumps_json(payload.get("params")) if payload.get("params") else None,
|
|
payload.get("summary"),
|
|
payload.get("error_message"),
|
|
),
|
|
)
|
|
conn.commit()
|
|
return int(cur.lastrowid)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def update_leave_request(request_id: int, **fields: Any) -> None:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
sets = []
|
|
values: list[Any] = []
|
|
for k, v in fields.items():
|
|
if k == "params":
|
|
sets.append("params_json = ?")
|
|
values.append(dumps_json(v))
|
|
elif k == "params_json":
|
|
sets.append("params_json = ?")
|
|
values.append(v)
|
|
else:
|
|
sets.append(f"{k} = ?")
|
|
values.append(v)
|
|
if not sets:
|
|
return
|
|
sets.append("updated_at = unixepoch()")
|
|
values.append(request_id)
|
|
cur.execute(
|
|
f"UPDATE {LEAVE_REQUESTS_TABLE} SET {', '.join(sets)} WHERE id = ?",
|
|
values,
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_request(request_id: int) -> Optional[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"SELECT * FROM {LEAVE_REQUESTS_TABLE} WHERE id = ?",
|
|
(request_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return _row_dict(row) if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_latest_completed_request(
|
|
employee_id: Optional[int] = None,
|
|
) -> Optional[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
if employee_id is not None:
|
|
cur.execute(
|
|
f"""SELECT * FROM {LEAVE_REQUESTS_TABLE}
|
|
WHERE status = 'completed' AND employee_id = ?
|
|
ORDER BY id DESC LIMIT 1""",
|
|
(employee_id,),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
f"""SELECT * FROM {LEAVE_REQUESTS_TABLE}
|
|
WHERE status = 'completed'
|
|
ORDER BY id DESC LIMIT 1"""
|
|
)
|
|
row = cur.fetchone()
|
|
return _row_dict(row) if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def list_requests_for_employee(employee_id: int) -> list[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"""SELECT * FROM {LEAVE_REQUESTS_TABLE}
|
|
WHERE employee_id = ? ORDER BY id DESC""",
|
|
(employee_id,),
|
|
)
|
|
return [_row_dict(r) for r in cur.fetchall()]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------- leave_balances ----------
|
|
|
|
def get_balance(
|
|
employee_id: int, balance_type: str
|
|
) -> Optional[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"""SELECT * FROM {LEAVE_BALANCES_TABLE}
|
|
WHERE employee_id = ? AND balance_type = ?""",
|
|
(employee_id, balance_type),
|
|
)
|
|
row = cur.fetchone()
|
|
return _row_dict(row) if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def upsert_balance(payload: dict[str, Any]) -> int:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
cur = conn.cursor()
|
|
employee_id = int(payload.get("employee_id"))
|
|
balance_type = str(payload.get("balance_type", "")).strip()
|
|
if not balance_type:
|
|
raise ValueError("balance_type 不能为空")
|
|
cur.execute(
|
|
f"""SELECT id FROM {LEAVE_BALANCES_TABLE}
|
|
WHERE employee_id = ? AND balance_type = ?""",
|
|
(employee_id, balance_type),
|
|
)
|
|
existing = cur.fetchone()
|
|
total_days = payload.get("total_days")
|
|
used_days = float(payload.get("used_days", 0) or 0)
|
|
remaining_days = payload.get("remaining_days")
|
|
if existing is not None:
|
|
bid = int(existing[0])
|
|
cur.execute(
|
|
f"""UPDATE {LEAVE_BALANCES_TABLE} SET
|
|
total_days = ?, used_days = ?, remaining_days = ?,
|
|
updated_at = unixepoch()
|
|
WHERE id = ?""",
|
|
(total_days, used_days, remaining_days, bid),
|
|
)
|
|
conn.commit()
|
|
return bid
|
|
cur.execute(
|
|
f"""INSERT INTO {LEAVE_BALANCES_TABLE} (
|
|
employee_id, balance_type, total_days, used_days, remaining_days
|
|
) VALUES (?, ?, ?, ?, ?)""",
|
|
(employee_id, balance_type, total_days, used_days, remaining_days),
|
|
)
|
|
conn.commit()
|
|
return int(cur.lastrowid)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def list_balances_for_employee(employee_id: int) -> list[dict[str, Any]]:
|
|
ensure_db()
|
|
conn = get_conn()
|
|
try:
|
|
conn.row_factory = sqlite3.Row
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"""SELECT * FROM {LEAVE_BALANCES_TABLE}
|
|
WHERE employee_id = ? ORDER BY balance_type""",
|
|
(employee_id,),
|
|
)
|
|
return [_row_dict(r) for r in cur.fetchall()]
|
|
finally:
|
|
conn.close()
|