diff --git a/backend/app/schemas/bx/invoice.py b/backend/app/schemas/bx/invoice.py new file mode 100644 index 000000000..c3cd8945c --- /dev/null +++ b/backend/app/schemas/bx/invoice.py @@ -0,0 +1,35 @@ +from typing import Optional + +from pydantic import BaseModel, ConfigDict +from pydantic.alias_generators import to_camel + +from app.schemas.bx.common import DatetimeToStr, OptDatetimeToStr + + +class WalletSummaryOut(BaseModel): + model_config = ConfigDict(from_attributes=True, populate_by_name=True, alias_generator=to_camel) + + id: str + owner_user_id: str + owner_user_name: str + total_amount: float + used_amount: float + balance: float + invoice_count: int + version: int + create_time: DatetimeToStr + update_time: OptDatetimeToStr = None + + +class LedgerOut(BaseModel): + model_config = ConfigDict(from_attributes=True, populate_by_name=True, alias_generator=to_camel) + + id: str + owner_user_id: str + entry_type: str + amount: float + invoice_id: Optional[str] = None + reimburse_id: Optional[str] = None + ref_ledger_id: Optional[str] = None + remark: Optional[str] = None + create_time: DatetimeToStr diff --git a/backend/app/services/bx/invoice_ledger.py b/backend/app/services/bx/invoice_ledger.py new file mode 100644 index 000000000..76376c06e --- /dev/null +++ b/backend/app/services/bx/invoice_ledger.py @@ -0,0 +1,327 @@ +from __future__ import annotations + +from decimal import Decimal +from typing import Any, Dict, Iterable, List, Optional + +from fastapi import HTTPException, status +from sqlalchemy.orm import Session + +from app.core.bx_auth import BxUserContext +from app.core.utils import generate_uuid +from app.models.bx_models import BxInvoiceLedger, BxInvoiceWalletSummary + +ENTRY_CREDIT_UPLOAD = "credit_upload" +ENTRY_DEBIT_DELETE = "debit_delete" +ENTRY_DEBIT_REIMBURSE = "debit_reimburse" +ENTRY_CREDIT_RELEASE = "credit_release" + +_CREDIT_TYPES = {ENTRY_CREDIT_UPLOAD, ENTRY_CREDIT_RELEASE} +_DEBIT_TYPES = {ENTRY_DEBIT_DELETE, ENTRY_DEBIT_REIMBURSE} + + +def summarize_entries(rows: Iterable[Dict[str, Any]]) -> Dict[str, Decimal]: + credit_upload = Decimal("0") + debit_delete = Decimal("0") + debit_reimburse = Decimal("0") + credit_release = Decimal("0") + for row in rows: + amt = Decimal(str(row["amount"])) + entry_type = row["entry_type"] + if entry_type == ENTRY_CREDIT_UPLOAD: + credit_upload += amt + elif entry_type == ENTRY_DEBIT_DELETE: + debit_delete += amt + elif entry_type == ENTRY_DEBIT_REIMBURSE: + debit_reimburse += amt + elif entry_type == ENTRY_CREDIT_RELEASE: + credit_release += amt + total = credit_upload - debit_delete + used = debit_reimburse - credit_release + return { + "total_amount": total, + "used_amount": used, + "balance": total - used, + } + + +def should_skip_release(*, has_debit: bool, has_release: bool) -> bool: + if not has_debit: + return True + return has_release + + +class InvoiceLedgerService: + @staticmethod + def list_entries( + db: Session, tenant_no: str, owner_user_id: str + ) -> List[BxInvoiceLedger]: + return ( + db.query(BxInvoiceLedger) + .filter( + BxInvoiceLedger.tenant_no == tenant_no, + BxInvoiceLedger.owner_user_id == owner_user_id, + BxInvoiceLedger.del_flag == 0, + ) + .order_by(BxInvoiceLedger.create_time.asc()) + .all() + ) + + @staticmethod + def calc_summary(db: Session, tenant_no: str, owner_user_id: str) -> Dict[str, Decimal]: + rows = InvoiceLedgerService.list_entries(db, tenant_no, owner_user_id) + payload = [{"entry_type": row.entry_type, "amount": row.amount} for row in rows] + return summarize_entries(payload) + + @staticmethod + def _lock_summary( + db: Session, + tenant_no: str, + owner_user_id: str, + *, + owner_user_name: str = "", + user_id: str = "", + dept_id: str = "", + ) -> BxInvoiceWalletSummary: + row = ( + db.query(BxInvoiceWalletSummary) + .filter( + BxInvoiceWalletSummary.tenant_no == tenant_no, + BxInvoiceWalletSummary.owner_user_id == owner_user_id, + BxInvoiceWalletSummary.del_flag == 0, + ) + .with_for_update() + .first() + ) + if row: + return row + row = BxInvoiceWalletSummary( + id=generate_uuid(), + tenant_no=tenant_no, + owner_user_id=owner_user_id, + owner_user_name=owner_user_name or "", + total_amount=Decimal("0"), + used_amount=Decimal("0"), + balance=Decimal("0"), + invoice_count=0, + version=0, + create_by=user_id or None, + create_dept_id=dept_id or "", + ) + db.add(row) + db.flush() + return ( + db.query(BxInvoiceWalletSummary) + .filter( + BxInvoiceWalletSummary.id == row.id, + BxInvoiceWalletSummary.del_flag == 0, + ) + .with_for_update() + .one() + ) + + @staticmethod + def refresh_summary( + db: Session, + tenant_no: str, + owner_user_id: str, + owner_user_name: str, + invoice_count: int, + user_id: str, + dept_id: str, + ) -> BxInvoiceWalletSummary: + totals = InvoiceLedgerService.calc_summary(db, tenant_no, owner_user_id) + row = ( + db.query(BxInvoiceWalletSummary) + .filter( + BxInvoiceWalletSummary.tenant_no == tenant_no, + BxInvoiceWalletSummary.owner_user_id == owner_user_id, + BxInvoiceWalletSummary.del_flag == 0, + ) + .first() + ) + if row: + row.total_amount = totals["total_amount"] + row.used_amount = totals["used_amount"] + row.balance = totals["balance"] + row.invoice_count = invoice_count + if owner_user_name: + row.owner_user_name = owner_user_name + row.version = (row.version or 0) + 1 + row.update_by = user_id or None + else: + row = BxInvoiceWalletSummary( + id=generate_uuid(), + tenant_no=tenant_no, + owner_user_id=owner_user_id, + owner_user_name=owner_user_name or "", + total_amount=totals["total_amount"], + used_amount=totals["used_amount"], + balance=totals["balance"], + invoice_count=invoice_count, + version=1, + create_by=user_id or None, + create_dept_id=dept_id or "", + ) + db.add(row) + db.flush() + return row + + @staticmethod + def add_entry( + db: Session, + *, + tenant_no: str, + owner_user_id: str, + entry_type: str, + amount: Decimal, + user: BxUserContext, + invoice_id: Optional[str] = None, + reimburse_id: Optional[str] = None, + ref_ledger_id: Optional[str] = None, + remark: Optional[str] = None, + ) -> BxInvoiceLedger: + amt = Decimal(str(amount)) + if amt <= 0: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="金额必须大于0") + if entry_type not in _CREDIT_TYPES | _DEBIT_TYPES: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="无效的流水类型") + row = BxInvoiceLedger( + id=generate_uuid(), + tenant_no=tenant_no, + owner_user_id=owner_user_id, + entry_type=entry_type, + amount=amt, + invoice_id=invoice_id, + reimburse_id=reimburse_id, + ref_ledger_id=ref_ledger_id, + remark=remark, + create_by=user.id, + create_dept_id=user.dept_id or "", + ) + db.add(row) + db.flush() + return row + + @staticmethod + def debit_reimburse( + db: Session, + *, + tenant_no: str, + owner_user_id: str, + amount: Decimal, + reimburse_id: str, + user: BxUserContext, + ) -> BxInvoiceLedger: + amt = Decimal(str(amount)) + if amt <= 0: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="金额必须大于0") + summary_row = InvoiceLedgerService._lock_summary( + db, + tenant_no, + owner_user_id, + owner_user_name=user.name, + user_id=user.id, + dept_id=user.dept_id, + ) + totals = InvoiceLedgerService.calc_summary(db, tenant_no, owner_user_id) + balance = totals["balance"] + if balance < amt: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"发票额度不足(余额 {balance},本次 {amt})", + ) + entry = InvoiceLedgerService.add_entry( + db, + tenant_no=tenant_no, + owner_user_id=owner_user_id, + entry_type=ENTRY_DEBIT_REIMBURSE, + amount=amt, + user=user, + reimburse_id=reimburse_id, + ) + InvoiceLedgerService.refresh_summary( + db, + tenant_no, + owner_user_id, + summary_row.owner_user_name or user.name, + summary_row.invoice_count, + user.id, + user.dept_id, + ) + return entry + + @staticmethod + def release_reimburse( + db: Session, + *, + tenant_no: str, + reimburse_id: str, + user: BxUserContext, + ) -> Optional[BxInvoiceLedger]: + debit = ( + db.query(BxInvoiceLedger) + .filter( + BxInvoiceLedger.tenant_no == tenant_no, + BxInvoiceLedger.reimburse_id == reimburse_id, + BxInvoiceLedger.entry_type == ENTRY_DEBIT_REIMBURSE, + BxInvoiceLedger.del_flag == 0, + ) + .order_by(BxInvoiceLedger.create_time.desc()) + .first() + ) + release = ( + db.query(BxInvoiceLedger) + .filter( + BxInvoiceLedger.tenant_no == tenant_no, + BxInvoiceLedger.reimburse_id == reimburse_id, + BxInvoiceLedger.entry_type == ENTRY_CREDIT_RELEASE, + BxInvoiceLedger.del_flag == 0, + ) + .first() + ) + if should_skip_release(has_debit=debit is not None, has_release=release is not None): + return None + summary_row = InvoiceLedgerService._lock_summary( + db, + tenant_no, + debit.owner_user_id, + owner_user_name=user.name, + user_id=user.id, + dept_id=user.dept_id, + ) + entry = InvoiceLedgerService.add_entry( + db, + tenant_no=tenant_no, + owner_user_id=debit.owner_user_id, + entry_type=ENTRY_CREDIT_RELEASE, + amount=Decimal(str(debit.amount)), + user=user, + reimburse_id=reimburse_id, + ref_ledger_id=debit.id, + remark="报销退回释放额度", + ) + InvoiceLedgerService.refresh_summary( + db, + tenant_no, + debit.owner_user_id, + summary_row.owner_user_name or user.name, + summary_row.invoice_count, + user.id, + user.dept_id, + ) + return entry + + @staticmethod + def assert_can_delete_invoice( + db: Session, + tenant_no: str, + owner_user_id: str, + invoice_amount: Decimal, + ) -> None: + amt = Decimal(str(invoice_amount)) + totals = InvoiceLedgerService.calc_summary(db, tenant_no, owner_user_id) + if totals["balance"] - amt < 0: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"发票金额已部分或全部被报销占用,无法删除(余额 {totals['balance']},本次 {amt})", + ) diff --git a/backend/tests/test_bx_invoice_ledger.py b/backend/tests/test_bx_invoice_ledger.py new file mode 100644 index 000000000..bfc81168a --- /dev/null +++ b/backend/tests/test_bx_invoice_ledger.py @@ -0,0 +1,41 @@ +"""发票额度流水汇总与退回幂等。""" +from decimal import Decimal + +from app.services.bx.invoice_ledger import ( + ENTRY_CREDIT_RELEASE, + ENTRY_CREDIT_UPLOAD, + ENTRY_DEBIT_DELETE, + ENTRY_DEBIT_REIMBURSE, + should_skip_release, + summarize_entries, +) + + +def test_summarize_upload_and_reimburse(): + rows = [ + {"entry_type": ENTRY_CREDIT_UPLOAD, "amount": Decimal("1000")}, + {"entry_type": ENTRY_DEBIT_REIMBURSE, "amount": Decimal("300")}, + ] + s = summarize_entries(rows) + assert s["total_amount"] == Decimal("1000") + assert s["used_amount"] == Decimal("300") + assert s["balance"] == Decimal("700") + + +def test_summarize_delete_and_release(): + rows = [ + {"entry_type": ENTRY_CREDIT_UPLOAD, "amount": Decimal("500")}, + {"entry_type": ENTRY_DEBIT_DELETE, "amount": Decimal("100")}, + {"entry_type": ENTRY_DEBIT_REIMBURSE, "amount": Decimal("200")}, + {"entry_type": ENTRY_CREDIT_RELEASE, "amount": Decimal("200")}, + ] + s = summarize_entries(rows) + assert s["total_amount"] == Decimal("400") + assert s["used_amount"] == Decimal("0") + assert s["balance"] == Decimal("400") + + +def test_release_idempotent_guard(): + assert should_skip_release(has_debit=True, has_release=True) is True + assert should_skip_release(has_debit=True, has_release=False) is False + assert should_skip_release(has_debit=False, has_release=False) is True