File size: 5,000 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 | """当事人 Admin 服务层。"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any
from django.utils.translation import gettext_lazy as _
from apps.client.services.client_admin_file_mixin import _DOC_TYPE_DISPLAY, ClientAdminFileMixin
from apps.core.exceptions import ValidationException
if TYPE_CHECKING:
from apps.client.models import Client
from .client_identity_doc_service import ClientIdentityDocService
from .client_internal_query_service import ClientInternalQueryService
logger = logging.getLogger("apps.client")
@dataclass
class ImportResult:
"""JSON 导入结果"""
success: bool
client: Client | None = None
error_message: str | None = None
class ClientAdminService(ClientAdminFileMixin):
"""客户 Admin 服务：JSON 导入、表单集文件上传、事务管理。"""
def __init__(
self,
identity_doc_service: ClientIdentityDocService | None = None,
internal_query_service: ClientInternalQueryService | None = None,
) -> None:
self._identity_doc_service = identity_doc_service
self._internal_query_service = internal_query_service
@property
def identity_doc_service(self) -> ClientIdentityDocService:
"""延迟获取 ClientIdentityDocService"""
if self._identity_doc_service is None:
from .client_identity_doc_service import ClientIdentityDocService
self._identity_doc_service = ClientIdentityDocService()
return self._identity_doc_service
@property
def internal_query_service(self) -> ClientInternalQueryService:
"""延迟获取 ClientInternalQueryService"""
if self._internal_query_service is None:
from .client_internal_query_service import ClientInternalQueryService
self._internal_query_service = ClientInternalQueryService()
return self._internal_query_service
def import_from_json(self, json_data: dict[str, Any], admin_user: str) -> ImportResult:
"""从 JSON 导入客户，委托给 ClientJsonImporter。"""
from .importer import ClientJsonImporter
importer = ClientJsonImporter()
result = importer.import_from_json(json_data, admin_user=admin_user)
if result.success and result.client_id is not None:
client = self.internal_query_service.get_client(client_id=result.client_id)
return ImportResult(success=True, client=client)
return ImportResult(success=False, error_message=result.error_message)
def process_formset_files(self, client_id: int, formset_data: list[dict[str, Any]], admin_user: str) -> None:
"""处理表单集文件上传。"""
client = self.internal_query_service.get_client(client_id=client_id)
if not client:
raise ValidationException(
message=_("客户不存在"),
code="CLIENT_NOT_FOUND",
errors={"client_id": _("ID 为 %(id)s 的客户不存在") % {"id": client_id}},
)
processed_files = []
for form_data in formset_data:
if self._should_process_form(form_data):
file_info = self._process_single_form(client, form_data, admin_user)
if file_info:
processed_files.append(file_info)
logger.info(
"表单集文件处理完成",
extra={
"client_id": client_id,
"processed_count": len(processed_files),
"admin_user": admin_user,
"action": "process_formset_files",
},
)
def _should_process_form(self, form_data: dict[str, Any]) -> bool:
"""判断是否应该处理该表单项。"""
if form_data.get("DELETE"):
return False
return bool(form_data.get("file_path") or form_data.get("uploaded_file"))
def _process_single_form(self, client: Client, form_data: dict[str, Any], admin_user: str) -> dict[str, Any] | None:
"""处理单个表单项。"""
doc_type = form_data.get("doc_type")
if not doc_type:
logger.warning(
"表单项缺少证件类型",
extra={"client_id": client.pk, "admin_user": admin_user, "action": "process_single_form"},
)
return None
doc_type_display = _DOC_TYPE_DISPLAY.get(doc_type, doc_type)
file_path = self._handle_file_storage(form_data, client.name, doc_type_display)
if not file_path:
return None
doc_id = form_data.get("id")
if doc_id:
self._update_identity_doc(doc_id, file_path, admin_user)
else:
self.identity_doc_service.add_identity_doc(
client_id=client.pk,
doc_type=doc_type,
file_path=file_path,
)
return {"doc_type": doc_type, "file_path": file_path, "doc_id": doc_id}
|