File size: 5,836 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 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | """模拟庭审 REST API."""
import logging
from typing import Any
from django.http import HttpRequest
from ninja import Router, Status
from apps.core.infrastructure.throttling import rate_limit_from_settings
from apps.core.security.auth import JWTOrSessionAuth
from .mock_trial_schemas import (
CreateMockTrialSessionRequest,
ErrorResponse,
MockTrialReportResponse,
MockTrialSessionDetailResponse,
MockTrialSessionListResponse,
MockTrialSessionResponse,
)
logger = logging.getLogger(__name__)
router = Router(tags=["模拟庭审"], auth=JWTOrSessionAuth())
def _get_service() -> Any:
from apps.litigation_ai.services import LitigationConversationService
return LitigationConversationService()
@router.post(
"/sessions",
response={200: MockTrialSessionDetailResponse, 400: ErrorResponse, 403: ErrorResponse},
)
@rate_limit_from_settings("TASK", by_user=True)
def create_session(request: HttpRequest, payload: CreateMockTrialSessionRequest) -> Any:
user = getattr(request, "user", None)
session = _get_service().create_session(
case_id=payload.case_id,
user_id=user.id if user else None,
session_type="mock_trial",
)
return {
"session_id": str(session.session_id),
"case_id": session.case_id,
"session_type": "mock_trial",
"status": session.status,
"metadata": session.metadata,
"created_at": session.created_at,
"updated_at": session.updated_at,
"messages": [],
}
@router.get("/sessions", response={200: MockTrialSessionListResponse, 403: ErrorResponse})
def list_sessions(request: HttpRequest, case_id: int | None = None, limit: int = 20, offset: int = 0) -> Any:
service = _get_service()
user = getattr(request, "user", None)
data = service.list_sessions(
user_id=user.id if user else None,
case_id=case_id,
session_type="mock_trial",
limit=limit,
offset=offset,
)
return {"count": data["total"], "results": data["sessions"]}
@router.get(
"/sessions/{session_id}",
response={200: MockTrialSessionDetailResponse, 404: ErrorResponse},
)
def get_session(request: HttpRequest, session_id: str) -> Any:
service = _get_service()
session = service.get_session(session_id)
messages = service.get_messages(session_id)
return {
"session_id": session.session_id,
"case_id": session.case_id,
"session_type": "mock_trial",
"status": session.status,
"metadata": session.metadata,
"created_at": session.created_at,
"updated_at": session.updated_at,
"messages": [
{"id": m.id, "role": m.role, "content": m.content, "metadata": m.metadata, "created_at": m.created_at}
for m in messages
],
}
@router.get(
"/sessions/{session_id}/report",
response={200: MockTrialReportResponse, 404: ErrorResponse},
)
def get_report(request: HttpRequest, session_id: str) -> Any:
from asgiref.sync import async_to_sync
from apps.litigation_ai.services.mock_trial.report_service import MockTrialReportService
report_data = async_to_sync(MockTrialReportService().get_report)(session_id)
return {
"session_id": session_id,
"mode": report_data.get("mode", ""),
"report": report_data,
}
@router.delete(
"/sessions/{session_id}",
response={204: None, 404: ErrorResponse},
)
def delete_session(request: HttpRequest, session_id: str) -> Any:
service = _get_service()
user = getattr(request, "user", None)
service.delete_session(session_id, user)
return Status(204, None)
@router.get(
"/sessions/{session_id}/export",
response={200: None, 404: ErrorResponse, 500: ErrorResponse},
)
@rate_limit_from_settings("EXPORT", by_user=True)
def export_report(request: HttpRequest, session_id: str) -> Any:
"""导出模拟庭审报告为Word文档."""
from pathlib import Path
from asgiref.sync import async_to_sync
from apps.litigation_ai.services.mock_trial.export_service import MockTrialExportService
from apps.litigation_ai.services.mock_trial.report_service import MockTrialReportService
# 获取报告数据
report_data = async_to_sync(MockTrialReportService().get_report)(session_id)
if not report_data:
return Status(404, {"message": "报告不存在"})
# 获取案件信息
from apps.litigation_ai.services.flow.session_repository import LitigationSessionRepository
repo = LitigationSessionRepository()
session = repo.get_session_sync(session_id)
if not session:
return Status(404, {"message": "会话不存在"})
from apps.litigation_ai.services.wiring import get_case_service
case_dto = get_case_service().get_case(session.case_id)
if not case_dto:
return Status(404, {"message": "案件不存在"})
case_name = case_dto.name or ""
case_info = {
"case_name": case_name,
"cause_of_action": case_dto.cause_of_action or "",
}
# 生成导出文件
export_service = MockTrialExportService()
try:
file_path = export_service.export_to_docx(
session_id=session_id,
report_data=report_data,
case_info=case_info,
)
# 读取文件并返回
from django.http import FileResponse
response = FileResponse(
Path(file_path).open("rb"),
content_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
)
response["Content-Disposition"] = f'attachment; filename="模拟庭审报告_{case_name}.docx"'
return response
except Exception as e:
logger.error(f"导出报告失败: {e}", exc_info=True)
return Status(500, {"message": f"导出失败: {e}"})
|