shanxia/bx/bx-backend/app/services/advance.py

from datetime import datetime

from app.utils.date_range import build_create_time_filters, current_month_bounds

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

from app.models.advance import AdvanceApply, AdvanceApproval
from app.models.reimburse import ReimburseOffset
from app.models.workflow import WorkflowDef, WorkflowNode
from app.services.base import CrudService
from app.utils.codegen import generate_apply_code


class AdvanceService:

    @staticmethod
    def _filter_apply_create_time(
        query,
        apply_date: str | None = None,
        apply_date_start: str | None = None,
        apply_date_end: str | None = None,
    ):
        """按申请时间（create_time）区间筛选，日期格式 YYYY-MM-DD。"""
        for cond in build_create_time_filters(
            AdvanceApply.create_time, apply_date, apply_date_start, apply_date_end
        ):
            query = query.filter(cond)
        return query

    @staticmethod
    def get_by_id(db: Session, id: int) -> AdvanceApply | None:
        return CrudService.get_by_id(AdvanceApply, db, id)

    @staticmethod
    def get_paged(db: Session, page: int, page_size: int, filters: list | None = None):
        return CrudService.get_paged(AdvanceApply, db, page, page_size, filters)

    @staticmethod
    def create(db: Session, obj_in: dict) -> AdvanceApply:
        return CrudService.create(AdvanceApply, db, obj_in)

    @staticmethod
    def update(db: Session, id: int, obj_in: dict) -> AdvanceApply | None:
        return CrudService.update(AdvanceApply, db, id, obj_in)

    @staticmethod
    def soft_delete(db: Session, id: int) -> bool:
        return CrudService.soft_delete(AdvanceApply, db, id)

    @staticmethod
    def pending_approval(
        db: Session,
        user,
        page: int,
        page_size: int,
        keyword: str = "",
        applicant_id: int | None = None,
        apply_date: str | None = None,
        apply_date_start: str | None = None,
        apply_date_end: str | None = None,
    ):
        """查询待当前用户审批的备用金申请（current_node 对应的 WorkflowNode.approver_id == user.id）"""
        subquery = (
            db.query(AdvanceApply.id)
            .join(WorkflowNode, and_(
                WorkflowNode.wf_id == AdvanceApply.wf_id,
                WorkflowNode.node_name == AdvanceApply.current_node,
            ))
            .filter(
                AdvanceApply.del_flag == 0,
                AdvanceApply.status == "pending",
                WorkflowNode.approver_id == user.id,
            )
        )
        if keyword:
            subquery = subquery.filter(
                AdvanceApply.apply_code.contains(keyword) | AdvanceApply.applicant_name.contains(keyword)
            )
        if applicant_id is not None:
            subquery = subquery.filter(AdvanceApply.applicant_id == applicant_id)
        subquery = AdvanceService._filter_apply_create_time(
            subquery, apply_date, apply_date_start, apply_date_end
        )
        subquery = subquery.subquery()
        total = db.query(func.count()).select_from(subquery).scalar()
        rows = (
            db.query(AdvanceApply)
            .join(subquery, AdvanceApply.id == subquery.c.id)
            .order_by(AdvanceApply.id.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
        return rows, total or 0

    @staticmethod
    def approved_by_me(
        db: Session,
        user,
        page: int,
        page_size: int,
        keyword: str = "",
        applicant_id: int | None = None,
        apply_date: str | None = None,
        apply_date_start: str | None = None,
        apply_date_end: str | None = None,
    ):
        """查询当前用户已审批过的备用金申请"""
        subquery = (
            db.query(AdvanceApproval.apply_id)
            .filter(AdvanceApproval.approver_id == user.id)
            .distinct()
            .subquery()
        )
        query = (
            db.query(AdvanceApply)
            .join(subquery, AdvanceApply.id == subquery.c.apply_id)
            .filter(AdvanceApply.del_flag == 0)
        )
        if keyword:
            query = query.filter(
                AdvanceApply.apply_code.contains(keyword) | AdvanceApply.applicant_name.contains(keyword)
            )
        if applicant_id is not None:
            query = query.filter(AdvanceApply.applicant_id == applicant_id)
        query = AdvanceService._filter_apply_create_time(
            query, apply_date, apply_date_start, apply_date_end
        )
        total = query.count()
        rows = (
            query.order_by(AdvanceApply.id.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
        return rows, total

    @staticmethod
    def submit(db: Session, apply_id: int, user) -> AdvanceApply | None:
        apply_obj = (
            db.query(AdvanceApply)
            .filter(AdvanceApply.id == apply_id, AdvanceApply.del_flag == 0)
            .first()
        )
        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")
        apply_obj.update_by = str(user.id)
        wf = (
            db.query(WorkflowDef)
            .filter(WorkflowDef.id == apply_obj.wf_id, WorkflowDef.del_flag == 0)
            .first()
        )
        if wf:
            nodes = (
                db.query(WorkflowNode)
                .filter(WorkflowNode.wf_id == wf.id)
                .order_by(WorkflowNode.node_order)
                .all()
            )
            if nodes:
                apply_obj.current_node = nodes[0].node_name
        db.flush()
        return apply_obj

    @staticmethod
    def approve(db: Session, apply_id: int, approver, opinion: str) -> AdvanceApply | None:
        apply_obj = (
            db.query(AdvanceApply)
            .filter(AdvanceApply.id == apply_id, AdvanceApply.del_flag == 0)
            .first()
        )
        if not apply_obj:
            return None
        wf = (
            db.query(WorkflowDef)
            .filter(WorkflowDef.id == apply_obj.wf_id, WorkflowDef.del_flag == 0)
            .first()
        )
        if not wf:
            return None
        nodes = (
            db.query(WorkflowNode)
            .filter(WorkflowNode.wf_id == wf.id)
            .order_by(WorkflowNode.node_order)
            .all()
        )
        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 = AdvanceApproval(
            apply_id=apply_id,
            approver_id=approver.id,
            approver_name=approver.username,
            approval_opinion=opinion,
            node_name=current_node.node_name,
            node_order=current_node.node_order,
            approval_status="approved",
            approval_time=datetime.now(),
        )
        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 = str(approver.id)
        db.flush()
        return apply_obj

    @staticmethod
    def reject(db: Session, apply_id: int, approver, opinion: str) -> AdvanceApply | None:
        apply_obj = (
            db.query(AdvanceApply)
            .filter(AdvanceApply.id == apply_id, AdvanceApply.del_flag == 0)
            .first()
        )
        if not apply_obj:
            return None
        wf = (
            db.query(WorkflowDef)
            .filter(WorkflowDef.id == apply_obj.wf_id, WorkflowDef.del_flag == 0)
            .first()
        )
        if not wf:
            return None
        nodes = (
            db.query(WorkflowNode)
            .filter(WorkflowNode.wf_id == wf.id)
            .order_by(WorkflowNode.node_order)
            .all()
        )
        current_node = None
        for node in nodes:
            if node.node_name == apply_obj.current_node:
                current_node = node
                break
        if current_node is None:
            return None
        if current_node.approver_id != approver.id:
            return None
        approval = AdvanceApproval(
            apply_id=apply_id,
            approver_id=approver.id,
            approver_name=approver.username,
            approval_opinion=opinion,
            node_name=current_node.node_name,
            node_order=current_node.node_order,
            approval_status="rejected",
            approval_time=datetime.now(),
        )
        db.add(approval)
        apply_obj.status = "rejected"
        apply_obj.current_node = None
        apply_obj.update_by = str(approver.id)
        db.flush()
        return apply_obj

    @staticmethod
    def issue(db: Session, apply_id: int, user, remark: str = "", actual_amount: float | None = None) -> AdvanceApply | None:
        apply_obj = (
            db.query(AdvanceApply)
            .filter(AdvanceApply.id == apply_id, AdvanceApply.del_flag == 0)
            .first()
        )
        if not apply_obj:
            return None
        apply_obj.status = "issued"
        apply_obj.update_by = str(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.real_name or user.username
        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 get_approvals(db: Session, apply_id: int) -> list[dict]:
        apply_obj = (
            db.query(AdvanceApply)
            .filter(AdvanceApply.id == apply_id, AdvanceApply.del_flag == 0)
            .first()
        )
        if not apply_obj or not apply_obj.wf_id:
            return []

        wf = (
            db.query(WorkflowDef)
            .filter(WorkflowDef.id == apply_obj.wf_id, WorkflowDef.del_flag == 0)
            .first()
        )
        if not wf:
            return []

        nodes = (
            db.query(WorkflowNode)
            .filter(WorkflowNode.wf_id == wf.id)
            .order_by(WorkflowNode.node_order)
            .all()
        )
        if not nodes:
            return []

        approvals = (
            db.query(AdvanceApproval)
            .filter(AdvanceApproval.apply_id == apply_id)
            .all()
        )
        approval_map = {a.node_order: a for a in approvals}

        result = []
        for node in nodes:
            approval = approval_map.get(node.node_order)
            if approval:
                result.append({
                    "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,
                })
            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 global_status_counts(db: Session) -> dict[str, int]:
        rows = (
            db.query(AdvanceApply.status, func.count())
            .filter(AdvanceApply.del_flag == 0)
            .group_by(AdvanceApply.status)
            .all()
        )
        return {str(s or ""): int(c) for s, c in rows}

    @staticmethod
    def my_apply_stats(db: Session, user_id: int) -> dict:
        """我的备用金：全量聚合；待发放=审批通过且财务未标记发放；累计冲抵=本人备用金被冲抵金额之和。"""
        base = and_(AdvanceApply.del_flag == 0, AdvanceApply.applicant_id == user_id)
        total_count = db.query(func.count()).select_from(AdvanceApply).filter(base).scalar() or 0
        pending = db.query(func.count()).select_from(AdvanceApply).filter(base, AdvanceApply.status == "pending").scalar() or 0
        approved_pass = (
            db.query(func.count())
            .select_from(AdvanceApply)
            .filter(base, AdvanceApply.status == "approved")
            .scalar()
            or 0
        )
        rejected_count = db.query(func.count()).select_from(AdvanceApply).filter(base, AdvanceApply.status == "rejected").scalar() or 0
        pending_issue = approved_pass
        issued = db.query(func.count()).select_from(AdvanceApply).filter(base, AdvanceApply.status == "issued").scalar() or 0
        issued_amount = (
            db.query(func.coalesce(func.sum(AdvanceApply.amount), 0)).filter(base, AdvanceApply.status == "issued").scalar() or 0
        )
        m0, m1 = current_month_bounds()
        month_apply_amount = (
            db.query(func.coalesce(func.sum(AdvanceApply.amount), 0))
            .filter(base, AdvanceApply.create_time >= m0, AdvanceApply.create_time < m1)
            .scalar()
            or 0
        )
        month_apply_count = (
            db.query(func.count())
            .select_from(AdvanceApply)
            .filter(base, AdvanceApply.create_time >= m0, AdvanceApply.create_time < m1)
            .scalar()
            or 0
        )
        offset_total = (
            db.query(func.coalesce(func.sum(ReimburseOffset.offset_amount), 0))
            .join(AdvanceApply, ReimburseOffset.advance_id == AdvanceApply.id)
            .filter(AdvanceApply.applicant_id == user_id, AdvanceApply.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) -> dict:
        """待我审批的备用金。"""
        subquery = (
            db.query(AdvanceApply.id)
            .join(
                WorkflowNode,
                and_(
                    WorkflowNode.wf_id == AdvanceApply.wf_id,
                    WorkflowNode.node_name == AdvanceApply.current_node,
                ),
            )
            .filter(
                AdvanceApply.del_flag == 0,
                AdvanceApply.status == "pending",
                WorkflowNode.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(AdvanceApply.amount), 0))
            .join(subquery, AdvanceApply.id == subquery.c.id)
            .filter(AdvanceApply.create_time >= m0, AdvanceApply.create_time < m1)
            .scalar()
            or 0
        )
        return {"pending_total": int(total), "month_pending_amount": float(month_sum)}

    @staticmethod
    def finance_issue_stats(db: Session) -> dict:
        """财务-备用金发放看板。"""
        base = AdvanceApply.del_flag == 0
        pending_f = and_(
            AdvanceApply.status == "approved",
            or_(AdvanceApply.issue_status.is_(None), AdvanceApply.issue_status != "issued"),
        )
        pending_n = db.query(func.count()).filter(base, pending_f).scalar() or 0
        issued_n = db.query(func.count()).filter(base, AdvanceApply.status == "issued").scalar() or 0
        pending_amount = (
            db.query(func.coalesce(func.sum(AdvanceApply.amount), 0)).filter(base, pending_f).scalar() or 0
        )
        m0, m1 = current_month_bounds()
        amt_expr = func.coalesce(AdvanceApply.actual_amount, AdvanceApply.amount)
        issued_month = (
            db.query(func.coalesce(func.sum(amt_expr), 0))
            .filter(
                base,
                AdvanceApply.status == "issued",
                AdvanceApply.issue_time.isnot(None),
                AdvanceApply.issue_time >= m0,
                AdvanceApply.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),
        }