from __future__ import annotations from dataclasses import dataclass from typing import Any from app.services.expense_claim_workflow_constants import ( APPROVAL_DONE_STAGE, ARCHIVE_ACCOUNTING_STAGE, BUDGET_MANAGER_APPROVAL_STAGE, DIRECT_MANAGER_APPROVAL_STAGE, FINANCE_APPROVAL_STAGE, PAYMENT_PAID_STAGE, PAYMENT_PENDING_STAGE, ) @dataclass(frozen=True, slots=True) class ExpenseClaimStatusSpec: code: int value: str label: str terminal: bool = False @dataclass(frozen=True, slots=True) class ExpenseClaimState: status: str approval_stage: str status_code: int | None status_label: str changed: bool CLAIM_STATUS_REGISTRY: dict[str, ExpenseClaimStatusSpec] = { "draft": ExpenseClaimStatusSpec(10, "draft", "草稿"), "submitted": ExpenseClaimStatusSpec(20, "submitted", "审批中"), "approved": ExpenseClaimStatusSpec(30, "approved", "已通过"), "pending_payment": ExpenseClaimStatusSpec(40, "pending_payment", "待付款"), "paid": ExpenseClaimStatusSpec(50, "paid", "已付款", terminal=True), "returned": ExpenseClaimStatusSpec(60, "returned", "待补充"), "rejected": ExpenseClaimStatusSpec(70, "rejected", "已驳回", terminal=True), } CLAIM_STATUS_ALIASES = { "review": "submitted", "pending_review": "submitted", "approving": "submitted", "manager_review": "submitted", "budget_review": "submitted", "finance_review": "submitted", "completed": "approved", "complete": "approved", "payment": "pending_payment", "supplement": "returned", "草稿": "draft", "待提交": "draft", "已提交": "submitted", "审批中": "submitted", "审核中": "submitted", "审批完成": "approved", "已通过": "approved", "归档入账": "approved", "待付款": "pending_payment", "已付款": "paid", "待补充": "returned", "已驳回": "rejected", } CANONICAL_APPROVAL_STAGES = { "", "待提交", DIRECT_MANAGER_APPROVAL_STAGE, BUDGET_MANAGER_APPROVAL_STAGE, FINANCE_APPROVAL_STAGE, APPROVAL_DONE_STAGE, ARCHIVE_ACCOUNTING_STAGE, PAYMENT_PENDING_STAGE, PAYMENT_PAID_STAGE, "待补充", "已驳回", } STAGE_ALIASES = { "draft": "待提交", "review": DIRECT_MANAGER_APPROVAL_STAGE, "pending_review": DIRECT_MANAGER_APPROVAL_STAGE, "approving": DIRECT_MANAGER_APPROVAL_STAGE, "manager_review": DIRECT_MANAGER_APPROVAL_STAGE, "budget_review": BUDGET_MANAGER_APPROVAL_STAGE, "finance_review": FINANCE_APPROVAL_STAGE, "pending_payment": PAYMENT_PENDING_STAGE, "supplement": "待补充", "rejected": "已驳回", "草稿": "待提交", "审核中": DIRECT_MANAGER_APPROVAL_STAGE, } STATUS_DEFAULT_STAGE = { "draft": "待提交", "submitted": DIRECT_MANAGER_APPROVAL_STAGE, "pending_payment": PAYMENT_PENDING_STAGE, "paid": PAYMENT_PAID_STAGE, "returned": "待补充", "rejected": "已驳回", } LEGACY_REVIEW_STATUS_STAGE = { "review": DIRECT_MANAGER_APPROVAL_STAGE, "pending_review": DIRECT_MANAGER_APPROVAL_STAGE, "approving": DIRECT_MANAGER_APPROVAL_STAGE, "manager_review": DIRECT_MANAGER_APPROVAL_STAGE, "budget_review": BUDGET_MANAGER_APPROVAL_STAGE, "finance_review": FINANCE_APPROVAL_STAGE, } def normalize_claim_status(value: Any) -> str: raw = str(value or "").strip() if not raw: return "" lowered = raw.lower() if lowered in CLAIM_STATUS_REGISTRY: return lowered return CLAIM_STATUS_ALIASES.get(lowered) or CLAIM_STATUS_ALIASES.get(raw) or raw def claim_status_code(value: Any) -> int | None: status = normalize_claim_status(value) spec = CLAIM_STATUS_REGISTRY.get(status) return spec.code if spec is not None else None def claim_status_label(value: Any) -> str: status = normalize_claim_status(value) spec = CLAIM_STATUS_REGISTRY.get(status) return spec.label if spec is not None else str(value or "").strip() def is_known_claim_status(value: Any) -> bool: return normalize_claim_status(value) in CLAIM_STATUS_REGISTRY def is_known_approval_stage(value: Any) -> bool: stage = str(value or "").strip() normalized_stage = _normalize_stage_alias(stage) return stage in CANONICAL_APPROVAL_STAGES or normalized_stage in CANONICAL_APPROVAL_STAGES def is_application_claim_reference( *, claim_no: str | None = None, expense_type: str | None = None, ) -> bool: normalized_no = str(claim_no or "").strip().upper() normalized_type = str(expense_type or "").strip().lower() return ( normalized_no.startswith(("AP-", "APP-")) or normalized_type == "application" or normalized_type.endswith("_application") ) def normalize_expense_claim_state( status: Any, approval_stage: Any, *, claim_no: str | None = None, expense_type: str | None = None, is_application_claim: bool | None = None, ) -> ExpenseClaimState: original_status = str(status or "").strip() original_stage = str(approval_stage or "").strip() normalized_status = normalize_claim_status(original_status) normalized_stage = _normalize_stage_alias(original_stage) application = ( is_application_claim if is_application_claim is not None else is_application_claim_reference(claim_no=claim_no, expense_type=expense_type) ) legacy_status = original_status.lower() if legacy_status in LEGACY_REVIEW_STATUS_STAGE: normalized_stage = LEGACY_REVIEW_STATUS_STAGE[legacy_status] elif normalized_status == "approved": normalized_stage = _approved_stage(original_stage, application) elif normalized_status == "pending_payment": normalized_stage = PAYMENT_PENDING_STAGE elif normalized_status == "paid": normalized_stage = PAYMENT_PAID_STAGE elif normalized_status in STATUS_DEFAULT_STAGE and not normalized_stage: normalized_stage = STATUS_DEFAULT_STAGE[normalized_status] if normalized_status == "submitted" and normalized_stage in {"payment", "completed"}: normalized_stage = DIRECT_MANAGER_APPROVAL_STAGE spec = CLAIM_STATUS_REGISTRY.get(normalized_status) return ExpenseClaimState( status=normalized_status, approval_stage=normalized_stage, status_code=spec.code if spec is not None else None, status_label=spec.label if spec is not None else normalized_status, changed=normalized_status != original_status or normalized_stage != original_stage, ) def _normalize_stage_alias(value: str) -> str: if not value: return "" lowered = value.lower() return STAGE_ALIASES.get(lowered) or STAGE_ALIASES.get(value) or value def _approved_stage(raw_stage: str, is_application_claim: bool) -> str: stage = _normalize_stage_alias(raw_stage) lowered = str(raw_stage or "").strip().lower() if is_application_claim: if not stage or lowered == "completed": return APPROVAL_DONE_STAGE return stage if stage in {ARCHIVE_ACCOUNTING_STAGE, PAYMENT_PAID_STAGE}: return stage if lowered in {"completed", "complete", ""} or stage == APPROVAL_DONE_STAGE: return ARCHIVE_ACCOUNTING_STAGE return stage