113 lines
3.8 KiB
Python
113 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.api.deps import CurrentUserContext
|
|
from app.models.financial_record import ExpenseClaim
|
|
from app.services.budget import BudgetService
|
|
from app.services.expense_claim_budget_risk_flags import dedupe_budget_risk_flags
|
|
from app.services.expense_claim_risk_stage import enrich_risk_flag_semantics
|
|
|
|
|
|
class ExpenseClaimBudgetFlowMixin:
|
|
def _reserve_budget_for_submission(
|
|
self,
|
|
claim: ExpenseClaim,
|
|
current_user: CurrentUserContext,
|
|
*,
|
|
is_application_claim: bool,
|
|
) -> list[dict[str, Any]]:
|
|
source_type = "application" if is_application_claim else "claim"
|
|
return BudgetService(self.db).reserve_for_claim(
|
|
claim,
|
|
source_type=source_type,
|
|
operator=self._resolve_budget_operator(current_user),
|
|
)
|
|
|
|
def _release_budget_for_return(
|
|
self,
|
|
claim: ExpenseClaim,
|
|
current_user: CurrentUserContext,
|
|
*,
|
|
reason: str,
|
|
) -> list[dict[str, Any]]:
|
|
is_application_claim = self._is_expense_application_claim(claim)
|
|
source_type = "application" if is_application_claim else "claim"
|
|
return BudgetService(self.db).release_for_claim(
|
|
claim,
|
|
source_type=source_type,
|
|
operator=self._resolve_budget_operator(current_user),
|
|
reason=reason,
|
|
)
|
|
|
|
def _release_budget_for_delete(
|
|
self,
|
|
claim: ExpenseClaim,
|
|
current_user: CurrentUserContext,
|
|
) -> None:
|
|
is_application_claim = self._is_expense_application_claim(claim)
|
|
source_type = "application" if is_application_claim else "claim"
|
|
BudgetService(self.db).release_for_claim(
|
|
claim,
|
|
source_type=source_type,
|
|
operator=self._resolve_budget_operator(current_user),
|
|
reason="单据删除释放预算预占",
|
|
)
|
|
|
|
def _consume_budget_for_finance_approval(
|
|
self,
|
|
claim: ExpenseClaim,
|
|
current_user: CurrentUserContext,
|
|
) -> dict[str, Any] | None:
|
|
return BudgetService(self.db).consume_for_claim(
|
|
claim,
|
|
operator=self._resolve_budget_operator(current_user),
|
|
reason="财务终审通过核销预算",
|
|
)
|
|
|
|
def _transfer_application_budget_to_reimbursement(
|
|
self,
|
|
*,
|
|
application_claim: ExpenseClaim,
|
|
draft_claim: ExpenseClaim,
|
|
current_user: CurrentUserContext,
|
|
) -> dict[str, Any] | None:
|
|
return BudgetService(self.db).transfer_application_reservation(
|
|
application_claim=application_claim,
|
|
draft_claim=draft_claim,
|
|
operator=self._resolve_budget_operator(current_user),
|
|
)
|
|
|
|
@staticmethod
|
|
def _append_budget_flags(
|
|
risk_flags: list[Any] | None,
|
|
budget_flags: list[dict[str, Any]] | dict[str, Any] | None,
|
|
*,
|
|
business_stage: str | None = None,
|
|
) -> list[Any]:
|
|
if budget_flags is None:
|
|
return list(risk_flags or [])
|
|
if isinstance(budget_flags, dict):
|
|
next_flags = [budget_flags]
|
|
else:
|
|
next_flags = list(budget_flags or [])
|
|
if not next_flags:
|
|
return list(risk_flags or [])
|
|
enriched_flags = [
|
|
enrich_risk_flag_semantics(
|
|
flag,
|
|
business_stage=business_stage,
|
|
risk_domain="budget",
|
|
visibility_scope="budget_manager",
|
|
actionability="budget_governance",
|
|
)
|
|
if isinstance(flag, dict)
|
|
else flag
|
|
for flag in next_flags
|
|
]
|
|
return dedupe_budget_risk_flags([*list(risk_flags or []), *enriched_flags])
|
|
|
|
@staticmethod
|
|
def _resolve_budget_operator(current_user: CurrentUserContext) -> str:
|
|
return current_user.name or current_user.username or "system"
|