| """Business logic services.""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from importlib import import_module |
| from typing import TYPE_CHECKING, Protocol, TypedDict, cast |
|
|
| from django.db import transaction |
| from django.utils.translation import gettext_lazy as _ |
|
|
| from apps.cases.models import Case, CaseAssignment, CaseNumber, CaseParty, SupervisingAuthority |
|
|
| from .wiring import get_case_filing_number_service, get_document_service |
|
|
| if TYPE_CHECKING: |
| from django.db.models import QuerySet |
|
|
| from apps.cases.services.case_import_service import CaseImportPayload |
| from apps.core.interfaces import ICaseFilingNumberService, IDocumentService |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| JSONDict = dict[str, object] |
| ImportData = dict[str, object] |
|
|
|
|
| class MaterialViewPayload(TypedDict): |
| party_types: list[JSONDict] |
| non_party_types: list[JSONDict] |
| our_parties: list[MaterialPartyPayload] |
| opponent_parties: list[MaterialPartyPayload] |
| authorities: list[AuthorityPayload] |
|
|
|
|
| class SimplePartyPayload(TypedDict): |
| id: int |
| name: str |
|
|
|
|
| class DetailPartyPayload(TypedDict): |
| id: int |
| name: str |
| client_type: str |
| legal_status: str | None |
| legal_status_display: str |
|
|
|
|
| class MaterialPartyPayload(TypedDict): |
| id: int |
| name: str |
| legal_status: str | None |
| legal_status_display: str |
|
|
|
|
| class AuthorityPayload(TypedDict): |
| id: int |
| name: str |
| authority_type: str |
| authority_type_display: str |
|
|
|
|
| class CaseMaterialServiceProtocol(Protocol): |
| def get_used_type_ids(self, *, case_id: int) -> list[int]: ... |
|
|
| def get_material_types_by_category( |
| self, |
| *, |
| category: str, |
| law_firm_id: int | None, |
| used_type_ids: list[int], |
| ) -> list[JSONDict]: ... |
|
|
|
|
| class CaseImportServiceProtocol(Protocol): |
| def import_one(self, data: CaseImportPayload) -> Case: ... |
|
|
|
|
| class CaseAdminService: |
| """案件 Admin 服务""" |
|
|
| def __init__( |
| self, |
| document_service: IDocumentService | None = None, |
| filing_number_service: ICaseFilingNumberService | None = None, |
| ) -> None: |
| """ |
| 构造函数支持依赖注入 |
| |
| document_service: 文档服务实例(可选,用于依赖注入) |
| filing_number_service: 建档编号服务实例(可选,用于依赖注入) |
| """ |
| self._document_service = document_service |
| self._filing_number_service = filing_number_service |
|
|
| @property |
| def document_service(self) -> IDocumentService: |
| """ |
| 延迟加载文档服务 |
| |
| 通过 ServiceLocator 获取 IDocumentService 实例, |
| 支持依赖注入以便于测试. |
| |
| IDocumentService 实例 |
| """ |
| if self._document_service is None: |
| self._document_service = get_document_service() |
| return self._document_service |
|
|
| @property |
| def filing_number_service(self) -> ICaseFilingNumberService: |
| """ |
| 延迟加载建档编号服务 |
| |
| 支持依赖注入以便于测试. |
| |
| FilingNumberService 实例 |
| """ |
| if self._filing_number_service is None: |
| self._filing_number_service = get_case_filing_number_service() |
| return self._filing_number_service |
|
|
| def get_matched_folder_templates(self, case_type: str, legal_statuses: list[str] | None = None) -> str: |
| """ |
| 获取匹配的文件夹模板 |
| |
| case_type: 案件类型 |
| legal_statuses: 案件的诉讼地位列表(我方当事人的诉讼地位) |
| |
| 模板名称字符串,多个模板用"、"分隔 |
| 如果查询失败返回 "查询失败" |
| """ |
| try: |
| if legal_statuses: |
| return cast( |
| str, |
| self.document_service.get_matched_folder_templates_with_legal_status(case_type, legal_statuses), |
| ) |
| return cast(str, self.document_service.get_matched_folder_templates(case_type)) |
| except Exception: |
| logger.exception( |
| "get_matched_folder_templates_failed", extra={"case_type": case_type, "legal_statuses": legal_statuses} |
| ) |
| return str(_("查询失败")) |
|
|
| def get_matched_folder_templates_list( |
| self, case_type: str, legal_statuses: list[str] | None = None |
| ) -> list[JSONDict]: |
| try: |
| module = import_module("apps.documents.services.template.template_matching_service") |
| template_matching_service_cls = module.TemplateMatchingService |
|
|
| return cast( |
| list[JSONDict], |
| template_matching_service_cls().find_matching_case_folder_templates_list( |
| case_type=case_type, |
| legal_statuses=legal_statuses, |
| ), |
| ) |
| except Exception: |
| logger.exception("get_matched_folder_templates_list_failed", extra={"case_type": case_type}) |
| return [] |
|
|
| def get_matched_case_file_templates( |
| self, |
| case_type: str, |
| case_stage: str, |
| applicable_institutions: list[str] | None = None, |
| ) -> list[JSONDict]: |
| try: |
| return cast( |
| list[JSONDict], |
| self.document_service.find_matching_case_file_templates( |
| case_type=case_type, |
| case_stage=case_stage, |
| applicable_institutions=applicable_institutions, |
| ), |
| ) |
| except Exception: |
| logger.exception( |
| "get_matched_case_file_templates_failed", |
| extra={ |
| "case_type": case_type, |
| "case_stage": case_stage, |
| "applicable_institutions": applicable_institutions, |
| }, |
| ) |
| return [] |
|
|
| def get_case_file_sub_type_choices(self) -> list[tuple[str, str]]: |
| """获取案件文件子类型选项。""" |
| try: |
| choices_module = import_module("apps.documents.models.choices") |
| return list(choices_module.DocumentCaseFileSubType.choices) |
| except Exception: |
| logger.exception("get_case_file_sub_type_choices_failed") |
| return [] |
|
|
| def get_case_file_templates_for_detail(self, case: Case) -> tuple[list[JSONDict], str]: |
| """获取详情页案件文件模板与缺失原因。""" |
| if not case.case_type: |
| return [], str(_("未设置案件类型")) |
| if not case.current_stage: |
| return [], str(_("未设置案件阶段")) |
| applicable_institutions = self._get_case_applicable_institutions(case) |
| return ( |
| self.get_matched_case_file_templates( |
| case_type=case.case_type, |
| case_stage=case.current_stage, |
| applicable_institutions=applicable_institutions, |
| ), |
| "", |
| ) |
|
|
| def _get_case_applicable_institutions(self, case: Case) -> list[str]: |
| names: list[str] = [] |
| seen: set[str] = set() |
| for authority in case.supervising_authorities.all(): |
| name = str(getattr(authority, "name", "") or "").strip() |
| if not name or name in seen: |
| continue |
| seen.add(name) |
| names.append(name) |
| return names |
|
|
| def build_our_legal_entities(self, case: Case) -> list[SimplePartyPayload]: |
| """构建我方主体（法人）视图数据。""" |
| return [ |
| {"id": party.client.id, "name": party.client.name} |
| for party in case.parties.all() |
| if getattr(party.client, "is_our_client", False) and getattr(party.client, "client_type", "") == "legal" |
| ] |
|
|
| def build_our_parties(self, case: Case) -> list[DetailPartyPayload]: |
| """构建我方当事人视图数据。""" |
| parties: list[DetailPartyPayload] = [] |
| for party in case.parties.all(): |
| client = party.client |
| if not getattr(client, "is_our_client", False): |
| continue |
| parties.append( |
| { |
| "id": client.id, |
| "name": client.name, |
| "client_type": getattr(client, "client_type", "") or "", |
| "legal_status": getattr(party, "legal_status", None), |
| "legal_status_display": ( |
| getattr(party, "get_legal_status_display", lambda: "")() |
| if getattr(party, "legal_status", None) |
| else "" |
| ), |
| } |
| ) |
| return parties |
|
|
| def build_respondents(self, case: Case) -> list[SimplePartyPayload]: |
| """构建对方当事人视图数据。""" |
| return [ |
| {"id": party.client.id, "name": party.client.name} |
| for party in case.parties.all() |
| if not getattr(party.client, "is_our_client", False) |
| ] |
|
|
| def build_material_view_parties(self, case: Case) -> tuple[list[MaterialPartyPayload], list[MaterialPartyPayload]]: |
| """构建材料页我方/对方当事人数据。""" |
| our_parties: list[MaterialPartyPayload] = [] |
| opponent_parties: list[MaterialPartyPayload] = [] |
| for party in case.parties.all(): |
| client = party.client |
| item: MaterialPartyPayload = { |
| "id": party.id, |
| "name": getattr(client, "name", "") or "", |
| "legal_status": getattr(party, "legal_status", None), |
| "legal_status_display": ( |
| getattr(party, "get_legal_status_display", lambda: "")() |
| if getattr(party, "legal_status", None) |
| else "" |
| ), |
| } |
| if getattr(client, "is_our_client", False): |
| our_parties.append(item) |
| else: |
| opponent_parties.append(item) |
| return our_parties, opponent_parties |
|
|
| def build_material_view_authorities(self, case: Case) -> list[AuthorityPayload]: |
| """构建材料页主管机关数据。""" |
| return [ |
| { |
| "id": authority.id, |
| "name": authority.name or "", |
| "authority_type": authority.authority_type or "", |
| "authority_type_display": authority.get_authority_type_display() if authority.authority_type else "", |
| } |
| for authority in case.supervising_authorities.all().order_by("created_at") |
| ] |
|
|
| def get_case_with_admin_relations(self, case_id: int) -> Case | None: |
| """按 admin 详情页需求查询案件及关联数据。""" |
| from django.db.models import Prefetch |
|
|
| from apps.cases.models import CaseLog |
|
|
| try: |
| return cast( |
| Case, |
| Case.objects.select_related( |
| "contract", |
| "folder_binding", |
| ) |
| .prefetch_related( |
| "case_numbers", |
| "supervising_authorities", |
| "parties__client", |
| "assignments__lawyer", |
| Prefetch( |
| "logs", |
| queryset=CaseLog.objects.select_related("actor") |
| .prefetch_related("attachments") |
| .order_by("-created_at"), |
| ), |
| "chats", |
| ) |
| .get(pk=case_id), |
| ) |
| except Case.DoesNotExist: |
| return None |
|
|
| def build_materials_view_payload( |
| self, |
| *, |
| case: Case, |
| material_service: CaseMaterialServiceProtocol, |
| law_firm_id: int | None, |
| ) -> MaterialViewPayload: |
| """构建材料页所需数据。""" |
| used_type_ids = material_service.get_used_type_ids(case_id=case.id) |
| party_types = material_service.get_material_types_by_category( |
| category="party", |
| law_firm_id=law_firm_id, |
| used_type_ids=used_type_ids, |
| ) |
| non_party_types = material_service.get_material_types_by_category( |
| category="non_party", |
| law_firm_id=law_firm_id, |
| used_type_ids=used_type_ids, |
| ) |
| our_parties, opponent_parties = self.build_material_view_parties(case) |
| authorities = self.build_material_view_authorities(case) |
| return { |
| "party_types": party_types, |
| "non_party_types": non_party_types, |
| "our_parties": our_parties, |
| "opponent_parties": opponent_parties, |
| "authorities": authorities, |
| } |
|
|
| def group_templates_by_sub_type( |
| self, |
| templates: list[JSONDict], |
| sub_type_choices: list[tuple[str, str]], |
| *, |
| exclude_special_sub_types: bool = True, |
| ) -> list[tuple[str, list[JSONDict]]]: |
| """按案件文件子类型分组模板。""" |
| special_sub_types = {"power_of_attorney_materials", "property_preservation_materials"} |
| label_map = dict(sub_type_choices) |
| groups: dict[str, list[JSONDict]] = {} |
| for template in templates: |
| sub_type = cast(str, template.get("case_sub_type", "other_materials")) |
| if exclude_special_sub_types and sub_type in special_sub_types: |
| continue |
| groups.setdefault(sub_type, []).append(template) |
| order = [choice[0] for choice in sub_type_choices] |
| order_set = set(order) |
| ordered_keys = [key for key in order if key in groups] |
| extra_keys = sorted(key for key in groups.keys() if key not in order_set) |
| ordered_keys.extend(extra_keys) |
| return [(label_map.get(key, key), groups[key]) for key in ordered_keys] |
|
|
| def detect_special_template_flags(self, unified_templates: list[JSONDict]) -> tuple[bool, bool]: |
| """识别详情页特殊模板标记。""" |
| has_preservation_template = any( |
| template.get("function_code") == "preservation_application" |
| or "财产保全申请书" in cast(str, template.get("name") or "") |
| for template in unified_templates |
| ) |
| has_delay_delivery_template = any( |
| template.get("function_code") == "delay_delivery_application" |
| or "暂缓送达申请书" in cast(str, template.get("name") or "") |
| for template in unified_templates |
| ) |
| return has_preservation_template, has_delay_delivery_template |
|
|
| def serialize_queryset_for_export(self, queryset: QuerySet[Case]) -> list[JSONDict]: |
| """序列化案件列表用于 Admin 导出。""" |
| from apps.cases.services.case.case_admin_export_bridge import get_case_admin_export_prefetches |
| from apps.cases.services.case.case_contract_export_bridge import ( |
| get_case_admin_contract_export_prefetches, |
| serialize_contract_for_case_export, |
| ) |
| from apps.cases.services.case.case_export_serializer_service import serialize_case_obj |
|
|
| result: list[JSONDict] = [] |
| for case in queryset.prefetch_related( |
| *get_case_admin_export_prefetches(), |
| *get_case_admin_contract_export_prefetches(), |
| ): |
| data = serialize_case_obj(case) |
| data["contract"] = serialize_contract_for_case_export(case.contract) if case.contract else None |
| result.append(data) |
| return result |
|
|
| def collect_file_paths_for_export(self, queryset: QuerySet[Case]) -> list[str]: |
| """收集案件导出相关文件路径。""" |
| from apps.cases.services.case.case_admin_export_bridge import ( |
| collect_case_file_paths_for_export, |
| get_case_admin_file_prefetches, |
| ) |
| from apps.cases.services.case.case_contract_export_bridge import ( |
| collect_contract_file_paths_for_case_export, |
| get_case_admin_contract_file_prefetches, |
| ) |
|
|
| seen: set[str] = set() |
| paths: list[str] = [] |
|
|
| def _add(path: str) -> None: |
| if path and path not in seen: |
| seen.add(path) |
| paths.append(path) |
|
|
| for case in queryset.prefetch_related( |
| *get_case_admin_file_prefetches(), |
| *get_case_admin_contract_file_prefetches(), |
| ): |
| if case.contract: |
| collect_contract_file_paths_for_case_export(case.contract, _add) |
| collect_case_file_paths_for_export(case, _add) |
| return paths |
|
|
| def import_cases_from_json_data( |
| self, |
| data_list: list[ImportData], |
| *, |
| case_import_service: CaseImportServiceProtocol, |
| ) -> tuple[int, int, list[str]]: |
| """执行 Admin JSON 导入并返回成功/跳过/错误统计。""" |
| success = skipped = 0 |
| errors: list[str] = [] |
|
|
| for index, item in enumerate(data_list, 1): |
| try: |
| filing_number = item.get("filing_number") |
| before = Case.objects.filter(filing_number=filing_number).exists() if filing_number else False |
| case_import_service.import_one(item) |
| if before: |
| skipped += 1 |
| else: |
| success += 1 |
| except Exception as exc: |
| logger.exception("导入案件失败", extra={"index": index, "case_name": item.get("name", "?")}) |
| errors.append(f"[{index}] {item.get('name', '?')} ({type(exc).__name__}): {exc}") |
|
|
| return success, skipped, errors |
|
|
| @transaction.atomic |
| def duplicate_case(self, case_id: int) -> Case: |
| """ |
| 复制案件及其所有关联数据(不复制日志和群聊) |
| |
| case_id: 原案件ID |
| |
| 新创建的案件对象 |
| """ |
| |
| original = Case.objects.get(pk=case_id) |
|
|
| |
| new_case = Case.objects.create( |
| contract=original.contract, |
| is_archived=False, |
| name=f"{original.name} (副本)", |
| status=original.status, |
| effective_date=original.effective_date, |
| cause_of_action=original.cause_of_action, |
| target_amount=original.target_amount, |
| case_type=original.case_type, |
| current_stage=original.current_stage, |
| ) |
|
|
| |
| parties_to_create = [ |
| CaseParty( |
| case=new_case, |
| client=party.client, |
| legal_status=party.legal_status, |
| ) |
| for party in original.parties.all() |
| ] |
| CaseParty.objects.bulk_create(parties_to_create) |
|
|
| |
| assignments_to_create = [ |
| CaseAssignment( |
| case=new_case, |
| lawyer=assignment.lawyer, |
| ) |
| for assignment in original.assignments.all() |
| ] |
| CaseAssignment.objects.bulk_create(assignments_to_create) |
|
|
| |
| authorities_to_create = [ |
| SupervisingAuthority( |
| case=new_case, |
| name=authority.name, |
| authority_type=authority.authority_type, |
| ) |
| for authority in original.supervising_authorities.all() |
| ] |
| SupervisingAuthority.objects.bulk_create(authorities_to_create) |
|
|
| |
| case_numbers_to_create = [ |
| CaseNumber( |
| case=new_case, |
| number=case_number.number, |
| remarks=case_number.remarks, |
| ) |
| for case_number in original.case_numbers.all() |
| ] |
| CaseNumber.objects.bulk_create(case_numbers_to_create) |
|
|
| |
|
|
| return new_case |
|
|
| @transaction.atomic |
| def handle_case_filing_change(self, case_id: int, is_archived: bool) -> str | None: |
| """ |
| 处理案件建档状态变化 |
| |
| 业务逻辑: |
| - 如果 is_archived=True 且 filing_number 为空,调用 FilingNumberService 生成编号 |
| - 如果 is_archived=True 且 filing_number 已存在,返回现有编号 |
| - 如果 is_archived=False,不修改 filing_number(保留在数据库中) |
| |
| case_id: 案件ID |
| is_archived: 是否已建档 |
| |
| str | None: 建档编号(如果已建档) |
| |
| NotFoundError: 案件不存在 |
| ValidationException: 数据验证失败 |
| |
| Requirements: 5.1, 5.2, 6.1, 6.2, 6.3, 6.4 |
| """ |
| from apps.core.exceptions import NotFoundError |
|
|
| try: |
| case = Case.objects.get(pk=case_id) |
| except Case.DoesNotExist: |
| raise NotFoundError( |
| message=_("案件不存在"), code="CASE_NOT_FOUND", errors={"case_id": f"ID为 {case_id} 的案件不存在"} |
| ) from None |
|
|
| |
| if not is_archived: |
| logger.info( |
| "取消案件建档,保留建档编号", |
| extra={"case_id": case_id, "filing_number": case.filing_number, "action": "handle_case_filing_change"}, |
| ) |
| return None |
|
|
| |
| if case.filing_number: |
| logger.info( |
| "案件已有建档编号,返回现有编号", |
| extra={"case_id": case_id, "filing_number": case.filing_number, "action": "handle_case_filing_change"}, |
| ) |
| return cast(str, case.filing_number) |
|
|
| |
| created_year = case.start_date.year |
| case_type = str(case.case_type or "") |
| filing_number = cast( |
| str, |
| self.filing_number_service.generate_case_filing_number_internal( |
| case_id=case_id, |
| case_type=case_type, |
| created_year=created_year, |
| ), |
| ) |
|
|
| |
| case.filing_number = filing_number |
| case.save(update_fields=["filing_number"]) |
|
|
| logger.info( |
| "生成并保存案件建档编号", |
| extra={"case_id": case_id, "filing_number": filing_number, "action": "handle_case_filing_change"}, |
| ) |
|
|
| return filing_number |
|
|
