fc / apps /preservation_date /api /preservation_date_api.py
jun2024's picture
Upload 2241 files
b4f2238 verified
Raw History Blame Contribute Delete
2.27 kB
"""Preservation date extraction API."""
import logging
from datetime import datetime
from typing import Any
from django.http import HttpRequest
from ninja import File, Router
from ninja.files import UploadedFile
from apps.core.infrastructure.throttling import rate_limit_from_settings
from apps.preservation_date.services import PreservationDateExtractionService
from apps.preservation_date.services.models import PreservationExtractionResult
logger = logging.getLogger("apps.preservation_date")
router = Router(tags=["财产保全日期识别"])
def _format_date(value: datetime | None) -> str | None:
if value is None:
return None
return value.strftime("%Y-%m-%d")
def _serialize_result(result: PreservationExtractionResult) -> dict[str, Any]:
return {
"success": result.success,
"measures": [
{
"measure_type": item.measure_type,
"property_description": item.property_description,
"duration": item.duration,
"start_date": _format_date(item.start_date),
"end_date": _format_date(item.end_date),
"is_pending": item.is_pending,
"pending_note": item.pending_note,
"raw_text": item.raw_text,
}
for item in result.measures
],
"reminders": [
{
"reminder_type": item.reminder_type,
"content": item.content,
"due_at": _format_date(item.due_at),
"metadata": item.metadata,
}
for item in result.reminders
],
"model_used": result.model_used,
"extraction_method": result.extraction_method,
"error": result.error,
"raw_response": result.raw_response,
}
@router.post("/extract")
@rate_limit_from_settings("UPLOAD", by_user=True)
def extract_preservation_date(
request: HttpRequest,
file: UploadedFile = File(...), # type: ignore[type-arg]
) -> dict[str, Any]:
"""Extract asset preservation measures and expiration dates from a PDF."""
service = PreservationDateExtractionService()
result = service.extract_from_uploaded_file(file.chunks(), file.name)
return _serialize_result(result)