feat(bx): invoice ledger balance service with unit tests

Co-authored-by: Cursor <cursoragent@cursor.com>

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

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}）",
+            )

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