shanxia2.0/backend/app/services/bx/invoice_ledger.py

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