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

from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from typing import Any, Dict, List, Optional, Sequence, Tuple

from fastapi import HTTPException, status
from sqlalchemy import and_, func, or_
from sqlalchemy.exc import IntegrityError
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.core.utils import generate_uuid
from app.models.bx_models import (
    BxInvoice,
    BxInvoiceLedger,
    BxReimburseApply,
    BxReimburseApproval,
    BxReimburseDetail,
    BxReimburseInvoice,
    BxReimburseOffset,
    BxWorkflowDef,
    BxWorkflowNode,
)
from app.models.models import SysUser
from app.schemas.bx.reimburse import ReimburseApplyOut, ReimburseDetailOut, ReimburseInvoiceBriefOut
from app.services.bx.approval_cycle import filter_current_cycle_approvals
from app.services.bx.attachment import AttachmentService
from app.services.bx.base import BxCrudService
from app.services.bx.invoice_ledger import ENTRY_DEBIT_REIMBURSE, InvoiceLedgerService
from app.services.bx_data_scope import apply_bx_scope
from app.services.web_auth_roles import iter_user_menu_perm_codes
from app.utils.bx_codegen import (
    allocate_reimburse_title_seq,
    generate_apply_code,
    format_reimburse_title,
    is_auto_reimburse_title,
    parse_reimburse_date,
    resolve_reimburse_title,
)
from app.utils.bx_date_range import current_month_bounds, resolve_create_time_bounds

REIMBURSE_CHANGE_APPLICANT_PERM = "bx:reimburseApply:changeApplicant"

_REIMBURSE_MY_STATS_PASSED_STATUSES = ("approved", "processed", "paid", "offset")


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

    @staticmethod
    def can_change_applicant(db: Session, user: BxUserContext) -> bool:
        """是否可选择他人作为报销申请人（超管或具备按钮权限）。"""
        if user.role == SUPER_ADMIN_ROLE:
            return True
        return REIMBURSE_CHANGE_APPLICANT_PERM in set(
            iter_user_menu_perm_codes(db, user.id, user.tenant_no)
        )

    @staticmethod
    def resolve_applicant(
        db: Session,
        user: BxUserContext,
        applicant_id: Optional[str],
        *,
        tenant_no: str,
    ) -> tuple[str, str]:
        """
        解析报销申请人。
        无 changeApplicant 权限时忽略客户端指定，强制本人；
        有权限时可指定同租户有效用户。
        返回 (applicant_id, applicant_name)。
        """
        requested = (applicant_id or "").strip() or None
        if not requested or requested == user.id:
            return user.id, (user.name or user.id)

        if not ReimburseService.can_change_applicant(db, user):
            return user.id, (user.name or user.id)

        target = (
            db.query(SysUser)
            .filter(
                SysUser.id == requested,
                SysUser.tenant_no == tenant_no,
                SysUser.del_flag == 0,
                SysUser.status == "1",
            )
            .first()
        )
        if not target:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="申请人不存在或已停用",
            )
        return target.id, (target.nick_name or target.user_no or target.id)

    @staticmethod
    def get_by_id(
        db: Session,
        reimburse_id: str,
        tenant_no: str,
        *,
        user: Optional[BxUserContext] = None,
        skip_scope: bool = False,
    ) -> Optional[BxReimburseApply]:
        return BxCrudService.get_by_id(
            BxReimburseApply, db, reimburse_id, tenant_no, user=user, skip_scope=skip_scope
        )

    @staticmethod
    def get_paged(
        db: Session,
        tenant_no: str,
        page: int,
        page_size: int,
        filters: Optional[List[Any]] = None,
        *,
        user: Optional[BxUserContext] = None,
    ) -> Tuple[Sequence[BxReimburseApply], int]:
        return BxCrudService.get_paged(
            BxReimburseApply, db, tenant_no, page, page_size, filters, user=user
        )

    @staticmethod
    def soft_delete(db: Session, reimburse_id: str, *, user: BxUserContext) -> bool:
        tenant_no = ReimburseService._tenant_no(db, user)
        return BxCrudService.soft_delete(
            BxReimburseApply, db, reimburse_id, tenant_no, user_id=user.id, user=user
        )

    @staticmethod
    def _workflow_nodes(db: Session, wf_id: str, tenant_no: str) -> List[BxWorkflowNode]:
        return (
            db.query(BxWorkflowNode)
            .filter(
                BxWorkflowNode.wf_id == wf_id,
                BxWorkflowNode.tenant_no == tenant_no,
                BxWorkflowNode.del_flag == 0,
            )
            .order_by(BxWorkflowNode.node_order)
            .all()
        )

    @staticmethod
    def pending_approval(
        db: Session,
        user: BxUserContext,
        tenant_no: str,
        page: int,
        page_size: int,
        keyword: str = "",
        applicant_id: Optional[str] = None,
        expense_name: Optional[str] = None,
        apply_date: Optional[str] = None,
        apply_date_from: Optional[str] = None,
        apply_date_to: Optional[str] = None,
    ):
        subquery = (
            db.query(BxReimburseApply.id)
            .join(
                BxWorkflowNode,
                and_(
                    BxWorkflowNode.wf_id == BxReimburseApply.wf_id,
                    BxWorkflowNode.node_name == BxReimburseApply.current_node,
                    BxWorkflowNode.tenant_no == tenant_no,
                    BxWorkflowNode.del_flag == 0,
                ),
            )
            .filter(
                BxReimburseApply.del_flag == 0,
                BxReimburseApply.tenant_no == tenant_no,
                BxReimburseApply.status == "pending",
                BxWorkflowNode.approver_id == user.id,
            )
        )
        if keyword:
            subquery = subquery.filter(
                or_(
                    BxReimburseApply.apply_code.contains(keyword),
                    BxReimburseApply.applicant_name.contains(keyword),
                    BxReimburseApply.title.contains(keyword),
                )
            )
        if applicant_id:
            subquery = subquery.filter(BxReimburseApply.applicant_id == applicant_id)
        expense_filter = ReimburseService.build_expense_name_condition(db, tenant_no, expense_name)
        if expense_filter is not None:
            subquery = subquery.filter(expense_filter)
        start, end = resolve_create_time_bounds(
            apply_date=apply_date,
            apply_date_from=apply_date_from,
            apply_date_to=apply_date_to,
        )
        if start is not None:
            subquery = subquery.filter(BxReimburseApply.create_time >= start)
        if end is not None:
            subquery = subquery.filter(BxReimburseApply.create_time < end)
        subquery = subquery.subquery()
        total = db.query(func.count()).select_from(subquery).scalar() or 0
        rows = (
            db.query(BxReimburseApply)
            .join(subquery, BxReimburseApply.id == subquery.c.id)
            .order_by(BxReimburseApply.create_time.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
        return rows, total

    @staticmethod
    def approved_by_me(
        db: Session,
        user: BxUserContext,
        tenant_no: str,
        page: int,
        page_size: int,
        keyword: str = "",
        applicant_id: Optional[str] = None,
        expense_name: Optional[str] = None,
        apply_date: Optional[str] = None,
        apply_date_from: Optional[str] = None,
        apply_date_to: Optional[str] = None,
    ):
        subquery = (
            db.query(BxReimburseApproval.apply_id)
            .filter(
                BxReimburseApproval.approver_id == user.id,
                BxReimburseApproval.tenant_no == tenant_no,
                BxReimburseApproval.del_flag == 0,
            )
            .distinct()
            .subquery()
        )
        query = (
            db.query(BxReimburseApply)
            .join(subquery, BxReimburseApply.id == subquery.c.apply_id)
            .filter(BxReimburseApply.del_flag == 0, BxReimburseApply.tenant_no == tenant_no)
        )
        if keyword:
            query = query.filter(
                or_(
                    BxReimburseApply.apply_code.contains(keyword),
                    BxReimburseApply.applicant_name.contains(keyword),
                    BxReimburseApply.title.contains(keyword),
                )
            )
        if applicant_id:
            query = query.filter(BxReimburseApply.applicant_id == applicant_id)
        expense_filter = ReimburseService.build_expense_name_condition(db, tenant_no, expense_name)
        if expense_filter is not None:
            query = query.filter(expense_filter)
        start, end = resolve_create_time_bounds(
            apply_date=apply_date,
            apply_date_from=apply_date_from,
            apply_date_to=apply_date_to,
        )
        if start is not None:
            query = query.filter(BxReimburseApply.create_time >= start)
        if end is not None:
            query = query.filter(BxReimburseApply.create_time < end)
        query = apply_bx_scope(query, BxReimburseApply, user, db, tenant_no)
        total = query.count()
        rows = (
            query.order_by(BxReimburseApply.create_time.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
        return rows, total

    @staticmethod
    def _has_debit_reimburse(db: Session, tenant_no: str, reimburse_id: str) -> bool:
        return (
            db.query(BxInvoiceLedger.id)
            .filter(
                BxInvoiceLedger.tenant_no == tenant_no,
                BxInvoiceLedger.reimburse_id == reimburse_id,
                BxInvoiceLedger.entry_type == ENTRY_DEBIT_REIMBURSE,
                BxInvoiceLedger.del_flag == 0,
            )
            .first()
            is not None
        )

    @staticmethod
    def sync_reimburse_invoices(
        db: Session,
        reimburse_id: str,
        invoice_ids: Optional[List[str]],
        tenant_no: str,
        user: BxUserContext,
        *,
        applicant_id: str,
    ) -> None:
        if invoice_ids is None:
            return
        normalized_ids: List[str] = []
        seen: set[str] = set()
        for raw_id in invoice_ids:
            invoice_id = str(raw_id or "").strip()
            if not invoice_id or invoice_id in seen:
                continue
            seen.add(invoice_id)
            normalized_ids.append(invoice_id)

        if normalized_ids:
            invoices = (
                db.query(BxInvoice)
                .filter(
                    BxInvoice.tenant_no == tenant_no,
                    BxInvoice.id.in_(normalized_ids),
                    BxInvoice.del_flag == 0,
                )
                .all()
            )
            invoice_map = {inv.id: inv for inv in invoices}
            for invoice_id in normalized_ids:
                invoice = invoice_map.get(invoice_id)
                if not invoice:
                    raise HTTPException(
                        status_code=status.HTTP_400_BAD_REQUEST,
                        detail=f"发票 {invoice_id} 不存在",
                    )
                if invoice.owner_user_id != applicant_id:
                    raise HTTPException(
                        status_code=status.HTTP_400_BAD_REQUEST,
                        detail=f"发票 {invoice_id} 不属于申请人",
                    )

        db.query(BxReimburseInvoice).filter(
            BxReimburseInvoice.reimburse_id == reimburse_id,
            BxReimburseInvoice.tenant_no == tenant_no,
            BxReimburseInvoice.del_flag == 0,
        ).update({"del_flag": 1, "update_by": user.id}, synchronize_session=False)

        for invoice_id in normalized_ids:
            db.add(
                BxReimburseInvoice(
                    id=generate_uuid(),
                    reimburse_id=reimburse_id,
                    invoice_id=invoice_id,
                    tenant_no=tenant_no,
                    create_by=user.id,
                    create_dept_id=user.dept_id or "",
                )
            )
        db.flush()

    @staticmethod
    def get_linked_invoice_ids(db: Session, reimburse_id: str, tenant_no: str) -> List[str]:
        rows = (
            db.query(BxReimburseInvoice.invoice_id)
            .filter(
                BxReimburseInvoice.reimburse_id == reimburse_id,
                BxReimburseInvoice.tenant_no == tenant_no,
                BxReimburseInvoice.del_flag == 0,
            )
            .order_by(BxReimburseInvoice.create_time.asc())
            .all()
        )
        return [row[0] for row in rows]

    @staticmethod
    def get_linked_invoices_brief(
        db: Session, reimburse_id: str, tenant_no: str
    ) -> List[ReimburseInvoiceBriefOut]:
        rows = (
            db.query(BxReimburseInvoice, BxInvoice)
            .join(
                BxInvoice,
                and_(
                    BxInvoice.id == BxReimburseInvoice.invoice_id,
                    BxInvoice.tenant_no == tenant_no,
                    BxInvoice.del_flag == 0,
                ),
            )
            .filter(
                BxReimburseInvoice.reimburse_id == reimburse_id,
                BxReimburseInvoice.tenant_no == tenant_no,
                BxReimburseInvoice.del_flag == 0,
            )
            .order_by(BxReimburseInvoice.create_time.asc())
            .all()
        )
        return [
            ReimburseInvoiceBriefOut(
                id=invoice.id,
                amount=float(invoice.amount),
                owner_user_name=invoice.owner_user_name,
            )
            for _, invoice in rows
        ]

    @staticmethod
    def get_linked_invoice_ids_map(
        db: Session, reimburse_ids: List[str], tenant_no: str
    ) -> Dict[str, List[str]]:
        if not reimburse_ids:
            return {}
        rows = (
            db.query(BxReimburseInvoice.reimburse_id, BxReimburseInvoice.invoice_id)
            .filter(
                BxReimburseInvoice.reimburse_id.in_(reimburse_ids),
                BxReimburseInvoice.tenant_no == tenant_no,
                BxReimburseInvoice.del_flag == 0,
            )
            .order_by(BxReimburseInvoice.create_time.asc())
            .all()
        )
        grouped: Dict[str, List[str]] = {}
        for reimburse_id, invoice_id in rows:
            grouped.setdefault(reimburse_id, []).append(invoice_id)
        return grouped

    @staticmethod
    def submit(
        db: Session,
        apply_id: str,
        user: BxUserContext,
        *,
        invoice_ids: Optional[List[str]] = None,
    ) -> Optional[BxReimburseApply]:
        tenant_no = ReimburseService._tenant_no(db, user)
        apply_obj = ReimburseService.get_by_id(db, apply_id, tenant_no, user=user)
        if not apply_obj:
            return None
        if invoice_ids is not None:
            ReimburseService.sync_reimburse_invoices(
                db,
                apply_id,
                invoice_ids,
                tenant_no,
                user,
                applicant_id=apply_obj.applicant_id,
            )
        if not ReimburseService._has_debit_reimburse(db, tenant_no, apply_id):
            amount = Decimal(str(apply_obj.amount or 0))
            InvoiceLedgerService.debit_reimburse(
                db,
                tenant_no=tenant_no,
                owner_user_id=apply_obj.applicant_id,
                amount=amount,
                reimburse_id=apply_id,
                user=user,
            )
        apply_obj.status = "pending"
        if not apply_obj.apply_code:
            apply_obj.apply_code = generate_apply_code(db, "BX", tenant_no)
        apply_obj.update_by = user.id
        if apply_obj.wf_id:
            nodes = ReimburseService._workflow_nodes(db, apply_obj.wf_id, tenant_no)
            if nodes:
                apply_obj.current_node = nodes[0].node_name
        db.flush()
        return apply_obj

    @staticmethod
    def approve(
        db: Session, apply_id: str, approver: BxUserContext, opinion: str
    ) -> Optional[BxReimburseApply]:
        tenant_no = ReimburseService._tenant_no(db, approver)
        apply_obj = ReimburseService.get_by_id(db, apply_id, tenant_no, skip_scope=True)
        if not apply_obj or not apply_obj.wf_id:
            return None
        nodes = ReimburseService._workflow_nodes(db, apply_obj.wf_id, tenant_no)
        if not nodes:
            return None
        current_idx = None
        for i, node in enumerate(nodes):
            if node.node_name == apply_obj.current_node:
                current_idx = i
                break
        if current_idx is None:
            return None
        current_node = nodes[current_idx]
        if current_node.approver_id != approver.id:
            return None
        approval = BxReimburseApproval(
            id=generate_uuid(),
            apply_id=apply_id,
            approver_id=approver.id,
            approver_name=approver.name,
            approval_opinion=opinion,
            node_name=current_node.node_name,
            node_order=current_node.node_order,
            approval_status="approved",
            approval_time=datetime.now(),
            tenant_no=tenant_no,
            create_by=approver.id,
            create_dept_id=approver.dept_id or "",
        )
        db.add(approval)
        if current_idx == len(nodes) - 1:
            apply_obj.status = "approved"
            apply_obj.current_node = None
        else:
            apply_obj.current_node = nodes[current_idx + 1].node_name
        apply_obj.update_by = approver.id
        db.flush()
        return apply_obj

    @staticmethod
    def reject(
        db: Session, apply_id: str, approver: BxUserContext, opinion: str
    ) -> Optional[BxReimburseApply]:
        tenant_no = ReimburseService._tenant_no(db, approver)
        apply_obj = ReimburseService.get_by_id(db, apply_id, tenant_no, skip_scope=True)
        if not apply_obj or not apply_obj.wf_id:
            return None
        nodes = ReimburseService._workflow_nodes(db, apply_obj.wf_id, tenant_no)
        current_node = None
        for node in nodes:
            if node.node_name == apply_obj.current_node:
                current_node = node
                break
        if current_node is None or current_node.approver_id != approver.id:
            return None
        approval = BxReimburseApproval(
            id=generate_uuid(),
            apply_id=apply_id,
            approver_id=approver.id,
            approver_name=approver.name,
            approval_opinion=opinion,
            node_name=current_node.node_name,
            node_order=current_node.node_order,
            approval_status="rejected",
            approval_time=datetime.now(),
            tenant_no=tenant_no,
            create_by=approver.id,
            create_dept_id=approver.dept_id or "",
        )
        db.add(approval)
        apply_obj.status = "rejected"
        apply_obj.current_node = None
        apply_obj.update_by = approver.id
        InvoiceLedgerService.release_reimburse(
            db, tenant_no=tenant_no, reimburse_id=apply_id, user=approver
        )
        db.flush()
        return apply_obj

    @staticmethod
    def withdraw(db: Session, apply_id: str, user: BxUserContext) -> Optional[BxReimburseApply]:
        tenant_no = ReimburseService._tenant_no(db, user)
        apply_obj = ReimburseService.get_by_id(db, apply_id, tenant_no, user=user)
        if not apply_obj:
            return None
        if apply_obj.applicant_id != user.id:
            return None
        if apply_obj.status != "pending":
            return None
        apply_obj.status = "draft"
        apply_obj.current_node = None
        apply_obj.update_by = user.id
        InvoiceLedgerService.release_reimburse(
            db, tenant_no=tenant_no, reimburse_id=apply_id, user=user
        )
        db.flush()
        return apply_obj

    @staticmethod
    def _approval_record_dict(approval: BxReimburseApproval, apply_id: str) -> Dict[str, Any]:
        return {
            "id": approval.id,
            "reimburse_id": apply_id,
            "node_name": approval.node_name,
            "node_order": approval.node_order,
            "approval_status": approval.approval_status,
            "approval_opinion": approval.approval_opinion,
            "approver_name": approval.approver_name,
            "approval_time": approval.approval_time,
        }

    @staticmethod
    def _latest_approval_per_node(
        approvals: Sequence[BxReimburseApproval],
    ) -> Dict[int, BxReimburseApproval]:
        """同一节点多次审批时，取 approval_time 最新的一条作为当前节点状态。"""
        approval_map: Dict[int, BxReimburseApproval] = {}
        for approval in approvals:
            existing = approval_map.get(approval.node_order)
            if existing is None:
                approval_map[approval.node_order] = approval
                continue
            approval_time = approval.approval_time or datetime.min
            existing_time = existing.approval_time or datetime.min
            if approval_time >= existing_time:
                approval_map[approval.node_order] = approval
        return approval_map

    @staticmethod
    def _query_approvals(db: Session, apply_id: str, tenant_no: str) -> List[BxReimburseApproval]:
        return (
            db.query(BxReimburseApproval)
            .filter(
                BxReimburseApproval.apply_id == apply_id,
                BxReimburseApproval.tenant_no == tenant_no,
                BxReimburseApproval.del_flag == 0,
            )
            .all()
        )

    @staticmethod
    def get_approvals(
        db: Session, apply_id: str, tenant_no: str, *, user: Optional[BxUserContext] = None
    ) -> List[Dict[str, Any]]:
        apply_obj = ReimburseService.get_by_id(db, apply_id, tenant_no, user=user)
        if not apply_obj or not apply_obj.wf_id:
            return []
        wf = BxCrudService.get_by_id(BxWorkflowDef, db, apply_obj.wf_id, tenant_no)
        if not wf:
            return []
        nodes = ReimburseService._workflow_nodes(db, wf.id, tenant_no)
        if not nodes:
            return []
        approvals = ReimburseService._query_approvals(db, apply_id, tenant_no)
        cycle_approvals = filter_current_cycle_approvals(approvals, apply_obj.status)
        approval_map = ReimburseService._latest_approval_per_node(cycle_approvals)
        result: List[Dict[str, Any]] = []
        for node in nodes:
            approval = approval_map.get(node.node_order)
            if approval:
                result.append(ReimburseService._approval_record_dict(approval, apply_id))
            else:
                result.append(
                    {
                        "id": "0",
                        "reimburse_id": apply_id,
                        "node_name": node.node_name,
                        "node_order": node.node_order,
                        "approval_status": "pending",
                        "approval_opinion": None,
                        "approver_name": node.approver_name or "",
                        "approval_time": None,
                    }
                )
        return result

    @staticmethod
    def get_approval_history(
        db: Session, apply_id: str, tenant_no: str, *, user: Optional[BxUserContext] = None
    ) -> List[Dict[str, Any]]:
        """返回全部审批动作，按时间正序排列，用于展示完整审批历史。"""
        if not ReimburseService.get_by_id(db, apply_id, tenant_no, user=user):
            return []
        approvals = (
            db.query(BxReimburseApproval)
            .filter(
                BxReimburseApproval.apply_id == apply_id,
                BxReimburseApproval.tenant_no == tenant_no,
                BxReimburseApproval.del_flag == 0,
            )
            .order_by(BxReimburseApproval.approval_time.asc(), BxReimburseApproval.create_time.asc())
            .all()
        )
        return [ReimburseService._approval_record_dict(a, apply_id) for a in approvals]

    @staticmethod
    def create_with_details(
        db: Session,
        obj_in: dict,
        details: List[dict],
        *,
        user: BxUserContext,
        applicant_id: Optional[str] = None,
        applicant_name: Optional[str] = None,
    ) -> BxReimburseApply:
        tenant_no = ReimburseService._tenant_no(db, user)
        data = dict(obj_in)
        total_amount = sum(float(d.get("amount", 0) or 0) for d in details)
        reimburse_date = parse_reimburse_date(data.pop("reimburse_date", None))
        requested_title = data.pop("title", None)
        first_detail = details[0] if details else {}
        resolved_id = applicant_id or user.id
        resolved_name = applicant_name or user.name

        apply_obj: BxReimburseApply | None = None
        for attempt in range(5):
            title, title_seq = resolve_reimburse_title(
                db, resolved_name, tenant_no, reimburse_date, requested_title
            )
            apply_obj = BxReimburseApply(
                id=generate_uuid(),
                title=title,
                reimburse_date=reimburse_date,
                title_seq=title_seq,
                amount=total_amount,
                expense_id=str(first_detail.get("expense_id") or ""),
                expense_name=first_detail.get("expense_name") or "",
                applicant_id=resolved_id,
                applicant_name=resolved_name,
                tenant_no=tenant_no,
                create_by=user.id,
                create_dept_id=user.dept_id or "",
                **data,
            )
            db.add(apply_obj)
            try:
                db.flush()
                break
            except IntegrityError:
                db.rollback()
                if attempt >= 4:
                    raise
                continue

        if apply_obj is None:
            raise RuntimeError("报销申请创建失败")

        for d in details:
            detail = BxReimburseDetail(
                id=generate_uuid(),
                reimburse_id=apply_obj.id,
                expense_id=str(d["expense_id"]),
                expense_name=d["expense_name"],
                amount=d["amount"],
                remark=d.get("remark"),
                tenant_no=tenant_no,
                create_by=user.id,
                create_dept_id=user.dept_id or "",
            )
            db.add(detail)
        db.flush()
        return apply_obj

    @staticmethod
    def update_with_details(
        db: Session,
        apply_id: str,
        obj_in: dict,
        details: Optional[List[dict]],
        *,
        user: BxUserContext,
    ) -> Optional[BxReimburseApply]:
        tenant_no = ReimburseService._tenant_no(db, user)
        apply_obj = ReimburseService.get_by_id(db, apply_id, tenant_no, user=user)
        if not apply_obj:
            return None

        reimburse_date = None
        date_changed = False
        if "reimburse_date" in obj_in:
            reimburse_date = parse_reimburse_date(obj_in.pop("reimburse_date"))
            date_changed = reimburse_date != apply_obj.reimburse_date

        requested_title = obj_in.pop("title", None)
        prev_applicant_id = apply_obj.applicant_id

        for key, value in obj_in.items():
            setattr(apply_obj, key, value)

        if reimburse_date is not None:
            apply_obj.reimburse_date = reimburse_date

        if requested_title is not None:
            apply_obj.title = str(requested_title).strip()

        applicant_changed = apply_obj.applicant_id != prev_applicant_id
        should_regen_title = (date_changed or applicant_changed) and is_auto_reimburse_title(
            apply_obj.title
        )
        if should_regen_title:
            seq = allocate_reimburse_title_seq(
                db, tenant_no, apply_obj.reimburse_date, exclude_apply_id=apply_id
            )
            apply_obj.title_seq = seq
            apply_obj.title = format_reimburse_title(
                apply_obj.applicant_name, apply_obj.reimburse_date, seq
            )
        elif requested_title is not None and not is_auto_reimburse_title(apply_obj.title):
            apply_obj.title_seq = None

        if details is not None:
            db.query(BxReimburseDetail).filter(
                BxReimburseDetail.reimburse_id == apply_id,
                BxReimburseDetail.tenant_no == tenant_no,
                BxReimburseDetail.del_flag == 0,
            ).update({"del_flag": 1}, synchronize_session=False)

            total_amount = Decimal("0")
            first_expense_name = ""
            first_expense_id = ""
            for i, d in enumerate(details):
                detail = BxReimburseDetail(
                    id=generate_uuid(),
                    reimburse_id=apply_id,
                    expense_id=str(d["expense_id"]),
                    expense_name=d["expense_name"],
                    amount=d["amount"],
                    remark=d.get("remark"),
                    tenant_no=tenant_no,
                    create_by=user.id,
                    create_dept_id=user.dept_id or "",
                )
                db.add(detail)
                total_amount += Decimal(str(d["amount"]))
                if i == 0:
                    first_expense_name = d["expense_name"]
                    first_expense_id = str(d["expense_id"])

            apply_obj.amount = total_amount
            apply_obj.expense_name = first_expense_name
            apply_obj.expense_id = first_expense_id

        apply_obj.update_by = user.id
        db.flush()
        return apply_obj

    @staticmethod
    def get_details(db: Session, reimburse_id: str, tenant_no: str) -> List[BxReimburseDetail]:
        return (
            db.query(BxReimburseDetail)
            .filter(
                BxReimburseDetail.reimburse_id == reimburse_id,
                BxReimburseDetail.tenant_no == tenant_no,
                BxReimburseDetail.del_flag == 0,
            )
            .order_by(BxReimburseDetail.create_time)
            .all()
        )

    @staticmethod
    def build_expense_name_condition(
        db: Session, tenant_no: str, expense_name: Optional[str]
    ):
        name_kw = str(expense_name or "").strip()
        if not name_kw:
            return None
        detail_reimburse_ids = (
            db.query(BxReimburseDetail.reimburse_id)
            .filter(
                BxReimburseDetail.tenant_no == tenant_no,
                BxReimburseDetail.del_flag == 0,
                BxReimburseDetail.expense_name.contains(name_kw),
            )
        )
        return or_(
            BxReimburseApply.expense_name.contains(name_kw),
            BxReimburseApply.id.in_(detail_reimburse_ids),
        )

    @staticmethod
    def get_details_map_by_reimburse_ids(
        db: Session, reimburse_ids: List[str], tenant_no: str
    ) -> Dict[str, List[BxReimburseDetail]]:
        if not reimburse_ids:
            return {}
        detail_rows = (
            db.query(BxReimburseDetail)
            .filter(
                BxReimburseDetail.reimburse_id.in_(reimburse_ids),
                BxReimburseDetail.tenant_no == tenant_no,
                BxReimburseDetail.del_flag == 0,
            )
            .order_by(BxReimburseDetail.create_time)
            .all()
        )
        grouped: Dict[str, List[BxReimburseDetail]] = {}
        for detail in detail_rows:
            grouped.setdefault(detail.reimburse_id, []).append(detail)
        return grouped

    @staticmethod
    def out_rows_with_attachment_counts(
        db: Session, tenant_no: str, rows: List[BxReimburseApply]
    ) -> List[ReimburseApplyOut]:
        if not rows:
            return []
        ids = [r.id for r in rows]
        counts = AttachmentService.count_invoice_attach_by_biz_ids(
            db, "reimburse_apply", ids, tenant_no
        )
        details_map = ReimburseService.get_details_map_by_reimburse_ids(db, ids, tenant_no)
        invoice_ids_map = ReimburseService.get_linked_invoice_ids_map(db, ids, tenant_no)
        out_list: List[ReimburseApplyOut] = []
        for row in rows:
            out = ReimburseApplyOut.model_validate(row)
            inv, att = counts.get(row.id, (0, 0))
            detail_rows = details_map.get(row.id, [])
            linked_ids = invoice_ids_map.get(row.id, [])
            out_list.append(
                out.model_copy(
                    update={
                        "invoice_count": inv,
                        "attach_count": att,
                        "details": [ReimburseDetailOut.model_validate(d) for d in detail_rows],
                        "invoice_ids": linked_ids,
                    }
                )
            )
        return out_list

    @staticmethod
    def out_one_with_attachment_counts(
        db: Session, tenant_no: str, row: BxReimburseApply
    ) -> ReimburseApplyOut:
        return ReimburseService.out_rows_with_attachment_counts(db, tenant_no, [row])[0]

    @staticmethod
    def build_apply_out_with_details(
        db: Session, tenant_no: str, reimburse: BxReimburseApply
    ) -> ReimburseApplyOut:
        out = ReimburseService.out_one_with_attachment_counts(db, tenant_no, reimburse)
        detail_rows = ReimburseService.get_details(db, reimburse.id, tenant_no)
        out.details = [ReimburseDetailOut.model_validate(d) for d in detail_rows]
        out.invoice_ids = ReimburseService.get_linked_invoice_ids(db, reimburse.id, tenant_no)
        out.invoices = ReimburseService.get_linked_invoices_brief(db, reimburse.id, tenant_no)
        return out

    @staticmethod
    def global_status_counts(db: Session, tenant_no: str, *, user: BxUserContext) -> Dict[str, int]:
        query = db.query(BxReimburseApply.status, func.count()).filter(
            BxReimburseApply.del_flag == 0, BxReimburseApply.tenant_no == tenant_no
        )
        query = apply_bx_scope(query, BxReimburseApply, user, db, tenant_no)
        rows = query.group_by(BxReimburseApply.status).all()
        return {str(s or ""): int(c) for s, c in rows}

    @staticmethod
    def _my_apply_base(db: Session, user: BxUserContext, tenant_no: str):
        query = db.query(BxReimburseApply).filter(
            BxReimburseApply.del_flag == 0,
            BxReimburseApply.tenant_no == tenant_no,
            or_(
                BxReimburseApply.applicant_id == user.id,
                BxReimburseApply.create_by == user.id,
            ),
        )
        return apply_bx_scope(query, BxReimburseApply, user, db, tenant_no)

    @staticmethod
    def my_apply_stats(db: Session, user: BxUserContext, tenant_no: str) -> dict:
        base_q = ReimburseService._my_apply_base(db, user, tenant_no).subquery()
        base = BxReimburseApply.id.in_(db.query(base_q.c.id))
        total_count = (
            db.query(func.count()).select_from(BxReimburseApply).filter(base).scalar() or 0
        )
        pending = (
            db.query(func.count())
            .select_from(BxReimburseApply)
            .filter(base, BxReimburseApply.status == "pending")
            .scalar()
            or 0
        )
        approved_count = (
            db.query(func.count())
            .select_from(BxReimburseApply)
            .filter(base, BxReimburseApply.status.in_(_REIMBURSE_MY_STATS_PASSED_STATUSES))
            .scalar()
            or 0
        )
        rejected_count = (
            db.query(func.count())
            .select_from(BxReimburseApply)
            .filter(base, BxReimburseApply.status == "rejected")
            .scalar()
            or 0
        )
        approved_amount = (
            db.query(func.coalesce(func.sum(BxReimburseApply.amount), 0))
            .filter(base, BxReimburseApply.status.in_(_REIMBURSE_MY_STATS_PASSED_STATUSES))
            .scalar()
            or 0
        )
        m0, m1 = current_month_bounds()
        month_apply_amount = (
            db.query(func.coalesce(func.sum(BxReimburseApply.amount), 0))
            .filter(base, BxReimburseApply.create_time >= m0, BxReimburseApply.create_time < m1)
            .scalar()
            or 0
        )
        month_apply_count = (
            db.query(func.count())
            .select_from(BxReimburseApply)
            .filter(base, BxReimburseApply.create_time >= m0, BxReimburseApply.create_time < m1)
            .scalar()
            or 0
        )
        return {
            "total_count": int(total_count),
            "pending_count": int(pending),
            "approved_count": int(approved_count),
            "rejected_count": int(rejected_count),
            "approved_amount": float(approved_amount),
            "month_apply_amount": float(month_apply_amount),
            "month_apply_count": int(month_apply_count),
        }

    @staticmethod
    def pending_approval_stats(db: Session, user: BxUserContext, tenant_no: str) -> dict:
        subquery = (
            db.query(BxReimburseApply.id)
            .join(
                BxWorkflowNode,
                and_(
                    BxWorkflowNode.wf_id == BxReimburseApply.wf_id,
                    BxWorkflowNode.node_name == BxReimburseApply.current_node,
                    BxWorkflowNode.tenant_no == tenant_no,
                    BxWorkflowNode.del_flag == 0,
                ),
            )
            .filter(
                BxReimburseApply.del_flag == 0,
                BxReimburseApply.tenant_no == tenant_no,
                BxReimburseApply.status == "pending",
                BxWorkflowNode.approver_id == user.id,
            )
        ).subquery()
        total = db.query(func.count()).select_from(subquery).scalar() or 0
        m0, m1 = current_month_bounds()
        month_sum = (
            db.query(func.coalesce(func.sum(BxReimburseApply.amount), 0))
            .join(subquery, BxReimburseApply.id == subquery.c.id)
            .filter(BxReimburseApply.create_time >= m0, BxReimburseApply.create_time < m1)
            .scalar()
            or 0
        )
        return {"pending_total": int(total), "month_pending_amount": float(month_sum)}

    @staticmethod
    def pay(
        db: Session,
        apply_id: str,
        user: BxUserContext,
        pay_remark: str = "",
        actual_amount: Optional[float] = None,
    ) -> Optional[BxReimburseApply]:
        tenant_no = ReimburseService._tenant_no(db, user)
        apply_obj = ReimburseService.get_by_id(db, apply_id, tenant_no, user=user)
        if not apply_obj:
            return None
        apply_obj.status = "processed"
        apply_obj.update_by = user.id
        apply_obj.pay_status = "paid"
        apply_obj.pay_time = datetime.now()
        apply_obj.pay_user_id = user.id
        apply_obj.pay_user_name = user.name
        if actual_amount is not None:
            apply_obj.actual_amount = actual_amount
        db.flush()
        return apply_obj

    @staticmethod
    def offset(
        db: Session,
        apply_id: str,
        advance_id: str,
        amount: float,
        user: BxUserContext,
        remark: str = "",
    ) -> Optional[BxReimburseApply]:
        tenant_no = ReimburseService._tenant_no(db, user)
        apply_obj = ReimburseService.get_by_id(db, apply_id, tenant_no, user=user)
        if not apply_obj:
            return None
        apply_obj.status = "offset"
        apply_obj.offset_advance_id = advance_id
        apply_obj.offset_amount = amount
        apply_obj.update_by = user.id
        offset_record = BxReimburseOffset(
            id=generate_uuid(),
            reimburse_id=apply_id,
            advance_id=advance_id,
            offset_amount=amount,
            remark=remark or None,
            tenant_no=tenant_no,
            create_by=user.name,
            create_dept_id=user.dept_id or "",
        )
        db.add(offset_record)
        db.flush()
        return apply_obj

    @staticmethod
    def batch_offset(
        db: Session,
        apply_id: str,
        items: List[dict],
        pay_amount: float,
        user: BxUserContext,
        remark: str = "",
    ) -> Optional[BxReimburseApply]:
        tenant_no = ReimburseService._tenant_no(db, user)
        apply_obj = ReimburseService.get_by_id(db, apply_id, tenant_no, user=user)
        if not apply_obj:
            return None

        has_offsets = bool(items)
        total_offset = 0.0
        first_advance_id = None

        for item in items:
            advance_id = str(item["advance_id"])
            offset_amount = float(item["offset_amount"])
            if first_advance_id is None:
                first_advance_id = advance_id
            total_offset += offset_amount
            offset_record = BxReimburseOffset(
                id=generate_uuid(),
                reimburse_id=apply_id,
                advance_id=advance_id,
                offset_amount=offset_amount,
                remark=remark or None,
                tenant_no=tenant_no,
                create_by=user.name,
                create_dept_id=user.dept_id or "",
            )
            db.add(offset_record)

        if first_advance_id is not None:
            apply_obj.offset_advance_id = first_advance_id
            apply_obj.offset_amount = total_offset

        if pay_amount > 0:
            apply_obj.pay_status = "paid"
            apply_obj.pay_time = datetime.now()
            apply_obj.pay_user_id = user.id
            apply_obj.pay_user_name = user.name
            apply_obj.actual_amount = pay_amount
            apply_obj.status = "processed"
        elif has_offsets:
            apply_obj.actual_amount = 0
            apply_obj.status = "processed"

        apply_obj.update_by = user.id
        db.flush()
        return apply_obj

    @staticmethod
    def finance_payment_stats(db: Session, tenant_no: str, *, user: BxUserContext) -> dict:
        base_q = (
            db.query(BxReimburseApply)
            .filter(BxReimburseApply.del_flag == 0, BxReimburseApply.tenant_no == tenant_no)
        )
        base_q = apply_bx_scope(base_q, BxReimburseApply, user, db, tenant_no).subquery()
        base = BxReimburseApply.id.in_(db.query(base_q.c.id))
        pending = (
            db.query(func.count())
            .select_from(BxReimburseApply)
            .filter(base, BxReimburseApply.status == "approved")
            .scalar()
            or 0
        )
        bank_paid = BxReimburseApply.pay_time.isnot(None)
        paid = (
            db.query(func.count()).select_from(BxReimburseApply).filter(base, bank_paid).scalar() or 0
        )
        offset_settled = or_(
            BxReimburseApply.status == "offset",
            and_(
                BxReimburseApply.status == "processed",
                BxReimburseApply.pay_time.is_(None),
                func.coalesce(BxReimburseApply.offset_amount, 0) > 0,
            ),
        )
        offset = (
            db.query(func.count()).select_from(BxReimburseApply).filter(base, offset_settled).scalar() or 0
        )
        m0, m1 = current_month_bounds()
        month_paid = (
            db.query(func.coalesce(func.sum(BxReimburseApply.amount), 0))
            .filter(base, bank_paid, BxReimburseApply.pay_time >= m0, BxReimburseApply.pay_time < m1)
            .scalar()
            or 0
        )
        month_offset = (
            db.query(func.coalesce(func.sum(BxReimburseApply.amount), 0))
            .filter(base, offset_settled, BxReimburseApply.update_time >= m0, BxReimburseApply.update_time < m1)
            .scalar()
            or 0
        )
        return {
            "pending_count": int(pending),
            "paid_count": int(paid),
            "offset_count": int(offset),
            "month_processed_amount": float(month_paid) + float(month_offset),
        }