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

"""文档控制器。"""

from typing import Any

from fastapi import File, Form, Query, UploadFile
from pydantic import BaseModel, Field
from sqlalchemy import text

from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
from fastapi_common.fastapi_common_web.controller import BaseController
from fastapi_common.fastapi_common_web.domain.responses import Result

from fastapi_modules.fastapi_leaudit.domian.vo.documentVo import (
    DocumentListPageVO,
    DocumentTypeCreateDTO,
    DocumentTypeItemVO,
    DocumentTypeUpdateDTO,
    DocumentUploadVO,
)
from fastapi_modules.fastapi_leaudit.services import IDocumentService
from fastapi_modules.fastapi_leaudit.services.impl.documentServiceImpl import DocumentServiceImpl


class QueueStatusVO(BaseModel):
    success: bool = Field(True)
    timestamp: str = Field("")
    queue: dict[str, int] = Field(default_factory=lambda: {"pending_tasks": 0, "processing_tasks": 0, "available_slots": 4, "max_concurrent": 4})
    documents: dict[str, object] = Field(default_factory=lambda: {"waiting": 0, "processing": 0, "processing_ids": []})


class DocumentController(BaseController):
    """文档控制器。"""

    def __init__(self):
        super().__init__(prefix="", tags=["文档"])
        self.DocumentService: IDocumentService = DocumentServiceImpl()

        @self.router.post("/upload", response_model=Result[DocumentUploadVO])
        async def UploadDocument(
            file: UploadFile = File(..., description="上传文档"),
            typeId: int | None = Form(None, description="文档类型ID"),
            typeCode: str | None = Form(None, description="文档类型编码"),
            region: str = Form("default", description="所属地区"),
            fileRole: str = Form("primary", description="文件角色"),
            createdBy: int | None = Form(None, description="上传用户ID"),
            autoRun: bool = Form(False, description="是否上传后自动触发评查"),
            speed: str = Form("normal", description="执行速度档位：urgent/normal"),
        ):
            """上传文档并建立评查输入。"""
            Content = await file.read()
            Data = await self.DocumentService.Upload(
                FileName=file.filename or "upload.bin",
                FileContent=Content,
                ContentType=file.content_type,
                TypeId=typeId,
                TypeCode=typeCode,
                Region=region,
                FileRole=fileRole,
                CreatedBy=createdBy,
                AutoRun=autoRun,
                Speed=speed,
            )
            return Result.success(data=Data)

        @self.router.get("/documents/list", response_model=Result[DocumentListPageVO])
        async def ListDocuments(
            page: int = 1,
            pageSize: int = 20,
            keyword: str | None = None,
            typeCode: str | None = None,
            region: str | None = None,
            processingStatus: str | None = None,
            resultStatus: str | None = None,
            userId: int | None = Query(None, description="按用户ID过滤"),
            dateFrom: str | None = Query(None, description="起始日期 (YYYY-MM-DD)"),
            dateTo: str | None = Query(None, description="结束日期 (YYYY-MM-DD)"),
        ):
            """获取文档列表（仅返回最新版本，附历史版本摘要）。"""
            Data = await self.DocumentService.ListDocuments(
                Page=page,
                PageSize=pageSize,
                Keyword=keyword,
                TypeCode=typeCode,
                Region=region,
                ProcessingStatus=processingStatus,
                ResultStatus=resultStatus,
                UserId=userId,
                DateFrom=dateFrom,
                DateTo=dateTo,
            )
            return Result.success(data=Data)

        @self.router.get("/document-types", response_model=Result[list[DocumentTypeItemVO]])
        async def ListDocumentTypes(
            ids: str | None = Query(None, description="逗号分隔的ID列表，不传则返回全部"),
        ):
            """获取文档类型列表。"""
            idList: list[int] | None = None
            if ids:
                idList = [int(x.strip()) for x in ids.split(",") if x.strip().isdigit()]
            Data = await self.DocumentService.ListDocumentTypes(Ids=idList)
            return Result.success(data=Data)

        @self.router.get("/document-types/{TypeId}", response_model=Result[DocumentTypeItemVO])
        async def GetDocumentType(TypeId: int):
            """获取文档类型详情。"""
            Data = await self.DocumentService.GetDocumentType(Id=TypeId)
            return Result.success(data=Data)

        @self.router.post("/document-types", response_model=Result[DocumentTypeItemVO])
        async def CreateDocumentType(Body: DocumentTypeCreateDTO):
            """创建文档类型。"""
            Data = await self.DocumentService.CreateDocumentType(Body=Body)
            return Result.success(data=Data, message="文档类型创建成功")

        @self.router.put("/document-types/{TypeId}", response_model=Result[DocumentTypeItemVO])
        async def UpdateDocumentType(TypeId: int, Body: DocumentTypeUpdateDTO):
            """更新文档类型。"""
            Data = await self.DocumentService.UpdateDocumentType(Id=TypeId, Body=Body)
            return Result.success(data=Data, message="文档类型更新成功")

        @self.router.delete("/document-types/{TypeId}", response_model=Result[None])
        async def DeleteDocumentType(TypeId: int):
            """删除文档类型（软删除）。"""
            await self.DocumentService.DeleteDocumentType(Id=TypeId)
            return Result.success(message="文档类型已删除")

        @self.router.get("/v2/system/queue/status", response_model=Result[QueueStatusVO])
        async def GetQueueStatus():
            """获取文档处理队列状态。"""
            from datetime import datetime

            async with GetAsyncSession() as Session:
                statusRows = (
                    await Session.execute(
                        text(
                            """
                            SELECT processing_status, COUNT(*) AS cnt
                            FROM leaudit_documents
                            WHERE deleted_at IS NULL AND is_latest_version = true
                            GROUP BY processing_status
                            """
                        )
                    )
                ).mappings().all()

            waiting = 0
            processing = 0
            for row in statusRows:
                s = str(row["processing_status"] or "")
                c = int(row["cnt"] or 0)
                if s == "waiting":
                    waiting = c
                elif s in ("processing", "running"):
                    processing += c

            processingIdsRows: list[int] = []
            if processing > 0:
                async with GetAsyncSession() as Session:
                    idRows = (
                        await Session.execute(
                            text(
                                """
                                SELECT id FROM leaudit_documents
                                WHERE deleted_at IS NULL
                                  AND is_latest_version = true
                                  AND processing_status IN ('processing', 'running')
                                ORDER BY updated_at DESC
                                LIMIT 50
                                """
                            )
                        )
                    ).fetchall()
                    processingIdsRows = [int(r[0]) for r in idRows]

            return Result.success(
                data=QueueStatusVO(
                    success=True,
                    timestamp=datetime.now().isoformat(),
                    queue={
                        "pending_tasks": waiting,
                        "processing_tasks": processing,
                        "available_slots": max(0, 4 - processing),
                        "max_concurrent": 4,
                    },
                    documents={
                        "waiting": waiting,
                        "processing": processing,
                        "processing_ids": processingIdsRows,
                    },
                )
            )