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

from __future__ import annotations

import mimetypes
from decimal import Decimal
from typing import List, Optional, Sequence, Tuple

from fastapi import HTTPException, status
from sqlalchemy import or_
from sqlalchemy.orm import Session

from app.core.bx_auth import BxUserContext
from app.core.menu_guard import SUPER_ADMIN_ROLE
from app.core.tenant_context import get_tenant_no
from app.models.bx_models import BxAttachment, BxInvoice, BxInvoiceLedger, BxInvoiceWalletSummary
from app.models.models import SysUser
from app.services.bx.attachment import AttachmentService
from app.services.bx.base import BxCrudService
from app.services.bx.invoice_ledger import (
    ENTRY_CREDIT_UPLOAD,
    ENTRY_DEBIT_DELETE,
    InvoiceLedgerService,
)
from app.services.bx.storage import upload_bx_file
from app.services.web_auth_roles import iter_user_menu_perm_codes

INVOICE_WALLET_ADMIN_PERM = "bx:invoiceWallet:admin"


class InvoiceService:
    @staticmethod
    def _tenant_no(db: Session, user: BxUserContext) -> str:
        return get_tenant_no(db, user.to_tenant_user_dict())

    @staticmethod
    def is_wallet_admin(db: Session, user: BxUserContext) -> bool:
        if user.role == SUPER_ADMIN_ROLE:
            return True
        codes = iter_user_menu_perm_codes(db, user.id, user.tenant_no)
        return INVOICE_WALLET_ADMIN_PERM in codes

    @staticmethod
    def _require_wallet_admin(db: Session, user: BxUserContext) -> None:
        if not InvoiceService.is_wallet_admin(db, user):
            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="需要发票夹管理权限")

    @staticmethod
    def _lookup_user_name(db: Session, user_id: str) -> str:
        row = (
            db.query(SysUser)
            .filter(SysUser.id == user_id, SysUser.del_flag == 0)
            .first()
        )
        if not row:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="目标用户不存在")
        return (row.nick_name or row.user_no or "").strip()

    @staticmethod
    def _resolve_owner(
        db: Session,
        user: BxUserContext,
        target_user_id: Optional[str],
    ) -> Tuple[str, str]:
        target = (target_user_id or "").strip()
        if not target or target == user.id:
            return user.id, user.name
        InvoiceService._require_wallet_admin(db, user)
        return target, InvoiceService._lookup_user_name(db, target)

    @staticmethod
    def _resolve_owner_for_query(
        db: Session,
        user: BxUserContext,
        user_id: Optional[str],
    ) -> str:
        owner_id = (user_id or "").strip() or user.id
        if owner_id != user.id:
            InvoiceService._require_wallet_admin(db, user)
        return owner_id

    @staticmethod
    def _count_invoices(db: Session, tenant_no: str, owner_user_id: str) -> int:
        return (
            db.query(BxInvoice)
            .filter(
                BxInvoice.tenant_no == tenant_no,
                BxInvoice.owner_user_id == owner_user_id,
                BxInvoice.del_flag == 0,
            )
            .count()
        )

    @staticmethod
    def get_mine_summary(db: Session, *, user: BxUserContext) -> dict:
        tenant_no = InvoiceService._tenant_no(db, user)
        row = (
            db.query(BxInvoiceWalletSummary)
            .filter(
                BxInvoiceWalletSummary.tenant_no == tenant_no,
                BxInvoiceWalletSummary.owner_user_id == user.id,
                BxInvoiceWalletSummary.del_flag == 0,
            )
            .first()
        )
        if row:
            return InvoiceService._summary_to_dict(row)
        totals = InvoiceLedgerService.calc_summary(db, tenant_no, user.id)
        return {
            "id": "",
            "owner_user_id": user.id,
            "owner_user_name": user.name,
            "total_amount": totals["total_amount"],
            "used_amount": totals["used_amount"],
            "balance": totals["balance"],
            "invoice_count": InvoiceService._count_invoices(db, tenant_no, user.id),
            "version": 0,
            "create_time": None,
            "update_time": None,
        }

    @staticmethod
    def list_wallets(
        db: Session,
        *,
        user: BxUserContext,
        page: int = 1,
        page_size: int = 10,
        keyword: str = "",
    ) -> Tuple[Sequence[BxInvoiceWalletSummary], int]:
        InvoiceService._require_wallet_admin(db, user)
        tenant_no = InvoiceService._tenant_no(db, user)
        query = db.query(BxInvoiceWalletSummary).filter(
            BxInvoiceWalletSummary.tenant_no == tenant_no,
            BxInvoiceWalletSummary.del_flag == 0,
        )
        kw = (keyword or "").strip()
        if kw:
            query = query.filter(
                or_(
                    BxInvoiceWalletSummary.owner_user_name.contains(kw),
                    BxInvoiceWalletSummary.owner_user_id.contains(kw),
                )
            )
        total = query.count()
        rows = (
            query.order_by(BxInvoiceWalletSummary.update_time.desc(), BxInvoiceWalletSummary.create_time.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
        return rows, total

    @staticmethod
    def list_invoices(
        db: Session,
        *,
        user: BxUserContext,
        owner_user_id: Optional[str] = None,
        page: int = 1,
        page_size: int = 10,
    ) -> Tuple[Sequence[BxInvoice], int]:
        tenant_no = InvoiceService._tenant_no(db, user)
        owner_id = InvoiceService._resolve_owner_for_query(db, user, owner_user_id)
        filters = [BxInvoice.owner_user_id == owner_id]
        rows, total = BxCrudService.get_paged(
            BxInvoice,
            db,
            tenant_no,
            page,
            page_size,
            filters=filters,
            user=user,
        )
        return rows, total

    @staticmethod
    def list_ledgers(
        db: Session,
        *,
        user: BxUserContext,
        owner_user_id: Optional[str] = None,
    ) -> List[BxInvoiceLedger]:
        tenant_no = InvoiceService._tenant_no(db, user)
        owner_id = InvoiceService._resolve_owner_for_query(db, user, owner_user_id)
        rows = InvoiceLedgerService.list_entries(db, tenant_no, owner_id)
        return list(reversed(rows))

    @staticmethod
    def create(
        db: Session,
        *,
        user: BxUserContext,
        amount: Decimal,
        file_data: bytes,
        filename: str,
        content_type: str,
        remark: Optional[str] = None,
        target_user_id: Optional[str] = None,
    ) -> BxInvoice:
        tenant_no = InvoiceService._tenant_no(db, user)
        owner_id, owner_name = InvoiceService._resolve_owner(db, user, target_user_id)
        amt = Decimal(str(amount))
        if amt <= 0:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="金额必须大于0")
        if not file_data:
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请选择文件")

        invoice = BxCrudService.create(
            BxInvoice,
            db,
            {
                "owner_user_id": owner_id,
                "owner_user_name": owner_name,
                "amount": amt,
                "upload_by": user.id,
                "upload_by_name": user.name,
                "remark": (remark or "").strip() or None,
            },
            tenant_no=tenant_no,
            user_id=user.id,
            dept_id=user.dept_id,
        )

        safe_name = filename or "unknown"
        mime = content_type or mimetypes.guess_type(safe_name)[0] or "application/octet-stream"
        relative_path, _full_url = upload_bx_file(file_data, safe_name, biz_type="invoice")
        AttachmentService.create(
            db,
            {
                "biz_type": "invoice",
                "biz_id": invoice.id,
                "file_name": safe_name,
                "file_path": relative_path,
                "file_type": mime,
                "file_size": len(file_data),
                "attach_type": "invoice",
            },
            user=user,
        )

        InvoiceLedgerService.add_entry(
            db,
            tenant_no=tenant_no,
            owner_user_id=owner_id,
            entry_type=ENTRY_CREDIT_UPLOAD,
            amount=amt,
            user=user,
            invoice_id=invoice.id,
        )
        invoice_count = InvoiceService._count_invoices(db, tenant_no, owner_id)
        InvoiceLedgerService.refresh_summary(
            db,
            tenant_no,
            owner_id,
            owner_name,
            invoice_count,
            user.id,
            user.dept_id,
        )
        return invoice

    @staticmethod
    def delete(db: Session, *, user: BxUserContext, invoice_id: str) -> bool:
        InvoiceService._require_wallet_admin(db, user)
        tenant_no = InvoiceService._tenant_no(db, user)
        invoice = BxCrudService.get_by_id(
            BxInvoice, db, invoice_id, tenant_no, user=user, skip_scope=True
        )
        if not invoice:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="发票不存在")

        InvoiceLedgerService.assert_can_delete_invoice(
            db, tenant_no, invoice.owner_user_id, invoice.amount
        )
        InvoiceLedgerService.add_entry(
            db,
            tenant_no=tenant_no,
            owner_user_id=invoice.owner_user_id,
            entry_type=ENTRY_DEBIT_DELETE,
            amount=Decimal(str(invoice.amount)),
            user=user,
            invoice_id=invoice.id,
        )

        invoice.del_flag = 1
        invoice.update_by = user.id
        db.query(BxAttachment).filter(
            BxAttachment.biz_type == "invoice",
            BxAttachment.biz_id == invoice.id,
            BxAttachment.tenant_no == tenant_no,
            BxAttachment.del_flag == 0,
        ).update(
            {BxAttachment.del_flag: 1, BxAttachment.update_by: user.id},
            synchronize_session="fetch",
        )
        db.flush()

        invoice_count = InvoiceService._count_invoices(db, tenant_no, invoice.owner_user_id)
        InvoiceLedgerService.refresh_summary(
            db,
            tenant_no,
            invoice.owner_user_id,
            invoice.owner_user_name,
            invoice_count,
            user.id,
            user.dept_id,
        )
        return True

    @staticmethod
    def _summary_to_dict(row: BxInvoiceWalletSummary) -> dict:
        return {
            "id": row.id,
            "owner_user_id": row.owner_user_id,
            "owner_user_name": row.owner_user_name,
            "total_amount": row.total_amount,
            "used_amount": row.used_amount,
            "balance": row.balance,
            "invoice_count": row.invoice_count,
            "version": row.version,
            "create_time": row.create_time,
            "update_time": row.update_time,
        }

    @staticmethod
    def out_rows_with_attachments(
        db: Session, tenant_no: str, invoices: Sequence[BxInvoice]
    ) -> List[dict]:
        if not invoices:
            return []
        ids = [inv.id for inv in invoices]
        attachments = (
            db.query(BxAttachment)
            .filter(
                BxAttachment.biz_type == "invoice",
                BxAttachment.biz_id.in_(ids),
                BxAttachment.tenant_no == tenant_no,
                BxAttachment.del_flag == 0,
            )
            .order_by(BxAttachment.create_time.asc())
            .all()
        )
        att_map: dict[str, BxAttachment] = {}
        for att in attachments:
            if att.biz_id not in att_map:
                att_map[att.biz_id] = att

        out: List[dict] = []
        for inv in invoices:
            att = att_map.get(inv.id)
            out.append(
                {
                    "id": inv.id,
                    "owner_user_id": inv.owner_user_id,
                    "owner_user_name": inv.owner_user_name,
                    "amount": inv.amount,
                    "upload_by": inv.upload_by,
                    "upload_by_name": inv.upload_by_name,
                    "remark": inv.remark,
                    "create_time": inv.create_time,
                    "update_time": inv.update_time,
                    "attachment_id": att.id if att else None,
                    "file_name": att.file_name if att else None,
                    "file_path": att.file_path if att else None,
                    "file_type": att.file_type if att else None,
                }
            )
        return out