File size: 4,612 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 | """
文书识别任务管理服务
负责 DocumentRecognitionTask 的 CRUD 操作和案件搜索查询,
将 ORM 操作从 API 层下沉到 Service 层.
"""
import logging
from datetime import datetime
from typing import Any
from django.utils.translation import gettext_lazy as _
from apps.core.exceptions import NotFoundError, ValidationException
logger = logging.getLogger("apps.document_recognition")
class DocumentRecognitionTaskService:
"""文书识别任务管理服务"""
def create_task(self, *, file_path: str, original_filename: str) -> Any:
"""创建识别任务记录
Args:
file_path: 文件路径
original_filename: 原始文件名
Returns:
DocumentRecognitionTask 实例
"""
from apps.document_recognition.models import DocumentRecognitionStatus, DocumentRecognitionTask
task = DocumentRecognitionTask.objects.create(
file_path=file_path,
original_filename=original_filename,
status=DocumentRecognitionStatus.PENDING,
)
logger.info("创建文书识别任务", extra={})
return task
def get_task(self, task_id: int, *, select_case: bool = False) -> Any:
"""获取识别任务
Args:
task_id: 任务 ID
select_case: 是否预加载关联案件
Returns:
DocumentRecognitionTask 实例
Raises:
NotFoundError: 任务不存在
"""
from apps.document_recognition.models import DocumentRecognitionTask
qs = DocumentRecognitionTask.objects.all()
if select_case:
qs = qs.select_related("case")
try:
return qs.get(id=task_id)
except DocumentRecognitionTask.DoesNotExist:
raise NotFoundError(message=_("任务不存在"), code="TASK_NOT_FOUND", errors={}) from None
def update_task_info(
self,
task_id: int,
*,
case_number: str | None = None,
key_time: str | None = None,
) -> Any:
"""更新识别任务信息(案号、关键时间)
Args:
task_id: 任务 ID
case_number: 案号(None 表示不更新)
key_time: 关键时间 ISO 格式字符串(None 表示不更新)
Returns:
更新后的 DocumentRecognitionTask 实例
Raises:
NotFoundError: 任务不存在
ValidationException: 时间格式不正确
"""
task = self.get_task(task_id)
updated_fields: list[str] = []
if case_number is not None:
task.case_number = case_number if case_number else None
updated_fields.append("case_number")
if key_time is not None:
if key_time:
try:
task.key_time = datetime.fromisoformat(key_time.replace("Z", "+00:00"))
except ValueError:
raise ValidationException(
message=_("时间格式不正确"),
code="INVALID_TIME_FORMAT",
errors={},
) from None
else:
task.key_time = None
updated_fields.append("key_time")
if updated_fields:
task.save(update_fields=updated_fields)
logger.info(
"识别信息已更新",
extra={
"action": "update_task_info",
"task_id": task_id,
"updated_fields": updated_fields,
"case_number": task.case_number,
"key_time": str(task.key_time) if task.key_time else None,
},
)
return task
def search_cases_for_binding(self, *, search_term: str = "", limit: int = 20) -> list[dict[str, Any]]:
"""搜索可绑定的案件
支持按案件名称、案号、当事人搜索.
Args:
search_term: 搜索关键词
limit: 返回数量限制
Returns:
案件信息字典列表
"""
from apps.core.interfaces import ServiceLocator
case_service = ServiceLocator.get_case_service()
results = case_service.search_cases_for_binding_internal(search_term=search_term, limit=limit)
logger.info(
"案件搜索完成",
extra={
"action": "search_cases_for_binding",
"query": search_term,
"result_count": len(results),
},
)
return results
|