File size: 7,168 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 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | """Module for recognize document."""
from __future__ import annotations
import logging
from dataclasses import dataclass
from datetime import date
from typing import Any
from apps.core.exceptions import RecognitionTimeoutError, ServiceUnavailableError, ValidationException
from apps.document_recognition.services.data_classes import (
BindingResult,
DocumentType,
RecognitionResponse,
RecognitionResult,
)
logger = logging.getLogger("apps.document_recognition")
@dataclass(frozen=True)
class RecognizeCourtDocumentUsecase:
text_extraction: Any
classifier: Any
extractor: Any
binding_service: Any
document_renamer: Any
def execute(self, *, file_path: str, user: Any | None = None) -> RecognitionResponse:
logger.info("开始识别文书", extra={"action": "recognize_document", "file_path": file_path})
try:
extraction_result = self.text_extraction.extract_text(file_path)
if not extraction_result.success or not extraction_result.text.strip():
return self._empty_response(file_path, extraction_result)
doc_type, confidence = self.classifier.classify(extraction_result.text)
case_number, key_time = self._extract_key_info(doc_type, extraction_result.text)
recognition = RecognitionResult(
document_type=doc_type,
case_number=case_number,
key_time=key_time,
raw_text=extraction_result.text,
confidence=confidence,
extraction_method=extraction_result.extraction_method,
)
binding, renamed_file_path = self._bind_document(
doc_type, case_number, key_time, extraction_result.text, file_path, user
)
logger.info(
"文书识别完成",
extra={"action": "recognize_document", "document_type": doc_type.value, "case_number": case_number},
)
return RecognitionResponse(recognition=recognition, binding=binding, file_path=renamed_file_path)
except (ValidationException, ServiceUnavailableError, RecognitionTimeoutError):
raise
except Exception:
logger.error("文书识别失败", extra={"action": "recognize_document", "file_path": file_path}, exc_info=True)
raise
def _empty_response(self, file_path: Any, extraction_result: Any) -> RecognitionResponse:
"""文本提取失败时返回空结果"""
return RecognitionResponse(
recognition=RecognitionResult(
document_type=DocumentType.OTHER,
case_number=None,
key_time=None,
raw_text="",
confidence=0.0,
extraction_method=extraction_result.extraction_method,
),
binding=BindingResult.failure_result(message="无法从文书中提取文字", error_code="TEXT_EXTRACTION_FAILED"),
file_path=file_path,
)
def _extract_key_info(self, doc_type: Any, text: Any) -> tuple[Any, Any]:
"""根据文书类型提取案号和关键时间"""
if doc_type == DocumentType.SUMMONS:
info = self.extractor.extract_summons_info(text)
return (info.get("case_number"), info.get("court_time"))
if doc_type == DocumentType.EXECUTION_RULING:
info = self.extractor.extract_execution_info(text)
return (info.get("case_number"), info.get("preservation_deadline"))
return (None, None)
def _bind_document(
self, doc_type: Any, case_number: Any, key_time: Any, raw_text: Any, file_path: Any, user: Any
) -> tuple[Any, Any]:
"""绑定文书到案件,返回 (binding, renamed_file_path)"""
_UNSUPPORTED = {
DocumentType.OTHER: ("暂时只支持传票识别,其他文书类型敬请期待", "UNSUPPORTED_DOCUMENT_TYPE"),
DocumentType.EXECUTION_RULING: ("执行裁定书绑定功能开发中,敬请期待", "FEATURE_NOT_IMPLEMENTED"),
}
if doc_type != DocumentType.SUMMONS or not case_number:
if not case_number and doc_type == DocumentType.SUMMONS:
return (
BindingResult.failure_result(
message="未识别到案号,无法绑定案件", error_code="CASE_NUMBER_NOT_FOUND"
),
file_path,
)
msg, code = _UNSUPPORTED.get(doc_type, ("不支持的文书类型", "UNSUPPORTED"))
return (BindingResult.failure_result(message=msg, error_code=code), file_path)
case_id = self.binding_service.find_case_by_number(case_number)
case_name = None
renamed_file_path = file_path
if case_id:
case_dto = self.binding_service.case_service.get_case_by_id_internal(case_id)
if case_dto:
case_name = case_dto.name
if case_name:
renamed_file_path = self._rename_document(file_path=file_path, document_type=doc_type, case_name=case_name)
log_content = self.binding_service.format_log_content(
document_type=doc_type, case_number=case_number, key_time=key_time, raw_text=raw_text
)
binding = self.binding_service.bind_document_to_case(
case_number=case_number,
document_type=doc_type,
content=log_content,
key_time=key_time,
file_path=renamed_file_path,
user=user,
)
return (binding, renamed_file_path)
def _rename_document(self, *, file_path: str, document_type: DocumentType, case_name: str) -> str:
try:
title_map = {
DocumentType.SUMMONS: "传票",
DocumentType.EXECUTION_RULING: "执行裁定书",
DocumentType.OTHER: "司法文书",
}
title = title_map.get(document_type, "司法文书")
new_filename = self.document_renamer.generate_filename(
title=title, case_name=case_name, received_date=date.today()
)
from apps.core.utils.path import Path
original_path = Path(file_path)
new_path = original_path.parent / new_filename
counter = 1
while new_path.exists():
base_filename = new_filename.replace("收.pdf", f"收{counter}.pdf")
new_path = original_path.parent / base_filename
counter += 1
if counter > 100:
break
original_path.rename(new_path)
logger.info(
"文书重命名成功",
extra={
"action": "rename_document",
"original_path": file_path,
"new_path": str(new_path),
"document_type": document_type.value,
"case_name": case_name,
},
)
return str(new_path)
except Exception:
logger.warning("文书重命名失败,保留原文件名", extra={})
return file_path
|