ai-agent-admin/backend-fastapi/online_dev/electronic_seal/service.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
电子签章服务层
"""
from typing import Optional, List, Tuple

from sqlalchemy import select, func, and_, or_
from sqlalchemy.ext.asyncio import AsyncSession

from app.base_service import BaseService
from online_dev.electronic_seal.model import ElectronicSeal, SealUsageLog
from online_dev.electronic_seal.schema import ElectronicSealCreate, ElectronicSealUpdate


class ElectronicSealService(BaseService[ElectronicSeal, ElectronicSealCreate, ElectronicSealUpdate]):
    """电子签章服务"""

    model = ElectronicSeal

    @classmethod
    async def get_list(
        cls,
        db: AsyncSession,
        page: int = 1,
        page_size: int = 20,
        seal_type: Optional[str] = None,
        status: Optional[str] = None,
        keyword: Optional[str] = None,
    ) -> Tuple[List[ElectronicSeal], int]:
        """获取签章列表"""
        conditions = [cls.model.is_deleted == False]

        if seal_type:
            conditions.append(cls.model.seal_type == seal_type)
        if status:
            conditions.append(cls.model.status == status)
        if keyword:
            conditions.append(
                or_(
                    cls.model.name.ilike(f"%{keyword}%"),
                    cls.model.description.ilike(f"%{keyword}%"),
                )
            )

        # 查询总数
        count_stmt = select(func.count()).select_from(cls.model).where(and_(*conditions))
        total = (await db.execute(count_stmt)).scalar() or 0

        # 查询列表
        stmt = (
            select(cls.model)
            .where(and_(*conditions))
            .order_by(cls.model.sys_create_datetime.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
        )
        result = await db.execute(stmt)
        items = list(result.scalars().all())

        return items, total

    @classmethod
    async def get_available_seals(
        cls,
        db: AsyncSession,
        user_id: str,
        dept_id: Optional[str] = None,
        role_ids: Optional[List[str]] = None,
        template_id: Optional[str] = None,
    ) -> List[ElectronicSeal]:
        """获取用户可用的签章"""
        conditions = [
            cls.model.is_deleted == False,
            cls.model.status == "active",
        ]

        stmt = select(cls.model).where(and_(*conditions))
        result = await db.execute(stmt)
        all_seals = list(result.scalars().all())

        available_seals = []
        for seal in all_seals:
            # 检查所有者权限
            if seal.owner_type == "all":
                pass
            elif seal.owner_type == "user":
                if not seal.owner_ids or user_id not in seal.owner_ids:
                    continue
            elif seal.owner_type == "dept":
                if not seal.owner_ids or not dept_id or dept_id not in seal.owner_ids:
                    continue
            elif seal.owner_type == "role":
                if not seal.owner_ids or not role_ids:
                    continue
                if not any(r in seal.owner_ids for r in role_ids):
                    continue

            # 检查模板范围
            if seal.scope == "specific" and template_id:
                if not seal.allowed_template_ids or template_id not in seal.allowed_template_ids:
                    continue

            available_seals.append(seal)

        return available_seals


class SealUsageLogService(BaseService[SealUsageLog, None, None]):
    """签章使用记录服务"""

    model = SealUsageLog

    @classmethod
    async def log_usage(
        cls,
        db: AsyncSession,
        seal_id: str,
        seal_name: str,
        document_id: str,
        document_name: str,
        user_id: str,
        user_name: str,
        position_x: int = 0,
        position_y: int = 0,
        page_number: int = 1,
    ) -> SealUsageLog:
        """记录签章使用"""
        log = SealUsageLog(
            seal_id=seal_id,
            seal_name=seal_name,
            document_id=document_id,
            document_name=document_name,
            user_id=user_id,
            user_name=user_name,
            position_x=position_x,
            position_y=position_y,
            page_number=page_number,
        )

        db.add(log)
        await db.commit()
        await db.refresh(log)
        return log

    @classmethod
    async def get_logs_by_seal(
        cls,
        db: AsyncSession,
        seal_id: str,
        page: int = 1,
        page_size: int = 20,
    ) -> Tuple[List[SealUsageLog], int]:
        """获取签章的使用记录"""
        conditions = [
            cls.model.is_deleted == False,
            cls.model.seal_id == seal_id,
        ]

        # 查询总数
        count_stmt = select(func.count()).select_from(cls.model).where(and_(*conditions))
        total = (await db.execute(count_stmt)).scalar() or 0

        # 查询列表
        stmt = (
            select(cls.model)
            .where(and_(*conditions))
            .order_by(cls.model.sys_create_datetime.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
        )
        result = await db.execute(stmt)
        items = list(result.scalars().all())

        return items, total

    @classmethod
    async def get_logs_by_document(
        cls,
        db: AsyncSession,
        document_id: str,
    ) -> List[SealUsageLog]:
        """获取文档的签章记录"""
        stmt = (
            select(cls.model)
            .where(
                and_(
                    cls.model.is_deleted == False,
                    cls.model.document_id == document_id,
                )
            )
            .order_by(cls.model.sys_create_datetime.desc())
        )
        result = await db.execute(stmt)
        return list(result.scalars().all())