File size: 4,771 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 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | from __future__ import annotations
import logging
from typing import Any
from django.http import HttpRequest
from ninja import Router
from apps.contracts.schemas import (
ContractAssignmentOut,
ContractIn,
ContractOut,
ContractPartySourceOut,
ContractPaymentIn,
ContractUpdate,
UpdateLawyersIn,
)
from apps.core.dto.request_context import extract_request_context
logger = logging.getLogger("apps.contracts.api")
router = Router()
def _get_contract_service() -> Any:
from apps.contracts.services.contract.wiring import get_contract_service
return get_contract_service()
def _get_domain_service() -> Any:
from apps.contracts.services.contract.wiring import get_contract_domain_service
return get_contract_domain_service()
@router.get("/contracts", response=list[ContractOut])
def list_contracts(
request: HttpRequest,
case_type: str | None = None,
status: str | None = None,
page: int = 1,
page_size: int = 20,
) -> Any:
"""
获取合同列表（分页）
Requirements: 6.1, 6.2, 6.3
"""
service = _get_domain_service()
ctx = extract_request_context(request)
result = service.list_contracts(
case_type=case_type,
status=status,
user=ctx.user,
org_access=ctx.org_access,
perm_open_access=ctx.perm_open_access,
)
# list_contracts 返回 dict（含分页），前端做客户端分页所以只返回 items
return result["items"] if isinstance(result, dict) else result
class ContractWithCasesIn(ContractIn):
cases: list[dict[str, Any]] | None = None
@router.post("/contracts/full", response=ContractOut)
def create_contract_with_cases(request: HttpRequest, payload: ContractWithCasesIn) -> Any:
service = _get_domain_service()
data = payload.model_dump()
cases_data = data.pop("cases", None)
lawyer_ids = data.pop("lawyer_ids", [])
return service.create_contract_with_cases(
contract_data=data,
cases_data=cases_data,
assigned_lawyer_ids=lawyer_ids,
)
@router.get("/contracts/{contract_id}", response=ContractOut)
def get_contract(request: HttpRequest, contract_id: int) -> Any:
"""
获取合同详情
Requirements: 6.1, 6.2, 6.3
"""
service = _get_domain_service()
ctx = extract_request_context(request)
return service.get_contract(
contract_id=contract_id,
user=ctx.user,
org_access=ctx.org_access,
perm_open_access=ctx.perm_open_access,
)
@router.put("/contracts/{contract_id}", response=ContractOut)
def update_contract(
request: HttpRequest,
contract_id: int,
payload: ContractUpdate,
sync_cases: bool = False,
confirm_finance: bool = False,
new_payments: list[ContractPaymentIn] | None = None,
) -> Any:
service = _get_domain_service()
ctx = extract_request_context(request)
data = payload.model_dump(exclude_unset=True)
return service.update_contract_with_finance(
contract_id=contract_id,
update_data=data,
user=ctx.user,
confirm_finance=confirm_finance,
new_payments=[p.model_dump() for p in new_payments] if new_payments else None,
)
@router.post("/contracts", response=ContractOut)
def create_contract(
request: HttpRequest,
payload: ContractIn,
payments: list[ContractPaymentIn] | None = None,
confirm_finance: bool = False,
) -> Any:
service = _get_domain_service()
ctx = extract_request_context(request)
data = payload.model_dump()
lawyer_ids = data.pop("lawyer_ids", [])
return service.create_contract_with_cases(
contract_data=data,
cases_data=None,
assigned_lawyer_ids=lawyer_ids,
payments_data=[p.model_dump() for p in payments] if payments else None,
confirm_finance=confirm_finance,
user=ctx.user,
)
@router.put("/contracts/{contract_id}/lawyers", response=list[ContractAssignmentOut])
def update_contract_lawyers(request: HttpRequest, contract_id: int, payload: UpdateLawyersIn) -> Any:
service = _get_domain_service()
assignments = service.update_contract_lawyers(contract_id=contract_id, lawyer_ids=payload.lawyer_ids)
return [ContractAssignmentOut.from_assignment(item) for item in assignments]
@router.delete("/contracts/{contract_id}")
def delete_contract(request: HttpRequest, contract_id: int) -> dict[str, bool]:
service = _get_domain_service()
service.delete_contract(contract_id)
return {"success": True}
@router.get("/contracts/{contract_id}/all-parties", response=list[ContractPartySourceOut])
def get_contract_all_parties(request: HttpRequest, contract_id: int) -> Any:
service = _get_domain_service()
return service.get_all_parties(contract_id)
|