| """案件文件夹自动捕获服务。""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| import os |
| import re |
| from pathlib import Path |
| from typing import Any |
| from urllib.parse import urlencode |
| from uuid import UUID |
|
|
| from django.core.files.uploadedfile import SimpleUploadedFile |
| from django.db import transaction |
| from django.urls import reverse |
| from django.utils import timezone |
| from django.utils.translation import gettext_lazy as _ |
|
|
| from apps.cases.models import Case, CaseFolderBinding, CaseFolderScanSession, CaseFolderScanStatus |
| from apps.cases.services.log.caselog_service import CaseLogService |
| from apps.core.dependencies.core import build_task_submission_service |
| from apps.core.exceptions import NotFoundError, ValidationException |
| from apps.core.services.bound_folder_scan_service import BoundFolderScanService |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class CaseFolderScanService: |
| """案件自动捕获扫描、轮询、导入附件服务。""" |
|
|
| _FORCE_OUR_PARTY_FOLDER_KEYWORDS = ( |
| "立案材料", |
| "递交给法院的资料", |
| "提交给法院的资料", |
| ) |
| _FORCE_OUR_PARTY_REASON = "命中目录规则：立案材料/提交给法院的资料目录默认归类为我方当事人材料" |
|
|
| _ACTIVE_STATUSES = { |
| CaseFolderScanStatus.PENDING, |
| CaseFolderScanStatus.RUNNING, |
| CaseFolderScanStatus.CLASSIFYING, |
| } |
|
|
| def __init__(self, *, scan_service: BoundFolderScanService | None = None) -> None: |
| self._scan_service = scan_service or BoundFolderScanService() |
| self._case_log_service = CaseLogService() |
|
|
| def start_scan( |
| self, |
| *, |
| case_id: int, |
| started_by: Any | None, |
| rescan: bool = False, |
| scan_subfolder: str = "", |
| enable_recognition: bool = False, |
| ) -> CaseFolderScanSession: |
| self._ensure_case_exists(case_id) |
| binding = self._get_accessible_binding(case_id) |
| scan_scope = self._resolve_scan_scope(binding.folder_path, scan_subfolder) |
| scan_options = {"enable_recognition": bool(enable_recognition)} |
|
|
| if not rescan: |
| existing = ( |
| CaseFolderScanSession.objects.filter(case_id=case_id, status__in=self._ACTIVE_STATUSES) |
| .order_by("-created_at") |
| .first() |
| ) |
| if existing: |
| existing_subfolder = self._extract_scan_subfolder(existing.result_payload) |
| existing_enable_recognition = self._extract_enable_recognition(existing.result_payload) |
| if ( |
| existing_subfolder == scan_scope["scan_subfolder"] |
| and existing_enable_recognition == scan_options["enable_recognition"] |
| ): |
| return existing |
| raise ValidationException( |
| message=_("已有进行中的扫描任务，请等待完成或使用“重新扫描”"), |
| errors={"session_id": str(existing.id)}, |
| ) |
|
|
| session = CaseFolderScanSession.objects.create( |
| case_id=case_id, |
| status=CaseFolderScanStatus.PENDING, |
| progress=0, |
| current_file="", |
| result_payload={ |
| "summary": {}, |
| "candidates": [], |
| "scan_scope": scan_scope, |
| "scan_options": scan_options, |
| }, |
| started_by=started_by if getattr(started_by, "is_authenticated", False) else None, |
| ) |
|
|
| task_id = build_task_submission_service().submit( |
| "apps.cases.services.material.folder_scan_service.run_case_folder_scan_task", |
| args=[str(session.id)], |
| task_name=f"case_folder_scan_{session.id}", |
| ) |
|
|
| CaseFolderScanSession.objects.filter(id=session.id).update( |
| status=CaseFolderScanStatus.RUNNING, |
| task_id=str(task_id), |
| updated_at=timezone.now(), |
| ) |
| session.refresh_from_db() |
| logger.info( |
| "case_folder_scan_submitted", |
| extra={ |
| "case_id": case_id, |
| "session_id": str(session.id), |
| "scan_subfolder": scan_scope["scan_subfolder"], |
| "enable_recognition": scan_options["enable_recognition"], |
| }, |
| ) |
| return session |
|
|
| def list_scan_subfolders(self, *, case_id: int) -> dict[str, Any]: |
| self._ensure_case_exists(case_id) |
| binding = self._get_accessible_binding(case_id) |
| root = Path(binding.folder_path).expanduser().resolve() |
|
|
| subfolders: list[dict[str, str]] = [] |
| for child in sorted(root.iterdir(), key=lambda item: item.name.lower()): |
| if child.name.startswith("."): |
| continue |
| if not child.is_dir(): |
| continue |
| resolved_child = child.resolve() |
| if not self._is_within_root(root, resolved_child): |
| continue |
| subfolders.append( |
| { |
| "relative_path": child.name, |
| "display_name": child.name, |
| } |
| ) |
|
|
| return { |
| "root_path": root.as_posix(), |
| "subfolders": subfolders, |
| } |
|
|
| def get_session(self, *, case_id: int, session_id: UUID) -> CaseFolderScanSession: |
| try: |
| return CaseFolderScanSession.objects.get(id=session_id, case_id=case_id) |
| except CaseFolderScanSession.DoesNotExist: |
| raise NotFoundError(_("扫描会话不存在")) from None |
|
|
| def build_status_payload(self, *, session: CaseFolderScanSession) -> dict[str, Any]: |
| payload = dict(session.result_payload or {}) |
| summary = payload.get("summary") or {} |
| candidates = self._normalize_candidates_for_scan_scope(payload.get("candidates") or [], payload) |
| stage_result = payload.get("stage_result") or {} |
| prefill_map = stage_result.get("prefill_map") or {} |
| scan_subfolder = self._extract_scan_subfolder(payload) |
| enable_recognition = self._extract_enable_recognition(payload) |
|
|
| return { |
| "session_id": str(session.id), |
| "status": session.status, |
| "progress": int(session.progress or 0), |
| "current_file": session.current_file or "", |
| "scan_subfolder": scan_subfolder, |
| "enable_recognition": enable_recognition, |
| "summary": { |
| "total_files": int(summary.get("total_files", 0) or 0), |
| "deduped_files": int(summary.get("deduped_files", 0) or 0), |
| "classified_files": int(summary.get("classified_files", 0) or 0), |
| }, |
| "candidates": candidates, |
| "error_message": session.error_message or "", |
| "prefill_map": prefill_map, |
| } |
|
|
| @transaction.atomic |
| def stage_to_attachments( |
| self, |
| *, |
| case_id: int, |
| session_id: UUID, |
| items: list[dict[str, Any]], |
| user: Any | None, |
| org_access: dict[str, Any] | None, |
| perm_open_access: bool, |
| ) -> dict[str, Any]: |
| session = self.get_session(case_id=case_id, session_id=session_id) |
| if session.status not in {CaseFolderScanStatus.COMPLETED, CaseFolderScanStatus.STAGED}: |
| raise ValidationException(message=_("扫描尚未完成"), errors={"status": session.status}) |
|
|
| payload = dict(session.result_payload or {}) |
| candidates = payload.get("candidates") or [] |
| candidate_map = {str(item.get("source_path") or ""): item for item in candidates} |
|
|
| selected_items = [item for item in items if bool(item.get("selected", True))] |
| if not selected_items: |
| raise ValidationException(message=_("未找到可导入的 PDF")) |
|
|
| log = self._case_log_service.create_log( |
| case_id=case_id, |
| content=str(_("自动捕获材料")), |
| user=user, |
| org_access=org_access, |
| perm_open_access=perm_open_access, |
| ) |
|
|
| uploads: list[SimpleUploadedFile] = [] |
| prefill_entries: list[dict[str, Any]] = [] |
|
|
| for item in selected_items: |
| source_path = str(item.get("source_path") or "").strip() |
| if not source_path or source_path not in candidate_map: |
| raise ValidationException(message=_("候选文件不存在"), errors={"source_path": source_path}) |
|
|
| file_path = Path(source_path) |
| if not file_path.exists() or not file_path.is_file(): |
| raise ValidationException(message=_("源文件不存在"), errors={"source_path": source_path}) |
|
|
| uploads.append( |
| SimpleUploadedFile( |
| name=file_path.name, |
| content=file_path.read_bytes(), |
| content_type="application/pdf", |
| ) |
| ) |
|
|
| category = str(item.get("category") or "").strip() |
| if category not in {"party", "non_party"}: |
| category = "" |
|
|
| side = str(item.get("side") or "").strip() |
| if category != "party" or side not in {"our", "opponent"}: |
| side = "" |
|
|
| supervising_authority_id: int | None = None |
| if category == "non_party": |
| supervising_authority_id = self._to_int(item.get("supervising_authority_id")) |
|
|
| party_ids: list[int] = [] |
| if category == "party": |
| raw_party_ids = item.get("party_ids") or [] |
| if isinstance(raw_party_ids, list): |
| seen_party_ids: set[int] = set() |
| for raw_pid in raw_party_ids: |
| pid = self._to_int(raw_pid) |
| if pid and pid not in seen_party_ids: |
| seen_party_ids.add(pid) |
| party_ids.append(pid) |
|
|
| prefill_entries.append( |
| { |
| "category": category, |
| "side": side, |
| "type_name_hint": str(item.get("type_name_hint") or "").strip(), |
| "supervising_authority_id": supervising_authority_id, |
| "party_ids": party_ids, |
| } |
| ) |
|
|
| created_attachments = self._case_log_service.upload_attachments( |
| log_id=log.id, |
| files=uploads, |
| user=user, |
| org_access=org_access, |
| perm_open_access=perm_open_access, |
| ) |
|
|
| prefill_map: dict[str, dict[str, Any]] = {} |
| for attachment, prefill in zip(created_attachments, prefill_entries, strict=True): |
| prefill_map[str(attachment.id)] = prefill |
|
|
| materials_url = self._build_materials_url(case_id=case_id, session_id=session_id) |
| stage_result = { |
| "log_id": int(log.id), |
| "attachment_ids": [int(att.id) for att in created_attachments], |
| "prefill_map": prefill_map, |
| "materials_url": materials_url, |
| "staged_at": timezone.now().isoformat(), |
| } |
|
|
| payload["stage_result"] = stage_result |
|
|
| CaseFolderScanSession.objects.filter(id=session.id).update( |
| status=CaseFolderScanStatus.STAGED, |
| progress=100, |
| current_file="", |
| result_payload=payload, |
| error_message="", |
| updated_at=timezone.now(), |
| ) |
|
|
| return { |
| "session_id": str(session.id), |
| "status": CaseFolderScanStatus.STAGED, |
| "log_id": int(log.id), |
| "attachment_ids": [int(att.id) for att in created_attachments], |
| "materials_url": materials_url, |
| "prefill_map": prefill_map, |
| } |
|
|
| def run_scan_task(self, *, session_id: str) -> None: |
| session = CaseFolderScanSession.objects.select_related("case").filter(id=session_id).first() |
| if not session: |
| logger.warning("case_folder_scan_session_missing", extra={"session_id": session_id}) |
| return |
|
|
| try: |
| binding = self._get_accessible_binding(session.case_id) |
| payload = dict(session.result_payload or {}) |
| scan_scope = self._resolve_scan_scope( |
| binding.folder_path, |
| self._extract_scan_subfolder(payload), |
| ) |
| enable_recognition = self._extract_enable_recognition(payload) |
| classification_context = self._build_classification_context(session.case) |
|
|
| def _progress(status: str, progress: int, current_file: str | None) -> None: |
| mapped_status = CaseFolderScanStatus.RUNNING |
| if status == "classifying": |
| mapped_status = CaseFolderScanStatus.CLASSIFYING |
| elif status == "completed": |
| mapped_status = CaseFolderScanStatus.COMPLETED |
|
|
| CaseFolderScanSession.objects.filter(id=session.id).update( |
| status=mapped_status, |
| progress=int(progress), |
| current_file=current_file or "", |
| updated_at=timezone.now(), |
| ) |
|
|
| result = self._scan_service.scan_folder( |
| folder_path=scan_scope["scan_folder"], |
| domain="case", |
| progress_callback=_progress, |
| enable_recognition=enable_recognition, |
| classification_context=classification_context, |
| scan_subfolder=scan_scope["scan_subfolder"], |
| ) |
| result["scan_scope"] = scan_scope |
| result["scan_options"] = {"enable_recognition": enable_recognition} |
|
|
| CaseFolderScanSession.objects.filter(id=session.id).update( |
| status=CaseFolderScanStatus.COMPLETED, |
| progress=100, |
| current_file="", |
| result_payload=result, |
| error_message="", |
| updated_at=timezone.now(), |
| ) |
| except Exception as exc: |
| logger.exception("case_folder_scan_failed", extra={"session_id": session_id}) |
| CaseFolderScanSession.objects.filter(id=session.id).update( |
| status=CaseFolderScanStatus.FAILED, |
| error_message=str(exc), |
| updated_at=timezone.now(), |
| ) |
|
|
| @staticmethod |
| def _ensure_case_exists(case_id: int) -> None: |
| if Case.objects.filter(id=case_id).exists(): |
| return |
| raise NotFoundError(_("案件不存在")) |
|
|
| @staticmethod |
| def _get_accessible_binding(case_id: int) -> CaseFolderBinding: |
| binding = CaseFolderBinding.objects.filter(case_id=case_id).first() |
| if not binding: |
| raise ValidationException(message=_("未绑定文件夹"), errors={"case_id": case_id}) |
|
|
| folder = Path(binding.folder_path) |
| if not folder.exists() or not folder.is_dir(): |
| raise ValidationException(message=_("绑定文件夹不可访问"), errors={"folder_path": binding.folder_path}) |
|
|
| return binding |
|
|
| def _extract_scan_subfolder(self, payload: dict[str, Any] | None) -> str: |
| scope = (payload or {}).get("scan_scope") or {} |
| return str(scope.get("scan_subfolder") or "").strip() |
|
|
| def _normalize_candidates_for_scan_scope( |
| self, |
| candidates: list[dict[str, Any]], |
| payload: dict[str, Any] | None, |
| ) -> list[dict[str, Any]]: |
| if not candidates: |
| return [] |
|
|
| should_force_for_scope = self._should_force_our_party_for_filing_materials(payload) |
| normalized: list[dict[str, Any]] = [] |
| for item in candidates: |
| candidate = dict(item or {}) |
| should_force_for_candidate = self._should_force_our_party_for_candidate(candidate) |
| if not should_force_for_scope and not should_force_for_candidate: |
| normalized.append(candidate) |
| continue |
| candidate["suggested_category"] = "party" |
| candidate["suggested_side"] = "our" |
| if not str(candidate.get("reason") or "").strip(): |
| candidate["reason"] = self._FORCE_OUR_PARTY_REASON |
| normalized.append(candidate) |
| return normalized |
|
|
| def _should_force_our_party_for_filing_materials(self, payload: dict[str, Any] | None) -> bool: |
| scope = (payload or {}).get("scan_scope") or {} |
| scan_subfolder = str(scope.get("scan_subfolder") or "").strip() |
| scan_folder = str(scope.get("scan_folder") or "").strip() |
| return self._contains_force_our_party_folder_keyword( |
| scan_subfolder |
| ) or self._contains_force_our_party_folder_keyword(scan_folder) |
|
|
| def _should_force_our_party_for_candidate(self, candidate: dict[str, Any] | None) -> bool: |
| source_path = str((candidate or {}).get("source_path") or "").strip() |
| if not source_path: |
| return False |
| return self._contains_force_our_party_folder_keyword(source_path) |
|
|
| @classmethod |
| def _contains_force_our_party_folder_keyword(cls, text: str) -> bool: |
| normalized = str(text or "").strip() |
| if not normalized: |
| return False |
| return any(keyword in normalized for keyword in cls._FORCE_OUR_PARTY_FOLDER_KEYWORDS) |
|
|
| def _extract_enable_recognition(self, payload: dict[str, Any] | None) -> bool: |
| options = (payload or {}).get("scan_options") or {} |
| if "enable_recognition" not in options: |
| return True |
| return bool(options.get("enable_recognition")) |
|
|
| def _resolve_scan_scope(self, root_folder: str, scan_subfolder: str) -> dict[str, str]: |
| root = Path(root_folder).expanduser().resolve() |
| normalized_subfolder = self._normalize_scan_subfolder(scan_subfolder) |
|
|
| scan_path = root |
| if normalized_subfolder: |
| scan_path = (root / normalized_subfolder).resolve() |
| if not self._is_within_root(root, scan_path): |
| raise ValidationException( |
| message=_("扫描子文件夹越界"), |
| errors={"scan_subfolder": normalized_subfolder}, |
| ) |
| if not scan_path.exists() or not scan_path.is_dir(): |
| raise ValidationException( |
| message=_("扫描子文件夹不可访问"), |
| errors={"scan_subfolder": normalized_subfolder}, |
| ) |
|
|
| return { |
| "root_folder": root.as_posix(), |
| "scan_folder": scan_path.as_posix(), |
| "scan_subfolder": normalized_subfolder, |
| } |
|
|
| def _normalize_scan_subfolder(self, scan_subfolder: str) -> str: |
| raw = str(scan_subfolder or "").strip().replace("\\", "/") |
| if not raw: |
| return "" |
| if raw.startswith("/") or raw.startswith("~") or re.match(r"^[A-Za-z]:/", raw): |
| raise ValidationException(message=_("扫描子文件夹必须使用相对路径"), errors={"scan_subfolder": raw}) |
|
|
| parts = [part for part in raw.split("/") if part not in {"", "."}] |
| if not parts: |
| return "" |
| if any(part == ".." for part in parts): |
| raise ValidationException(message=_("扫描子文件夹路径非法"), errors={"scan_subfolder": raw}) |
| return "/".join(parts) |
|
|
| def _is_within_root(self, root: Path, target: Path) -> bool: |
| try: |
| return os.path.commonpath([root.as_posix(), target.as_posix()]) == root.as_posix() |
| except ValueError: |
| return False |
|
|
| def _build_classification_context(self, case: Case) -> dict[str, Any]: |
| our_party_ids: list[int] = [] |
| opponent_party_ids: list[int] = [] |
| our_party_names: list[str] = [] |
| opponent_party_names: list[str] = [] |
|
|
| for party in case.parties.select_related("client").all(): |
| client = getattr(party, "client", None) |
| if not client: |
| continue |
| party_id = int(getattr(party, "id", 0) or 0) |
| party_name = str(getattr(client, "name", "") or "").strip() |
| is_our = bool(getattr(client, "is_our_client", False)) |
| if is_our: |
| if party_id: |
| our_party_ids.append(party_id) |
| if party_name: |
| our_party_names.append(party_name) |
| else: |
| if party_id: |
| opponent_party_ids.append(party_id) |
| if party_name: |
| opponent_party_names.append(party_name) |
|
|
| authority_ids = [int(auth.id) for auth in case.supervising_authorities.all()] |
| primary_authority_id = authority_ids[0] if authority_ids else None |
|
|
| return { |
| "our_party_ids": our_party_ids, |
| "opponent_party_ids": opponent_party_ids, |
| "our_party_names": our_party_names, |
| "opponent_party_names": opponent_party_names, |
| "supervising_authority_ids": authority_ids, |
| "primary_supervising_authority_id": primary_authority_id, |
| } |
|
|
| @staticmethod |
| def _to_int(value: Any) -> int | None: |
| try: |
| parsed = int(value) |
| except (TypeError, ValueError): |
| return None |
| if parsed <= 0: |
| return None |
| return parsed |
|
|
| @staticmethod |
| def _build_materials_url(*, case_id: int, session_id: UUID) -> str: |
| base = reverse("admin:cases_case_materials", args=[case_id]) |
| return f"{base}?{urlencode({'scan_session': str(session_id)})}" |
|
|
|
|
| def run_case_folder_scan_task(session_id: str) -> None: |
| """Django-Q 任务入口。""" |
| CaseFolderScanService().run_scan_task(session_id=session_id) |
|
|
