alien_py_cloud/alien_contract/api/router.py

from typing import Optional, Union, Literal

from fastapi import APIRouter, Depends, Query

from alien_contract.api.deps import get_contract_service
from alien_contract.schemas.request.contract import BundleCreateRequest
from alien_contract.schemas.response.contract import (
    ModuleStatusResponse,
    BundleCreateResponse,
    ErrorResponse,
    PaginatedBundleResponse,
    SuccessResponse,
)
from alien_contract.services.contract_server import ContractCenterService

router = APIRouter()


@router.get("/", response_model=ModuleStatusResponse)
async def index() -> ModuleStatusResponse:
    return ModuleStatusResponse(module="ContractCenter", status="Ok")


@router.post("/bundles", response_model=Union[BundleCreateResponse, ErrorResponse])
async def create_bundle(
    request: BundleCreateRequest,
    service: ContractCenterService = Depends(get_contract_service),
) -> Union[BundleCreateResponse, ErrorResponse]:
    """创建合同包"""
    result = await service.create_bundle(request)
    if not result.get("success"):
        return ErrorResponse(**result)
    return BundleCreateResponse(**result)


@router.get("/bundles", response_model=PaginatedBundleResponse)
async def list_bundles(
    subject_type: Literal["store", "lawyer"] = Query(...),
    subject_id: int = Query(..., gt=0),
    status: Optional[int] = Query(None, ge=0, le=1, description="文档签署状态过滤: 0=未签署, 1=已签署"),
    page: int = Query(1, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    service: ContractCenterService = Depends(get_contract_service),
) -> PaginatedBundleResponse:
    """"查询合同列表"""
    result = await service.list_bundles(subject_type, subject_id, page, page_size, doc_status=status)
    return PaginatedBundleResponse(**result)


@router.get("/documents/{sign_flow_id}", response_model=Union[dict, ErrorResponse])
async def get_document_detail(
    sign_flow_id: str,
    service: ContractCenterService = Depends(get_contract_service),
) -> Union[dict, ErrorResponse]:
    """查询合同详情"""
    result = await service.get_document_detail(sign_flow_id)
    if not result.get("success", True):
        return ErrorResponse(**result)
    return result


@router.post("/esign/callback", response_model=Union[SuccessResponse, ErrorResponse])
async def esign_callback(
    payload: dict,
    service: ContractCenterService = Depends(get_contract_service),
) -> Union[SuccessResponse, ErrorResponse]:
    """e签宝回调"""
    result = await service.process_esign_callback(payload)
    if not result.get("success"):
        return ErrorResponse(**result)
    return SuccessResponse(code=result["code"], msg=result["msg"])