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

from __future__ import annotations

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

from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session

from app.core.bx_auth import BxUserContext
from app.core.tenant_context import get_tenant_no
from app.core.utils import generate_uuid
from app.models.bx_models import (
    BxAdvanceApply,
    BxAdvanceApproval,
    BxReimburseOffset,
    BxWorkflowDef,
    BxWorkflowNode,
)
from app.services.bx.approval_cycle import filter_current_cycle_approvals
from app.services.bx.base import BxCrudService
from app.services.bx_data_scope import apply_bx_scope
from app.utils.bx_codegen import generate_apply_code
from app.utils.bx_date_range import append_create_time_filters, current_month_bounds, resolve_create_time_bounds


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

    @staticmethod
    def get_by_id(
        db: Session,
        advance_id: str,
        tenant_no: str,
        *,
        user: Optional[BxUserContext] = None,
        skip_scope: bool = False,
    ) -> Optional[BxAdvanceApply]:
        return BxCrudService.get_by_id(
            BxAdvanceApply, db, advance_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,
        order_by: Optional[Any] = None,
    ) -> Tuple[Sequence[BxAdvanceApply], int]:
        return BxCrudService.get_paged(
            BxAdvanceApply,
            db,
            tenant_no,
            page,
            page_size,
            filters,
            user=user,
            order_by=order_by,
        )

    @staticmethod
    def create(
        db: Session,
        obj_in: dict,
        *,
        user: BxUserContext,
    ) -> BxAdvanceApply:
        from sqlalchemy.exc import IntegrityError

        tenant_no = AdvanceService._tenant_no(db, user)
        data = dict(obj_in)
        data.pop("apply_code", None)
        last_error: Exception | None = None
        for attempt in range(5):
            payload = {
                **data,
                "apply_code": generate_apply_code(db, "BYJ", tenant_no),
            }
            try:
                return BxCrudService.create(
                    BxAdvanceApply,
                    db,
                    payload,
                    tenant_no=tenant_no,
                    user_id=user.id,
                    dept_id=user.dept_id,
                )
            except IntegrityError as exc:
                db.rollback()
                last_error = exc
                if attempt >= 4:
                    raise
                continue
        if last_error:
            raise last_error
        raise RuntimeError("备用金申请创建失败")

    @staticmethod
    def update(
        db: Session,
        advance_id: str,
        obj_in: dict,
        *,
        user: BxUserContext,
    ) -> Optional[BxAdvanceApply]:
        tenant_no = AdvanceService._tenant_no(db, user)
        return BxCrudService.update(
            BxAdvanceApply,
            db,
            advance_id,
            tenant_no,
            obj_in,
            user_id=user.id,
            user=user,
        )

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

    @staticmethod
    def pending_approval(
        db: Session,
        user: BxUserContext,
        tenant_no: str,
        page: int,
        page_size: int,
        keyword: str = "",
        applicant_id: Optional[str] = None,
        apply_date: Optional[str] = None,
        apply_date_from: Optional[str] = None,
        apply_date_to: Optional[str] = None,
        extra_filters: Optional[list] = None,
        order_by: Optional[Any] = None,
    ):
        subquery = (
            db.query(BxAdvanceApply.id)
            .join(
                BxWorkflowNode,
                and_(
                    BxWorkflowNode.wf_id == BxAdvanceApply.wf_id,
                    BxWorkflowNode.node_name == BxAdvanceApply.current_node,
                    BxWorkflowNode.tenant_no == tenant_no,
                    BxWorkflowNode.del_flag == 0,
                ),
            )
            .filter(
                BxAdvanceApply.del_flag == 0,
                BxAdvanceApply.tenant_no == tenant_no,
                BxAdvanceApply.status == "pending",
                BxWorkflowNode.approver_id == user.id,
            )
        )
        if extra_filters:
            subquery = subquery.filter(*extra_filters)
        else:
            if keyword:
                subquery = subquery.filter(
                    or_(
                        BxAdvanceApply.apply_code.contains(keyword),
                        BxAdvanceApply.applicant_name.contains(keyword),
                    )
                )
            if applicant_id:
                subquery = subquery.filter(BxAdvanceApply.applicant_id == applicant_id)
            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(BxAdvanceApply.create_time >= start)
            if end is not None:
                subquery = subquery.filter(BxAdvanceApply.create_time < end)
        subquery = subquery.subquery()
        total = db.query(func.count()).select_from(subquery).scalar() or 0
        order_expr = order_by if order_by is not None else BxAdvanceApply.create_time.desc()
        rows = (
            db.query(BxAdvanceApply)
            .join(subquery, BxAdvanceApply.id == subquery.c.id)
            .order_by(order_expr)
            .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,
        apply_date: Optional[str] = None,
        apply_date_from: Optional[str] = None,
        apply_date_to: Optional[str] = None,
        extra_filters: Optional[list] = None,
        order_by: Optional[Any] = None,
    ):
        subquery = (
            db.query(BxAdvanceApproval.apply_id)
            .filter(
                BxAdvanceApproval.approver_id == user.id,
                BxAdvanceApproval.tenant_no == tenant_no,
                BxAdvanceApproval.del_flag == 0,
            )
            .distinct()
            .subquery()
        )
        query = (
            db.query(BxAdvanceApply)
            .join(subquery, BxAdvanceApply.id == subquery.c.apply_id)
            .filter(BxAdvanceApply.del_flag == 0, BxAdvanceApply.tenant_no == tenant_no)
        )
        if extra_filters:
            query = query.filter(*extra_filters)
        else:
            if keyword:
                query = query.filter(
                    or_(
                        BxAdvanceApply.apply_code.contains(keyword),
                        BxAdvanceApply.applicant_name.contains(keyword),
                    )
                )
            if applicant_id:
                query = query.filter(BxAdvanceApply.applicant_id == applicant_id)
            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(BxAdvanceApply.create_time >= start)
            if end is not None:
                query = query.filter(BxAdvanceApply.create_time < end)
        query = apply_bx_scope(query, BxAdvanceApply, user, db, tenant_no)
        total = query.count()
        order_expr = order_by if order_by is not None else BxAdvanceApply.create_time.desc()
        rows = (
            query.order_by(order_expr)
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
        return rows, total

    @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 submit(db: Session, apply_id: str, user: BxUserContext) -> Optional[BxAdvanceApply]:
        tenant_no = AdvanceService._tenant_no(db, user)
        apply_obj = AdvanceService.get_by_id(db, apply_id, tenant_no, user=user)
        if not apply_obj:
            return None
        apply_obj.status = "pending"
        if not apply_obj.apply_code:
            apply_obj.apply_code = generate_apply_code(db, "BYJ", tenant_no)
        apply_obj.update_by = user.id
        if apply_obj.wf_id:
            nodes = AdvanceService._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[BxAdvanceApply]:
        tenant_no = AdvanceService._tenant_no(db, approver)
        apply_obj = AdvanceService.get_by_id(db, apply_id, tenant_no, skip_scope=True)
        if not apply_obj or not apply_obj.wf_id:
            return None
        nodes = AdvanceService._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 = BxAdvanceApproval(
            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[BxAdvanceApply]:
        tenant_no = AdvanceService._tenant_no(db, approver)
        apply_obj = AdvanceService.get_by_id(db, apply_id, tenant_no, skip_scope=True)
        if not apply_obj or not apply_obj.wf_id:
            return None
        nodes = AdvanceService._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 = BxAdvanceApproval(
            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
        db.flush()
        return apply_obj

    @staticmethod
    def withdraw(db: Session, apply_id: str, user: BxUserContext) -> Optional[BxAdvanceApply]:
        tenant_no = AdvanceService._tenant_no(db, user)
        apply_obj = AdvanceService.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
        db.flush()
        return apply_obj

    @staticmethod
    def _approval_record_dict(approval: BxAdvanceApproval, apply_id: str) -> Dict[str, Any]:
        return {
            "id": approval.id,
            "advance_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[BxAdvanceApproval],
    ) -> Dict[int, BxAdvanceApproval]:
        """同一节点多次审批时，取 approval_time 最新的一条作为当前节点状态。"""
        approval_map: Dict[int, BxAdvanceApproval] = {}
        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[BxAdvanceApproval]:
        return (
            db.query(BxAdvanceApproval)
            .filter(
                BxAdvanceApproval.apply_id == apply_id,
                BxAdvanceApproval.tenant_no == tenant_no,
                BxAdvanceApproval.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 = AdvanceService.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 = AdvanceService._workflow_nodes(db, wf.id, tenant_no)
        if not nodes:
            return []
        approvals = AdvanceService._query_approvals(db, apply_id, tenant_no)
        cycle_approvals = filter_current_cycle_approvals(approvals, apply_obj.status)
        approval_map = AdvanceService._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(AdvanceService._approval_record_dict(approval, apply_id))
            else:
                result.append(
                    {
                        "id": "0",
                        "advance_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 AdvanceService.get_by_id(db, apply_id, tenant_no, user=user):
            return []
        approvals = (
            db.query(BxAdvanceApproval)
            .filter(
                BxAdvanceApproval.apply_id == apply_id,
                BxAdvanceApproval.tenant_no == tenant_no,
                BxAdvanceApproval.del_flag == 0,
            )
            .order_by(BxAdvanceApproval.approval_time.asc(), BxAdvanceApproval.create_time.asc())
            .all()
        )
        return [AdvanceService._approval_record_dict(a, apply_id) for a in approvals]

    @staticmethod
    def global_status_counts(db: Session, tenant_no: str, *, user: BxUserContext) -> Dict[str, int]:
        query = db.query(BxAdvanceApply.status, func.count()).filter(
            BxAdvanceApply.del_flag == 0, BxAdvanceApply.tenant_no == tenant_no
        )
        query = apply_bx_scope(query, BxAdvanceApply, user, db, tenant_no)
        rows = query.group_by(BxAdvanceApply.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(BxAdvanceApply).filter(
            BxAdvanceApply.del_flag == 0,
            BxAdvanceApply.tenant_no == tenant_no,
            or_(
                BxAdvanceApply.applicant_id == user.id,
                BxAdvanceApply.create_by == user.id,
            ),
        )
        return apply_bx_scope(query, BxAdvanceApply, user, db, tenant_no)

    @staticmethod
    def my_apply_stats(db: Session, user: BxUserContext, tenant_no: str) -> dict:
        base_q = AdvanceService._my_apply_base(db, user, tenant_no).subquery()
        base = BxAdvanceApply.id.in_(db.query(base_q.c.id))
        total_count = (
            db.query(func.count()).select_from(BxAdvanceApply).filter(base).scalar() or 0
        )
        pending = (
            db.query(func.count())
            .select_from(BxAdvanceApply)
            .filter(base, BxAdvanceApply.status == "pending")
            .scalar()
            or 0
        )
        approved_pass = (
            db.query(func.count())
            .select_from(BxAdvanceApply)
            .filter(base, BxAdvanceApply.status == "approved")
            .scalar()
            or 0
        )
        rejected_count = (
            db.query(func.count())
            .select_from(BxAdvanceApply)
            .filter(base, BxAdvanceApply.status == "rejected")
            .scalar()
            or 0
        )
        pending_issue = approved_pass
        issued = (
            db.query(func.count())
            .select_from(BxAdvanceApply)
            .filter(base, BxAdvanceApply.status == "issued")
            .scalar()
            or 0
        )
        issued_amount = (
            db.query(func.coalesce(func.sum(BxAdvanceApply.amount), 0))
            .filter(base, BxAdvanceApply.status == "issued")
            .scalar()
            or 0
        )
        m0, m1 = current_month_bounds()
        month_apply_amount = (
            db.query(func.coalesce(func.sum(BxAdvanceApply.amount), 0))
            .filter(base, BxAdvanceApply.create_time >= m0, BxAdvanceApply.create_time < m1)
            .scalar()
            or 0
        )
        month_apply_count = (
            db.query(func.count())
            .select_from(BxAdvanceApply)
            .filter(base, BxAdvanceApply.create_time >= m0, BxAdvanceApply.create_time < m1)
            .scalar()
            or 0
        )
        offset_total = (
            db.query(func.coalesce(func.sum(BxReimburseOffset.offset_amount), 0))
            .join(BxAdvanceApply, BxReimburseOffset.advance_id == BxAdvanceApply.id)
            .filter(
                BxAdvanceApply.id.in_(db.query(base_q.c.id)),
                BxReimburseOffset.tenant_no == tenant_no,
                BxReimburseOffset.del_flag == 0,
            )
            .scalar()
            or 0
        )
        return {
            "total_count": int(total_count),
            "pending_count": int(pending),
            "approved_pass_count": int(approved_pass),
            "rejected_count": int(rejected_count),
            "pending_issue_count": int(pending_issue),
            "issued_count": int(issued),
            "issued_amount": float(issued_amount),
            "month_apply_amount": float(month_apply_amount),
            "month_apply_count": int(month_apply_count),
            "offset_total": float(offset_total),
        }

    @staticmethod
    def pending_approval_stats(db: Session, user: BxUserContext, tenant_no: str) -> dict:
        subquery = (
            db.query(BxAdvanceApply.id)
            .join(
                BxWorkflowNode,
                and_(
                    BxWorkflowNode.wf_id == BxAdvanceApply.wf_id,
                    BxWorkflowNode.node_name == BxAdvanceApply.current_node,
                    BxWorkflowNode.tenant_no == tenant_no,
                    BxWorkflowNode.del_flag == 0,
                ),
            )
            .filter(
                BxAdvanceApply.del_flag == 0,
                BxAdvanceApply.tenant_no == tenant_no,
                BxAdvanceApply.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(BxAdvanceApply.amount), 0))
            .join(subquery, BxAdvanceApply.id == subquery.c.id)
            .filter(BxAdvanceApply.create_time >= m0, BxAdvanceApply.create_time < m1)
            .scalar()
            or 0
        )
        return {"pending_total": int(total), "month_pending_amount": float(month_sum)}

    @staticmethod
    def offset_amount_map(db: Session, tenant_no: str, advance_ids: List[str]) -> Dict[str, float]:
        if not advance_ids:
            return {}
        rows = (
            db.query(
                BxReimburseOffset.advance_id,
                func.coalesce(func.sum(BxReimburseOffset.offset_amount), 0),
            )
            .filter(
                BxReimburseOffset.advance_id.in_(advance_ids),
                BxReimburseOffset.tenant_no == tenant_no,
                BxReimburseOffset.del_flag == 0,
            )
            .group_by(BxReimburseOffset.advance_id)
            .all()
        )
        return {str(advance_id): float(amount) for advance_id, amount in rows}

    @staticmethod
    def issue(
        db: Session,
        advance_id: str,
        user: BxUserContext,
        remark: str = "",
        actual_amount: Optional[float] = None,
    ) -> Optional[BxAdvanceApply]:
        tenant_no = AdvanceService._tenant_no(db, user)
        apply_obj = AdvanceService.get_by_id(db, advance_id, tenant_no, user=user)
        if not apply_obj:
            return None
        apply_obj.status = "issued"
        apply_obj.update_by = user.id
        apply_obj.issue_status = "issued"
        apply_obj.issue_time = datetime.now()
        apply_obj.issue_user_id = user.id
        apply_obj.issue_user_name = user.name
        if remark:
            apply_obj.issue_remark = remark
        if actual_amount is not None:
            apply_obj.actual_amount = actual_amount
        db.flush()
        return apply_obj

    @staticmethod
    def finance_issue_stats(db: Session, tenant_no: str, *, user: BxUserContext) -> dict:
        base_q = (
            db.query(BxAdvanceApply)
            .filter(BxAdvanceApply.del_flag == 0, BxAdvanceApply.tenant_no == tenant_no)
        )
        base_q = apply_bx_scope(base_q, BxAdvanceApply, user, db, tenant_no).subquery()
        base = BxAdvanceApply.id.in_(db.query(base_q.c.id))
        pending_f = and_(
            BxAdvanceApply.status == "approved",
            or_(BxAdvanceApply.issue_status.is_(None), BxAdvanceApply.issue_status != "issued"),
        )
        pending_n = db.query(func.count()).select_from(BxAdvanceApply).filter(base, pending_f).scalar() or 0
        issued_n = (
            db.query(func.count())
            .select_from(BxAdvanceApply)
            .filter(base, BxAdvanceApply.status == "issued")
            .scalar()
            or 0
        )
        pending_amount = (
            db.query(func.coalesce(func.sum(BxAdvanceApply.amount), 0))
            .filter(base, pending_f)
            .scalar()
            or 0
        )
        m0, m1 = current_month_bounds()
        amt_expr = func.coalesce(BxAdvanceApply.actual_amount, BxAdvanceApply.amount)
        issued_month = (
            db.query(func.coalesce(func.sum(amt_expr), 0))
            .filter(
                base,
                BxAdvanceApply.status == "issued",
                BxAdvanceApply.issue_time.isnot(None),
                BxAdvanceApply.issue_time >= m0,
                BxAdvanceApply.issue_time < m1,
            )
            .scalar()
            or 0
        )
        return {
            "pending_count": int(pending_n),
            "issued_count": int(issued_n),
            "pending_amount": float(pending_amount),
            "issued_month_amount": float(issued_month),
        }