from __future__ import annotations from decimal import Decimal from uuid import UUID from sqlalchemy import desc, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.exceptions import AppException, NotFoundError from app.domains.hr_tools.calculator import Segment, calculate from app.domains.hr_tools.models import HrToolsAnnualLeaveCalculation from app.domains.hr_tools.schemas import ( AnnualLeaveCalculateRequest, AnnualLeaveCalculateResponse, AnnualLeaveHistoryItem, BlockResponse, SegmentResponse, ) def _segments_to_json(segments: list[Segment]) -> list[dict]: return [s.to_dict() for s in segments] def _segment_dict_to_response(segment: dict) -> SegmentResponse: """Rehydrate a stored segment JSON entry into a response model. Mirrors ``Segment.to_dict`` exactly. Used by ``get_history_item`` so a stored audit row is replayed VERBATIM, not recomputed against the current rule engine — important when the policy table evolves later and we still need to defend a historical settlement number. """ from datetime import date as _date return SegmentResponse( index=int(segment["index"]), start_date=_date.fromisoformat(segment["start_date"]), end_date=_date.fromisoformat(segment["end_date"]), days=int(segment["days"]), legal_basis=int(segment["legal_basis"]), welfare_basis=int(segment["welfare_basis"]), legal_formula=segment["legal_formula"], welfare_formula=segment["welfare_formula"], legal_raw_value=Decimal(segment["legal_raw_value"]), welfare_raw_value=Decimal(segment["welfare_raw_value"]), ) def _record_to_response( record: HrToolsAnnualLeaveCalculation, ) -> AnnualLeaveCalculateResponse: """Build the response from PERSISTED columns only. `get_history_item` calls this so a historical row replays with the exact numbers we showed HR at compute time — even after the rule engine has changed. `calculate_and_record` also calls this on the write path so the response shape is generated by one code path. """ full_year_segments = [ _segment_dict_to_response(s) for s in (record.full_year_segments or []) ] settlement_segments = [ _segment_dict_to_response(s) for s in (record.settlement_segments or []) ] return AnnualLeaveCalculateResponse( id=UUID(record.id), person_id=UUID(record.person_id) if record.person_id else None, employee_no=record.employee_no, employee_name=record.employee_name, work_start_date=record.work_start_date, hire_date=record.hire_date, termination_date=record.termination_date, fiscal_year_start=record.fiscal_year_start, fiscal_year_end=record.fiscal_year_end, entitlement=BlockResponse( label="entitlement", range_start=record.fiscal_year_start, range_end=record.fiscal_year_end, segments=full_year_segments, legal_raw=record.full_year_legal_raw, welfare_raw=record.full_year_welfare_raw, legal_rounded=record.full_year_legal_rounded, welfare_rounded=record.full_year_welfare_rounded, legal_capped=record.full_year_legal_capped, welfare_capped=record.full_year_welfare_capped, cap_applied=record.cap_applied_full_year, basis_cap_applied=bool(record.basis_cap_applied_full_year), ), settlement=BlockResponse( label="settlement", range_start=( settlement_segments[0].start_date if settlement_segments else record.fiscal_year_start ), range_end=( settlement_segments[-1].end_date if settlement_segments else record.termination_date ), segments=settlement_segments, legal_raw=record.settlement_legal_raw, welfare_raw=record.settlement_welfare_raw, legal_rounded=record.settlement_legal_rounded, welfare_rounded=record.settlement_welfare_rounded, legal_capped=record.settlement_legal_capped, welfare_capped=record.settlement_welfare_capped, cap_applied=record.cap_applied_settlement, basis_cap_applied=bool(record.basis_cap_applied_settlement), ), carry_legal_days=record.carry_legal_days, carry_welfare_days=record.carry_welfare_days, used_legal_days=record.used_legal_days, used_welfare_days=record.used_welfare_days, # 本期余额 — derived from the persisted settlement cap and used # days (NOT a stored column), so historical rows replay the same # value they showed at compute time. Equals settlement − used; # period_balance + carry == balance_*_days by construction. period_balance_legal_days=( record.settlement_legal_capped - record.used_legal_days ).quantize(Decimal("0.1")), period_balance_welfare_days=( record.settlement_welfare_capped - record.used_welfare_days ).quantize(Decimal("0.1")), balance_legal_days=record.balance_legal_days, balance_welfare_days=record.balance_welfare_days, calculated_by_user_id=UUID(record.calculated_by_user_id), calculated_by_name=record.calculated_by_name, notes=record.notes, created_at=record.created_at, ) class AnnualLeaveCalculatorService: def __init__(self, db: AsyncSession): self.db = db async def calculate_and_record( self, *, body: AnnualLeaveCalculateRequest, actor: AuthUser, trace_id: str, ip_address: str | None = None, user_agent: str | None = None, ) -> AnnualLeaveCalculateResponse: if body.termination_date < body.hire_date: raise AppException( code="HR_TOOLS.INVALID_TERMINATION_DATE", message="离职日期不能早于入职日期", status_code=422, ) if body.hire_date < body.work_start_date: raise AppException( code="HR_TOOLS.INVALID_HIRE_DATE", message="入职日期不能早于参加工作日期", status_code=422, ) result = calculate( work_start_date=body.work_start_date, hire_date=body.hire_date, termination_date=body.termination_date, carry_legal_days=body.carry_legal_days, carry_welfare_days=body.carry_welfare_days, used_legal_days=body.used_legal_days, used_welfare_days=body.used_welfare_days, ) record = HrToolsAnnualLeaveCalculation( person_id=str(body.person_id) if body.person_id else None, employee_no=body.employee_no, employee_name=body.employee_name, work_start_date=result.work_start_date, hire_date=result.hire_date, termination_date=result.termination_date, fiscal_year_start=result.fiscal_year_start, fiscal_year_end=result.fiscal_year_end, carry_legal_days=result.carry_legal_days, carry_welfare_days=result.carry_welfare_days, used_legal_days=result.used_legal_days, used_welfare_days=result.used_welfare_days, full_year_segments=_segments_to_json(result.entitlement.segments), full_year_legal_raw=result.entitlement.legal_raw, full_year_welfare_raw=result.entitlement.welfare_raw, full_year_legal_rounded=result.entitlement.legal_rounded, full_year_welfare_rounded=result.entitlement.welfare_rounded, full_year_legal_capped=result.entitlement.legal_capped, full_year_welfare_capped=result.entitlement.welfare_capped, settlement_segments=_segments_to_json(result.settlement.segments), settlement_legal_raw=result.settlement.legal_raw, settlement_welfare_raw=result.settlement.welfare_raw, settlement_legal_rounded=result.settlement.legal_rounded, settlement_welfare_rounded=result.settlement.welfare_rounded, settlement_legal_capped=result.settlement.legal_capped, settlement_welfare_capped=result.settlement.welfare_capped, balance_legal_days=result.balance_legal_days, balance_welfare_days=result.balance_welfare_days, cap_applied_full_year=result.entitlement.cap_applied, cap_applied_settlement=result.settlement.cap_applied, basis_cap_applied_full_year=result.entitlement.basis_cap_applied, basis_cap_applied_settlement=result.settlement.basis_cap_applied, calculated_by_user_id=str(actor.id), # Defensive truncation: ``identity_persons.work_email`` is # VARCHAR(255) but our column is String(128). A service # account with an unusually long email would silently # truncate (non-strict MySQL) or 500 (strict). Clamp here # so the displayed actor name survives a long email. calculated_by_name=(actor.display_name or actor.work_email)[:128], notes=body.notes, ) self.db.add(record) await self.db.flush() await AuditService(self.db).log( trace_id=trace_id, actor_user_id=actor.id, actor_type="user", action="hr_tools.annual_leave.calculate", resource_type="hr_tools_annual_leave_calculation", resource_id=record.id, status="success", detail={ "person_id": record.person_id, "employee_no": record.employee_no, "balance_legal": str(record.balance_legal_days), "balance_welfare": str(record.balance_welfare_days), "cap_applied": record.cap_applied_settlement, "basis_cap_applied": record.basis_cap_applied_settlement, }, ip_address=ip_address, user_agent=user_agent, ) await self.db.commit() await self.db.refresh(record) return _record_to_response(record) async def list_history( self, *, person_id: UUID | None, visible_to_user_id: UUID | None = None, limit: int, offset: int, ) -> tuple[list[AnnualLeaveHistoryItem], int]: """List annual-leave history rows. ``person_id`` — if set, ANDed in as a subject filter (used by callers that legitimately want one specific person's history, e.g. an admin tool that already enforces its own access check). ``visible_to_user_id`` — if set, ORs the row's ``person_id == visible_to_user_id`` with ``calculated_by_user_id == visible_to_user_id`` so the caller sees both rows ABOUT them and rows they personally calculated. Used by the per-user history feed under ``GET /annual-leave/history`` (2026-05-14 fix after the strict ``person_id`` self-scope returned 0 rows for HR who had only ever calculated for others). Sandbox rows (``person_id IS NULL``) are reachable via the ``calculated_by_user_id`` branch only — i.e. only the HR who created the sandbox sees it. """ from sqlalchemy import func as sqlfunc from sqlalchemy import or_ base_query = select(HrToolsAnnualLeaveCalculation) count_query = select(sqlfunc.count(HrToolsAnnualLeaveCalculation.id)) if person_id is not None: base_query = base_query.where( HrToolsAnnualLeaveCalculation.person_id == str(person_id) ) count_query = count_query.where( HrToolsAnnualLeaveCalculation.person_id == str(person_id) ) if visible_to_user_id is not None: visibility_clause = or_( HrToolsAnnualLeaveCalculation.person_id == str(visible_to_user_id), HrToolsAnnualLeaveCalculation.calculated_by_user_id == str(visible_to_user_id), ) base_query = base_query.where(visibility_clause) count_query = count_query.where(visibility_clause) base_query = ( base_query.order_by(desc(HrToolsAnnualLeaveCalculation.created_at)) .limit(limit) .offset(offset) ) rows = (await self.db.execute(base_query)).scalars().all() total = (await self.db.execute(count_query)).scalar() or 0 items = [ AnnualLeaveHistoryItem( id=UUID(r.id), person_id=UUID(r.person_id) if r.person_id else None, employee_no=r.employee_no, employee_name=r.employee_name, work_start_date=r.work_start_date, hire_date=r.hire_date, termination_date=r.termination_date, balance_legal_days=r.balance_legal_days, balance_welfare_days=r.balance_welfare_days, calculated_by_name=r.calculated_by_name, created_at=r.created_at, ) for r in rows ] return items, total async def get_history_item( self, record_id: UUID, ) -> AnnualLeaveCalculateResponse: """Replay a historical calculation from the stored row. Builds the response from the persisted JSON segments + Decimal column values — does NOT re-invoke ``calculate()``. This keeps the audit-grade record stable across rule-engine changes: if the policy table is updated next year, the row for last year's settlement still shows the exact numbers HR computed at the time. """ record = await self.db.get(HrToolsAnnualLeaveCalculation, str(record_id)) if not record: raise NotFoundError("AnnualLeaveCalculation", str(record_id)) return _record_to_response(record)