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

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

from fastapi import Depends, File, Form, Query, UploadFile
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 (
    ContractTemplateCreateDTO,
    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="分类名称"),
            region: str | None = Query(None, description="兼容保留字段：租户展示值/旧地区"),
            tenant_code: 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,
                region=region,
                tenant_code=tenant_code,
                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, int(payload["user_id"]))
            return JSONResponse(status_code=200, content={"code": 200, "message": "ok", "data": data.model_dump()})

        @self.router.post("")
        async def CreateContractTemplate(
            title: str = Form(...),
            template_code: str = Form(...),
            category_id: int = Form(...),
            region: str | None = Form(default=None),
            tenant_code: str | None = Form(default=None),
            description: str | None = Form(default=None),
            is_featured: bool = Form(default=False),
            file: UploadFile = File(...),
            pdf_file: UploadFile | None = File(default=None),
            payload: dict = Depends(verify_access_token),
        ):
            if not await self._check_permission(int(payload["user_id"]), ["contract_template:create:write"]):
                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前仅允许租户管理员上传合同模板", "data": None})
            body = ContractTemplateCreateDTO(
                title=title,
                template_code=template_code,
                category_id=category_id,
                region=region,
                tenant_code=tenant_code,
                description=description,
                is_featured=is_featured,
            )
            data = await self.ContractTemplateService.CreateTemplate(body, file, pdf_file, int(payload["user_id"]))
            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="分类名称"),
            region: str | None = Query(None, description="兼容保留字段：租户展示值/旧地区"),
            tenant_code: 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,
                region=region,
                tenant_code=tenant_code,
                page=page,
                page_size=page_size,
                sort_by=sort_by,
                sort_order=sort_order,
            )
            data = await self.ContractTemplateService.SearchTemplates(query, int(payload["user_id"]))
            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, int(payload["user_id"]))
            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()})

        @self.router.delete("/{TemplateId}")
        async def DeleteContractTemplate(
            TemplateId: int,
            payload: dict = Depends(verify_access_token),
        ):
            if not await self._check_permission(int(payload["user_id"]), ["contract_template:delete:delete"]):
                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有删除合同模板权限", "data": None})
            await self.ContractTemplateService.DeleteTemplate(TemplateId, int(payload["user_id"]))
            return JSONResponse(status_code=200, content={"code": 200, "message": "ok", "data": True})

    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