alien_py_cloud/alien_lawyer/api/router.py

from typing import Any, Union, Optional

from fastapi import APIRouter, Depends, Query
from pydantic import ValidationError

from alien_lawyer.api.deps import get_contract_service
from alien_lawyer.schemas.request.contract_lawyer import LawyerTemplatesCreate
from alien_lawyer.schemas.response.contract_lawyer import (
    ModuleStatusResponse,
    TemplatesCreateResponse,
    ErrorResponse,
    PaginatedResponse,
    SuccessResponse,
)
from alien_lawyer.services.contract_server import LawyerContractServer

router = APIRouter()


def _format_validation_errors(exc: ValidationError) -> list[dict[str, str]]:
    errors = []
    for err in exc.errors():
        loc = err.get("loc", ())
        field = ".".join(str(item) for item in loc if item != "body")
        errors.append(
            {
                "field": field or "body",
                "type": err.get("type", "validation_error"),
                "message": err.get("msg", "参数校验失败"),
            }
        )
    return errors


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


@router.post("/get_esign_templates", response_model=Union[TemplatesCreateResponse, ErrorResponse])
async def create_esign_templates(
    templates_data_raw: dict[str, Any],
    templates_server: LawyerContractServer = Depends(get_contract_service),
) -> Union[TemplatesCreateResponse, ErrorResponse]:
    try:
        templates_data = LawyerTemplatesCreate.model_validate(templates_data_raw)
    except ValidationError as exc:
        return ErrorResponse(success=False, message="请求参数校验失败", raw={"errors": _format_validation_errors(exc)})
    result = await templates_server.create_esign_templates(templates_data)
    if not result.get("success"):
        return ErrorResponse(**result)
    return TemplatesCreateResponse(**result)


@router.get("/contracts/{lawyer_id}", response_model=PaginatedResponse)
async def list_contracts(
    lawyer_id: int,
    status: Optional[int] = Query(None, description="筛选合同状态：0 未签署，1 已签署"),
    page: int = Query(1, ge=1, description="页码，从1开始"),
    page_size: int = Query(10, ge=1, le=100, description="每页条数，默认10"),
    templates_server: LawyerContractServer = Depends(get_contract_service),
) -> PaginatedResponse:
    result = await templates_server.list_contracts(lawyer_id, status, page, page_size)
    return PaginatedResponse(**result)


@router.get("/contracts/detail/{sign_flow_id}", response_model=Union[dict, ErrorResponse])
async def get_contract_detail(
    sign_flow_id: str,
    templates_server: LawyerContractServer = Depends(get_contract_service),
) -> Union[dict, ErrorResponse]:
    result = await templates_server.get_contract_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,
    templates_server: LawyerContractServer = Depends(get_contract_service),
) -> Union[SuccessResponse, ErrorResponse]:
    result = await templates_server.process_esign_callback(payload)
    if not result.get("success"):
        return ErrorResponse(**result)
    return SuccessResponse(code=result["code"], msg=result["msg"])