203 lines
8.0 KiB
Python
203 lines
8.0 KiB
Python
from __future__ import annotations
|
||
|
||
from calendar import monthrange
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from pathlib import Path
|
||
import re
|
||
from typing import BinaryIO
|
||
|
||
from ..core.logger import AppLogger
|
||
from ..database.orm import MonthlyPayrollRunORM, PayrollExceptionORM, PayrollJobORM
|
||
from ..domain.payroll_exceptions import PayrollException, detect_payroll_exceptions
|
||
from ..repositories import EmployeeRepository, MonthlyPayrollRepository
|
||
from .payroll_service import PayrollApplicationService, PayrollComputation
|
||
|
||
logger = AppLogger.get_logger(__name__)
|
||
SALARY_MONTH_RE = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MonthlyPayrollDetailData:
|
||
run: MonthlyPayrollRunORM
|
||
job: PayrollJobORM | None
|
||
exceptions: list[PayrollExceptionORM]
|
||
|
||
|
||
class MonthlyPayrollService:
|
||
"""月度核算应用服务：编排正式工资计算、异常识别、锁定和导出。"""
|
||
|
||
def __init__(
|
||
self,
|
||
repository: MonthlyPayrollRepository,
|
||
payroll_service: PayrollApplicationService,
|
||
employee_repository: EmployeeRepository,
|
||
):
|
||
self.repository = repository
|
||
self.payroll_service = payroll_service
|
||
self.employee_repository = employee_repository
|
||
|
||
def list_runs(self, salary_month: str | None = None) -> list[MonthlyPayrollRunORM]:
|
||
logger.info("查询月度核算批次 salary_month=%s", salary_month)
|
||
return self.repository.list_runs(salary_month)
|
||
|
||
def get_detail(self, run_id: int) -> MonthlyPayrollDetailData:
|
||
run = self.repository.require_run(run_id)
|
||
job = self.payroll_service.get_job(run.source_job_id) if run.source_job_id else None
|
||
exceptions = self.repository.list_exceptions(run.id)
|
||
logger.info("查询月度核算详情 run_id=%s job_found=%s exception_count=%s", run_id, job is not None, len(exceptions))
|
||
return MonthlyPayrollDetailData(run=run, job=job, exceptions=exceptions)
|
||
|
||
def calculate_from_excel(
|
||
self,
|
||
*,
|
||
salary_month: str,
|
||
filename: str,
|
||
stream: BinaryIO,
|
||
created_by: str,
|
||
export_excel: bool,
|
||
) -> MonthlyPayrollDetailData:
|
||
self._ensure_month_can_calculate(salary_month)
|
||
computation = self.payroll_service.calculate_from_excel_upload(
|
||
filename=filename,
|
||
stream=stream,
|
||
config_path=None,
|
||
export_excel=export_excel,
|
||
)
|
||
content_month = self._month_from_computation(computation)
|
||
if content_month != salary_month:
|
||
raise ValueError(f"Excel 内容月份为 {content_month}，与当前核算月份 {salary_month} 不一致")
|
||
run = self._save_calculation(
|
||
salary_month=salary_month,
|
||
source_type="excel",
|
||
created_by=created_by,
|
||
computation=computation,
|
||
)
|
||
return self.get_detail(run.id)
|
||
|
||
async def calculate_from_dingtalk(
|
||
self,
|
||
*,
|
||
salary_month: str,
|
||
created_by: str,
|
||
export_excel: bool,
|
||
) -> MonthlyPayrollDetailData:
|
||
self._ensure_month_can_calculate(salary_month)
|
||
employees = [
|
||
employee
|
||
for employee in self.employee_repository.list_employees(employment_status="active")
|
||
if employee.dingtalk_user_id
|
||
]
|
||
if not employees:
|
||
raise ValueError("没有绑定钉钉 userId 的在职员工，请先在员工维护中补齐钉钉用户ID")
|
||
start_date, end_date = _month_range(salary_month)
|
||
computation = await self.payroll_service.calculate_from_dingtalk_realtime(
|
||
user_ids=[employee.dingtalk_user_id for employee in employees],
|
||
start_date=start_date,
|
||
end_date=end_date,
|
||
config_path=None,
|
||
export_excel=export_excel,
|
||
)
|
||
run = self._save_calculation(
|
||
salary_month=salary_month,
|
||
source_type="dingtalk",
|
||
created_by=created_by,
|
||
computation=computation,
|
||
)
|
||
return self.get_detail(run.id)
|
||
|
||
def lock_run(self, run_id: int, locked_by: str) -> MonthlyPayrollRunORM:
|
||
logger.info("锁定月度核算 run_id=%s locked_by=%s", run_id, locked_by)
|
||
return self.repository.lock_run(run_id, locked_by)
|
||
|
||
def export_path(self, run_id: int) -> Path:
|
||
detail = self.get_detail(run_id)
|
||
if detail.job is None or not detail.job.output_file:
|
||
raise FileNotFoundError("该月度核算批次没有可导出的工资表")
|
||
path = Path(detail.job.output_file)
|
||
if path.exists():
|
||
return self.payroll_service.localize_export_file(path, f"monthly_{detail.run.source_type}")
|
||
resolved_path = self.payroll_service.resolve_output_file(path.name)
|
||
return self.payroll_service.localize_export_file(resolved_path, f"monthly_{detail.run.source_type}")
|
||
|
||
def _save_calculation(
|
||
self,
|
||
*,
|
||
salary_month: str,
|
||
source_type: str,
|
||
created_by: str,
|
||
computation: PayrollComputation,
|
||
) -> MonthlyPayrollRunORM:
|
||
run = self.repository.create_run(
|
||
salary_month=salary_month,
|
||
source_type=source_type,
|
||
source_job_id=computation.job_id,
|
||
created_by=created_by,
|
||
status="calculated",
|
||
)
|
||
self.repository.update_run_totals(
|
||
run.id,
|
||
employee_count=computation.employee_count,
|
||
gross_total=round(sum(result.gross_salary or 0 for result in computation.results), 2),
|
||
net_total=round(sum(result.net_salary or 0 for result in computation.results), 2),
|
||
deduction_total=round(sum(result.total_deduction for result in computation.results), 2),
|
||
overtime_total=round(sum(result.remaining_overtime_hours for result in computation.results), 2),
|
||
status="calculated",
|
||
)
|
||
self.repository.replace_exceptions(
|
||
run.id,
|
||
self._exception_rows(run.id, detect_payroll_exceptions(computation.results)),
|
||
)
|
||
logger.info("月度核算保存完成 run_id=%s job_id=%s", run.id, computation.job_id)
|
||
return self.repository.require_run(run.id)
|
||
|
||
def _exception_rows(
|
||
self,
|
||
run_id: int,
|
||
exceptions: list[PayrollException],
|
||
) -> list[PayrollExceptionORM]:
|
||
rows: list[PayrollExceptionORM] = []
|
||
for item in exceptions:
|
||
employee = self.employee_repository.get_by_employee_no(item.employee_no) if item.employee_no else None
|
||
rows.append(
|
||
PayrollExceptionORM(
|
||
run_id=run_id,
|
||
employee_id=employee.id if employee else None,
|
||
employee_no=item.employee_no,
|
||
employee_name=item.employee_name,
|
||
salary_month=item.salary_month,
|
||
exception_type=item.exception_type,
|
||
severity=item.severity,
|
||
message=item.message,
|
||
status="open",
|
||
)
|
||
)
|
||
return rows
|
||
|
||
def _ensure_month_can_calculate(self, salary_month: str) -> None:
|
||
_validate_salary_month(salary_month)
|
||
existing = self.repository.get_run_by_month(salary_month)
|
||
if existing and existing.status == "locked":
|
||
raise ValueError("本月工资已锁定，不能重新计算")
|
||
|
||
def _month_from_computation(self, computation: PayrollComputation) -> str:
|
||
for result in computation.results:
|
||
if result.salary_month:
|
||
return result.salary_month
|
||
raise ValueError("无法从工资结果识别工资月份")
|
||
|
||
|
||
def _month_range(salary_month: str) -> tuple[date, date]:
|
||
_validate_salary_month(salary_month)
|
||
year_text, month_text = salary_month.split("-", 1)
|
||
year = int(year_text)
|
||
month = int(month_text)
|
||
return date(year, month, 1), date(year, month, monthrange(year, month)[1])
|
||
|
||
|
||
def _validate_salary_month(salary_month: str) -> None:
|
||
if not salary_month:
|
||
raise ValueError("请选择核算月份")
|
||
if not SALARY_MONTH_RE.match(salary_month):
|
||
raise ValueError("核算月份格式错误，请选择正确的月份")
|