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"
REIMBURSE_CHANGE_APPLICANT_PERM = "bx:reimburseApply:changeApplicant"


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 can_query_others_wallet(db: Session, user: BxUserContext) -> bool:
        """发票夹管理 或 报销改选申请人 权限可查询他人发票/余额。"""
        if InvoiceService.is_wallet_admin(db, user):
            return True
        codes = iter_user_menu_perm_codes(db, user.id, user.tenant_no)
        return REIMBURSE_CHANGE_APPLICANT_PERM in set(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 and not InvoiceService.can_query_others_wallet(db, user):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="无权查看他人发票夹",
            )
        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,
        owner_user_id: Optional[str] = None,
    ) -> dict:
        tenant_no = InvoiceService._tenant_no(db, user)
        owner_id = InvoiceService._resolve_owner_for_query(db, user, owner_user_id)
        owner_name = (
            user.name
            if owner_id == user.id
            else InvoiceService._lookup_user_name(db, owner_id)
        )
        row = (
            db.query(BxInvoiceWalletSummary)
            .filter(
                BxInvoiceWalletSummary.tenant_no == tenant_no,
                BxInvoiceWalletSummary.owner_user_id == owner_id,
                BxInvoiceWalletSummary.del_flag == 0,
            )
            .first()
        )
        if row:
            return InvoiceService._summary_to_dict(row)
        totals = InvoiceLedgerService.calc_summary(db, tenant_no, owner_id)
        return {
            "id": "",
            "owner_user_id": owner_id,
            "owner_user_name": owner_name,
            "total_amount": totals["total_amount"],
            "used_amount": totals["used_amount"],
            "balance": totals["balance"],
            "invoice_count": InvoiceService._count_invoices(db, tenant_no, owner_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[dict]:
        """返回流水字典列表；流水备注为空时回填关联发票备注（含已软删发票）。"""
        tenant_no = InvoiceService._tenant_no(db, user)
        owner_id = InvoiceService._resolve_owner_for_query(db, user, owner_user_id)
        rows = list(reversed(InvoiceLedgerService.list_entries(db, tenant_no, owner_id)))
        invoice_ids = {r.invoice_id for r in rows if r.invoice_id}
        inv_remark_map: dict[str, str] = {}
        if invoice_ids:
            inv_rows = (
                db.query(BxInvoice.id, BxInvoice.remark)
                .filter(BxInvoice.id.in_(list(invoice_ids)), BxInvoice.tenant_no == tenant_no)
                .all()
            )
            for inv_id, inv_remark in inv_rows:
                text = (inv_remark or "").strip()
                if text:
                    inv_remark_map[inv_id] = text
        out: List[dict] = []
        for r in rows:
            remark = (r.remark or "").strip()
            if not remark and r.invoice_id:
                remark = inv_remark_map.get(r.invoice_id, "")
            out.append(
                {
                    "id": r.id,
                    "owner_user_id": r.owner_user_id,
                    "entry_type": r.entry_type,
                    "amount": r.amount,
                    "invoice_id": r.invoice_id,
                    "reimburse_id": r.reimburse_id,
                    "ref_ledger_id": r.ref_ledger_id,
                    "remark": remark or None,
                    "create_time": r.create_time,
                }
            )
        return out

    @staticmethod
    def create(
        db: Session,
        *,
        user: BxUserContext,
        amount: Decimal,
        files: Sequence[Tuple[bytes, str, str]],
        remark: Optional[str] = None,
        target_user_id: Optional[str] = None,
    ) -> BxInvoice:
        """一次上传创建一条发票档案：金额为总额，files 全部挂到同一 biz_id，只写一条 credit_upload。"""
        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 files:
            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,
        )

        for file_data, filename, content_type in files:
            if not file_data:
                raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="文件内容为空")
            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,
            remark=(remark or "").strip() or None,
        )
        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,
            remark=(invoice.remark or "").strip() or "删除发票冲减",
        )

        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, List[BxAttachment]] = {}
        for att in attachments:
            att_map.setdefault(att.biz_id, []).append(att)

        out: List[dict] = []
        for inv in invoices:
            atts = att_map.get(inv.id, [])
            first = atts[0] if atts else None
            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": first.id if first else None,
                    "file_name": first.file_name if first else None,
                    "file_path": first.file_path if first else None,
                    "file_type": first.file_type if first else None,
                    "attachments": [
                        {
                            "id": a.id,
                            "file_name": a.file_name,
                            "file_path": a.file_path,
                            "file_type": a.file_type,
                        }
                        for a in atts
                    ],
                }
            )
        return out