leaudit-platform-backend/fastapi_modules/fastapi_leaudit/controllers/contractTemplateController.py

"""合同模板控制器。"""

from fastapi import Depends, Query
from fastapi.responses import JSONResponse

from fastapi_common.fastapi_common_security.security import verify_access_token
from fastapi_common.fastapi_common_web.controller import BaseController
from fastapi_modules.fastapi_leaudit.domian.Dto.contractTemplateDto import (
    ContractTemplateListQueryDTO,
    ContractTemplateSearchQueryDTO,
)
from fastapi_modules.fastapi_leaudit.services.contractTemplateService import IContractTemplateService
from fastapi_modules.fastapi_leaudit.services.impl.contractTemplateServiceImpl import ContractTemplateServiceImpl
from fastapi_modules.fastapi_leaudit.services.impl.permissionServiceImpl import PermissionServiceImpl
from fastapi_modules.fastapi_leaudit.services.permissionService import IPermissionService


class ContractTemplateController(BaseController):
    """合同模板控制器。"""

    def __init__(self):
        super().__init__(prefix="/v3/contract-templates", tags=["合同模板"])
        self.ContractTemplateService: IContractTemplateService = ContractTemplateServiceImpl()
        self.PermissionService: IPermissionService = PermissionServiceImpl()

        @self.router.get("/categories")
        async def ListContractTemplateCategories(
            include_disabled: bool = Query(False, description="是否包含禁用分类"),
            with_template_count: bool = Query(True, description="是否附带模板数量"),
            payload: dict = Depends(verify_access_token),
        ):
            if not await self._check_permission(int(payload["user_id"]), ["contract_template:list:read", "contract_template:search:read"]):
                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有查看合同模板分类权限", "data": None})
            data = await self.ContractTemplateService.ListCategories(include_disabled, with_template_count)
            return JSONResponse(status_code=200, content={"code": 200, "message": "ok", "data": [item.model_dump() for item in data]})

        @self.router.get("")
        async def ListContractTemplates(
            keyword: str | None = Query(None, description="关键词"),
            category_id: int | None = Query(None, description="分类ID"),
            category_name: str | None = Query(None, description="分类名称"),
            file_format: str | None = Query(None, description="文件格式"),
            is_featured: bool | None = Query(None, description="是否推荐"),
            page: int = Query(1, ge=1, description="页码"),
            page_size: int = Query(12, ge=1, le=200, description="分页大小"),
            sort_by: str = Query("updated_at", description="排序字段"),
            sort_order: str = Query("desc", description="排序方向"),
            payload: dict = Depends(verify_access_token),
        ):
            if not await self._check_permission(int(payload["user_id"]), ["contract_template:list:read"]):
                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有查看合同模板列表权限", "data": None})
            query = ContractTemplateListQueryDTO(
                keyword=keyword,
                category_id=category_id,
                category_name=category_name,
                file_format=file_format,
                is_featured=is_featured,
                page=page,
                page_size=page_size,
                sort_by=sort_by,
                sort_order=sort_order,
            )
            data = await self.ContractTemplateService.ListTemplates(query)
            return JSONResponse(status_code=200, content={"code": 200, "message": "ok", "data": data.model_dump()})

        @self.router.get("/search")
        async def SearchContractTemplates(
            q: str = Query(..., min_length=1, description="搜索关键词"),
            category_id: int | None = Query(None, description="分类ID"),
            category_name: str | None = Query(None, description="分类名称"),
            page: int = Query(1, ge=1, description="页码"),
            page_size: int = Query(12, ge=1, le=200, description="分页大小"),
            sort_by: str = Query("updated_at", description="排序字段"),
            sort_order: str = Query("desc", description="排序方向"),
            payload: dict = Depends(verify_access_token),
        ):
            if not await self._check_permission(int(payload["user_id"]), ["contract_template:search:read"]):
                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有搜索合同模板权限", "data": None})
            query = ContractTemplateSearchQueryDTO(
                q=q,
                category_id=category_id,
                category_name=category_name,
                page=page,
                page_size=page_size,
                sort_by=sort_by,
                sort_order=sort_order,
            )
            data = await self.ContractTemplateService.SearchTemplates(query)
            return JSONResponse(status_code=200, content={"code": 200, "message": "ok", "data": data.model_dump()})

        @self.router.get("/{TemplateId}")
        async def GetContractTemplateDetail(
            TemplateId: int,
            payload: dict = Depends(verify_access_token),
        ):
            if not await self._check_permission(int(payload["user_id"]), ["contract_template:detail:read", "contract_template:list:read"]):
                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有查看合同模板详情权限", "data": None})
            data = await self.ContractTemplateService.GetTemplateDetail(TemplateId)
            if not data:
                return JSONResponse(status_code=404, content={"code": 404, "msg": "合同模板不存在", "data": None})
            return JSONResponse(status_code=200, content={"code": 200, "message": "ok", "data": data.model_dump()})

    async def _check_permission(self, user_id: int, permission_keys: list[str]) -> bool:
        for permission_key in permission_keys:
            if await self.PermissionService.CheckPermission(user_id, permission_key):
                return True
        return False