47c6a4bb73
- DOCUMENT_NUMBER_PREFIXES 改为 A/R/D，新增短格式与旧格式正则并存识别，提取正则加边界锚定避免误匹配 - build_document_number 去掉时间戳段，统一生成 A+token 等紧凑单号，is_application_claim_no 兼容旧 AP-/APP- 前缀 - access_policy/status_registry/reimbursements/expense_claims/budget_support 统一复用 is_application_claim_no 判定申请单 - 同步 document_numbering 单元测试覆盖新旧两种格式
232 lines
7.4 KiB
Python
232 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from app.services.expense_claim_workflow_constants import (
|
|
APPROVAL_DONE_STAGE,
|
|
APPLICATION_ARCHIVE_STAGE,
|
|
APPLICATION_LINK_STATUS_STAGE,
|
|
ARCHIVE_ACCOUNTING_STAGE,
|
|
BUDGET_MANAGER_APPROVAL_STAGE,
|
|
DIRECT_MANAGER_APPROVAL_STAGE,
|
|
FINANCE_APPROVAL_STAGE,
|
|
PAYMENT_PAID_STAGE,
|
|
PAYMENT_PENDING_STAGE,
|
|
)
|
|
from app.services.document_numbering import is_application_claim_no
|
|
|
|
|
|
@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,
|
|
APPLICATION_LINK_STATUS_STAGE,
|
|
APPLICATION_ARCHIVE_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 (
|
|
is_application_claim_no(normalized_no)
|
|
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 stage == APPLICATION_ARCHIVE_STAGE:
|
|
return APPLICATION_ARCHIVE_STAGE
|
|
if not stage or lowered == "completed" or stage == APPROVAL_DONE_STAGE:
|
|
return APPLICATION_LINK_STATUS_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
|