"""案件导出序列化服务。""" from __future__ import annotations from datetime import datetime from typing import TYPE_CHECKING, cast if TYPE_CHECKING: from apps.cases.models import Case, CaseLog from apps.client.models import Client ReminderPayload = dict[str, object] SerializedPayload = dict[str, object] def _serialize_client(client: Client) -> SerializedPayload: from apps.client.services.client_export_serializer_service import serialize_client_obj return cast(SerializedPayload, serialize_client_obj(client)) def _export_case_log_reminders_map(logs: list[CaseLog]) -> dict[int, list[ReminderPayload]]: log_ids = [int(log.id) for log in logs if getattr(log, "id", None)] if not log_ids: return {} from apps.core.interfaces import ServiceLocator reminder_service = ServiceLocator.get_reminder_service() return cast( dict[int, list[ReminderPayload]], reminder_service.export_case_log_reminders_batch_internal(case_log_ids=log_ids), ) def _serialize_exported_reminders(reminders: list[ReminderPayload]) -> list[SerializedPayload]: result: list[SerializedPayload] = [] for reminder in reminders: due_at = reminder.get("due_at") if isinstance(due_at, datetime): due_at_value = due_at.isoformat() elif due_at is None: due_at_value = "" else: due_at_value = str(due_at) metadata = reminder.get("metadata") metadata_value = metadata if isinstance(metadata, dict) else {} result.append( { "reminder_type": reminder.get("reminder_type"), "content": reminder.get("content"), "due_at": due_at_value, "metadata": metadata_value, } ) return result def serialize_case_obj(obj: Case) -> SerializedPayload: """将单个 Case 实例序列化为 dict。""" logs = list(obj.logs.all()) reminders_by_log_id = _export_case_log_reminders_map(logs) return { "name": obj.name, "filing_number": obj.filing_number, "status": obj.status, "case_type": obj.case_type, "cause_of_action": obj.cause_of_action, "target_amount": str(obj.target_amount) if obj.target_amount is not None else None, "preservation_amount": str(obj.preservation_amount) if obj.preservation_amount is not None else None, "current_stage": obj.current_stage, "is_archived": obj.is_archived, "effective_date": str(obj.effective_date) if obj.effective_date else None, "specified_date": str(obj.specified_date) if obj.specified_date else None, "parties": [{"legal_status": p.legal_status, "client": _serialize_client(p.client)} for p in obj.parties.all()], "assignments": [ {"lawyer": {"real_name": a.lawyer.real_name, "phone": a.lawyer.phone, "username": a.lawyer.username}} for a in obj.assignments.all() ], "supervising_authorities": [ {"name": sa.name, "authority_type": sa.authority_type} for sa in obj.supervising_authorities.all() ], "case_numbers": [ {"number": cn.number, "is_active": cn.is_active, "remarks": cn.remarks} for cn in obj.case_numbers.all() ], "chats": [ { "platform": ch.platform, "chat_id": ch.chat_id, "name": ch.name, "is_active": ch.is_active, "owner_id": ch.owner_id, } for ch in obj.chats.all() ], "logs": [ { "content": log.content, "created_at": log.created_at.isoformat(), "actor": {"real_name": log.actor.real_name, "phone": log.actor.phone, "username": log.actor.username}, "attachments": [ {"file_path": att.file.name, "filename": att.file.name.split("/")[-1]} for att in log.attachments.all() if att.file ], "reminders": _serialize_exported_reminders(reminders_by_log_id.get(log.id) or []), } for log in logs ], }