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

from sqlalchemy import case as sql_case, func, or_
from sqlalchemy.orm import Session

from app.models.attachment import Attachment
from app.services.base import CrudService


class AttachmentService:

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

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

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

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

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

    @staticmethod
    def get_by_biz(db: Session, biz_type: str, biz_id: int) -> list[Attachment]:
        return (
            db.query(Attachment)
            .filter(
                Attachment.biz_type == biz_type,
                Attachment.biz_id == biz_id,
                Attachment.del_flag == 0,
            )
            .order_by(Attachment.id)
            .all()
        )

    @staticmethod
    def bind_biz(db: Session, ids: list[int], biz_type: str, biz_id: int) -> int:
        """Batch update biz_type and biz_id for given attachment IDs. Returns count updated."""
        count = (
            db.query(Attachment)
            .filter(Attachment.id.in_(ids), Attachment.del_flag == 0)
            .update({Attachment.biz_type: biz_type, Attachment.biz_id: biz_id}, synchronize_session="fetch")
        )
        db.flush()
        return count

    @staticmethod
    def count_invoice_attach_by_biz_ids(db: Session, biz_type: str, biz_ids: list[int]) -> dict[int, tuple[int, int]]:
        """按业务主键统计 (invoice_count, attach_count)。

        - 显式标记 attach_type='invoice' 的算「张」。
        - 报销申请附件：上传接口未写 attach_type 时默认全是 attachment，故对 reimburse_apply
          再把常见票据 MIME（图片、PDF）计入「张」，其余计入「个」，避免列表里永远「0 张」。
        """
        if not biz_ids:
            return {}
        if biz_type == "reimburse_apply":
            inv_cond = or_(
                Attachment.attach_type == "invoice",
                Attachment.file_type.isnot(None) & Attachment.file_type.like("image/%"),
                Attachment.file_type == "application/pdf",
            )
        else:
            inv_cond = Attachment.attach_type == "invoice"
        inv_case = sql_case((inv_cond, 1), else_=0)
        att_case = sql_case((inv_cond, 0), else_=1)
        rows = (
            db.query(
                Attachment.biz_id,
                func.coalesce(func.sum(inv_case), 0).label("inv"),
                func.coalesce(func.sum(att_case), 0).label("att"),
            )
            .filter(
                Attachment.biz_type == biz_type,
                Attachment.biz_id.in_(biz_ids),
                Attachment.del_flag == 0,
            )
            .group_by(Attachment.biz_id)
            .all()
        )
        return {int(r.biz_id): (int(r.inv), int(r.att)) for r in rows}