File size: 3,001 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | """合同文件夹自动捕获 API。"""
from __future__ import annotations
from uuid import UUID
from django.http import HttpRequest
from ninja import Router
from apps.contracts.schemas import (
ContractFolderScanConfirmIn,
ContractFolderScanConfirmOut,
ContractFolderScanStartIn,
ContractFolderScanStartOut,
ContractFolderScanStatusOut,
ContractFolderScanSubfolderListOut,
)
from apps.contracts.services.contract.integrations.folder_scan_service import ContractFolderScanService
from apps.core.infrastructure.throttling import rate_limit_from_settings
from apps.core.security import get_request_access_context
router = Router()
def _get_service() -> ContractFolderScanService:
return ContractFolderScanService()
def _require_contract_access(request: HttpRequest, contract_id: int) -> None:
from apps.contracts.services.contract import wiring
ctx = get_request_access_context(request)
wiring.get_contract_query_facade().get_contract_ctx(contract_id=contract_id, ctx=ctx)
@router.post("/{contract_id}/folder-scan", response=ContractFolderScanStartOut)
@rate_limit_from_settings("TASK", by_user=True)
def start_contract_scan(request: HttpRequest, contract_id: int, payload: ContractFolderScanStartIn) -> dict[str, str]:
_require_contract_access(request, contract_id)
ctx = get_request_access_context(request)
session = _get_service().start_scan(
contract_id=contract_id,
started_by=ctx.user,
rescan=bool(payload.rescan),
scan_subfolder=str(payload.scan_subfolder or ""),
)
return {
"session_id": str(session.id),
"status": str(session.status),
"task_id": str(session.task_id or ""),
}
@router.get("/{contract_id}/folder-scan/subfolders", response=ContractFolderScanSubfolderListOut)
def list_contract_scan_subfolders(request: HttpRequest, contract_id: int) -> dict[str, object]:
_require_contract_access(request, contract_id)
return _get_service().list_scan_subfolders(contract_id=contract_id)
@router.get("/{contract_id}/folder-scan/{session_id}", response=ContractFolderScanStatusOut)
def get_contract_scan_status(request: HttpRequest, contract_id: int, session_id: UUID) -> dict[str, object]:
_require_contract_access(request, contract_id)
service = _get_service()
session = service.get_session(contract_id=contract_id, session_id=session_id)
return service.build_status_payload(session=session)
@router.post("/{contract_id}/folder-scan/{session_id}/confirm", response=ContractFolderScanConfirmOut)
@rate_limit_from_settings("TASK", by_user=True)
def confirm_contract_scan(
request: HttpRequest,
contract_id: int,
session_id: UUID,
payload: ContractFolderScanConfirmIn,
) -> dict[str, object]:
_require_contract_access(request, contract_id)
return _get_service().confirm_import(
contract_id=contract_id,
session_id=session_id,
items=[item.model_dump() for item in payload.items],
)
|