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

from __future__ import annotations

from typing import List, Optional, Optional

from sqlalchemy import case as sql_case, 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.models.bx_models import BxAttachment
from app.services.bx.base import BxCrudService
from app.services.bx_data_scope import assert_bx_attachment_biz_in_scope, assert_bx_biz_in_scope


class AttachmentService:
    @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,
        attachment_id: str,
        tenant_no: str,
        *,
        user: Optional[BxUserContext] = None,
    ) -> Optional[BxAttachment]:
        row = BxCrudService.get_by_id(BxAttachment, db, attachment_id, tenant_no)
        if row and user:
            assert_bx_attachment_biz_in_scope(db, row, user, tenant_no)
        return row

    @staticmethod
    def create(
        db: Session,
        obj_in: dict,
        *,
        user: BxUserContext,
    ) -> BxAttachment:
        tenant_no = AttachmentService._tenant_no(db, user)
        return BxCrudService.create(
            BxAttachment,
            db,
            obj_in,
            tenant_no=tenant_no,
            user_id=user.id,
            dept_id=user.dept_id,
        )

    @staticmethod
    def soft_delete(db: Session, attachment_id: str, *, user: BxUserContext) -> bool:
        tenant_no = AttachmentService._tenant_no(db, user)
        row = AttachmentService.get_by_id(db, attachment_id, tenant_no, user=user)
        if not row:
            return False
        return BxCrudService.soft_delete(
            BxAttachment, db, attachment_id, tenant_no, user_id=user.id
        )

    @staticmethod
    def get_by_biz(
        db: Session, biz_type: str, biz_id: str, tenant_no: str, *, user: BxUserContext
    ) -> List[BxAttachment]:
        assert_bx_biz_in_scope(db, biz_type, biz_id, user, tenant_no)
        query = db.query(BxAttachment).filter(
            BxAttachment.biz_type == biz_type,
            BxAttachment.biz_id == biz_id,
            BxAttachment.tenant_no == tenant_no,
            BxAttachment.del_flag == 0,
        )
        return query.order_by(BxAttachment.sort_order, BxAttachment.create_time).all()

    @staticmethod
    def bind_biz(
        db: Session,
        ids: List[str],
        biz_type: str,
        biz_id: str,
        *,
        user: BxUserContext,
    ) -> int:
        tenant_no = AttachmentService._tenant_no(db, user)
        assert_bx_biz_in_scope(db, biz_type, biz_id, user, tenant_no)
        if not ids:
            return 0
        count = (
            db.query(BxAttachment)
            .filter(
                BxAttachment.id.in_(ids),
                BxAttachment.tenant_no == tenant_no,
                BxAttachment.del_flag == 0,
            )
            .update(
                {
                    BxAttachment.biz_type: biz_type,
                    BxAttachment.biz_id: biz_id,
                    BxAttachment.update_by: user.id,
                },
                synchronize_session="fetch",
            )
        )
        db.flush()
        return int(count or 0)

    @staticmethod
    def count_invoice_attach_by_biz_ids(
        db: Session, biz_type: str, biz_ids: List[str], tenant_no: str
    ) -> dict[str, tuple[int, int]]:
        if not biz_ids:
            return {}
        if biz_type == "reimburse_apply":
            inv_cond = or_(
                BxAttachment.attach_type == "invoice",
                BxAttachment.file_type.isnot(None) & BxAttachment.file_type.like("image/%"),
                BxAttachment.file_type == "application/pdf",
            )
        else:
            inv_cond = BxAttachment.attach_type == "invoice"
        inv_case = sql_case((inv_cond, 1), else_=0)
        att_case = sql_case((inv_cond, 0), else_=1)
        rows = (
            db.query(
                BxAttachment.biz_id,
                func.coalesce(func.sum(inv_case), 0).label("inv"),
                func.coalesce(func.sum(att_case), 0).label("att"),
            )
            .filter(
                BxAttachment.biz_type == biz_type,
                BxAttachment.biz_id.in_(biz_ids),
                BxAttachment.tenant_no == tenant_no,
                BxAttachment.del_flag == 0,
            )
            .group_by(BxAttachment.biz_id)
            .all()
        )
        return {str(r.biz_id): (int(r.inv), int(r.att)) for r in rows}