""" 关键信息提取器 调用 Ollama 大模型从法院文书中提取关键信息（案号、开庭时间等）。 支持正则表达式提取和 Ollama 交叉校验机制。 Requirements: 4.3, 4.4, 4.7 """ import logging from datetime import datetime from typing import Any from django.utils.translation import gettext_lazy as _ from apps.core.exceptions import RecognitionTimeoutError, ServiceUnavailableError from apps.core.interfaces import ServiceLocator from apps.core.llm.config import LLMConfig from apps.core.llm.exceptions import LLMNetworkError, LLMTimeoutError from ._case_number_mixin import CaseNumberMixin from ._datetime_extraction_mixin import DatetimeExtractionMixin from ._response_parser_mixin import ResponseParserMixin logger = logging.getLogger("apps.document_recognition") def get_ollama_model() -> str: """兼容旧测试与调用方：保留模块级配置读取入口。""" return LLMConfig.get_ollama_model() def get_ollama_base_url() -> str: """兼容旧测试与调用方：保留模块级配置读取入口。""" return LLMConfig.get_ollama_base_url() def chat( *, messages: list[dict[str, str]], model: str | None = None, llm_service: Any | None = None, **kwargs: Any, ) -> dict[str, Any]: """ 兼容旧测试与调用方：保留模块级 chat 入口，内部转发到统一 LLM 服务。 """ service = llm_service or ServiceLocator.get_llm_service() llm_response = service.chat( messages=messages, backend="ollama", model=model, fallback=False, **kwargs, ) return {"message": {"content": llm_response.content}} class InfoExtractor(CaseNumberMixin, DatetimeExtractionMixin, ResponseParserMixin): """ 关键信息提取器 使用 Ollama 大模型从法院文书中提取关键信息： - 传票：案号、开庭时间 - 执行裁定书：案号、财产保全到期时间（预留扩展） Requirements: 4.3, 4.4 """ SUMMONS_PROMPT = """请从以下传票内容中提取案号和开庭时间。 传票内容： {text} 提取要求： 1. 案号格式：（年份）法院代码+案件类型字号+序号+号 - 年份：4位数字，如2024、2025 - 法院代码：省份简称+区县代码，如"粤0604"、"京0105"、"沪0115" - 案件类型字号（必须包含）：民初、民终、刑初、刑终、执、执保、执异、执恢、破、行初、行终等 - 序号：数字 - 必须以"号"字结尾 - 示例：（2024）粤0604民初41257号、（2025）京0105刑初12345号 2. 重要：案号中必须包含案件类型字号（如民初、民终、刑初等），不能省略！ - 错误示例：（2025）粤060441257（缺少案件类型） - 正确示例：（2025）粤0604民初41257号 3. 开庭时间需要包含完整的日期和时间，格式为：YYYY-MM-DD HH:MM 4. 如果无法确定某个字段，请返回 null 请严格按照以下 JSON 格式返回结果，不要包含其他内容： {{"case_number": "案号或null", "court_time": "YYYY-MM-DD HH:MM或null"}} """ EXECUTION_PROMPT = """请从以下执行裁定书中提取案号和财产保全到期时间。 裁定书内容： {text} 提取要求： 1. 案号格式：（年份）法院代码+案件类型字号+序号+号 - 年份：4位数字，如2024、2025 - 法院代码：省份简称+区县代码，如"粤0604"、"京0105" - 案件类型字号（必须包含）：执、执保、执异、执恢、民初、民终等 - 序号：数字 - 必须以"号"字结尾 - 示例：（2024）粤0604执保12345号、（2025）京0105执12345号 2. 重要：案号中必须包含案件类型字号，不能省略！ 3. 财产保全到期时间格式为：YYYY-MM-DD 4. 如果无法确定某个字段，请返回 null 请严格按照以下 JSON 格式返回结果，不要包含其他内容： {{"case_number": "案号或null", "preservation_deadline": "YYYY-MM-DD或null"}} """ def __init__( self, ollama_model: str | None = None, ollama_base_url: str | None = None, llm_service: Any | None = None, ): self.ollama_model = ollama_model or get_ollama_model() self.ollama_base_url = ollama_base_url or get_ollama_base_url() self._llm_service = llm_service @property def llm_service(self) -> Any: if self._llm_service is None: self._llm_service = ServiceLocator.get_llm_service() return self._llm_service def extract_summons_info(self, text: str) -> dict[str, Any]: """ 提取传票信息 Requirements: 4.3 """ if not text or not text.strip(): logger.warning( "传票内容为空，无法提取信息", extra={"action": "extract_summons_info", "result": "empty_text"}, ) return {"case_number": None, "court_time": None, "extraction_method": None} truncated_text = text[:4000] if len(text) > 4000 else text logger.info( "开始提取传票信息", extra={"action": "extract_summons_info", "text_length": len(text), "truncated_length": len(truncated_text)}, ) regex_case_number = self._extract_case_number_by_regex(truncated_text) if regex_case_number: logger.info(f"正则成功提取案号: {regex_case_number}") regex_datetimes = self._extract_datetime_by_regex(truncated_text) logger.info(f"正则提取到 {len(regex_datetimes)} 个时间候选") for dt, matched_text, score in regex_datetimes: logger.info(f" - {dt} (原文: {matched_text}, 得分: {score})") ollama_result: dict[str, Any] = {"case_number": None, "court_time": None} ollama_datetime: datetime | None = None try: prompt = self.SUMMONS_PROMPT.format(text=truncated_text) response = chat( messages=[{"role": "user", "content": prompt}], model=self.ollama_model, llm_service=self.llm_service, ) ollama_result = self._parse_summons_response(response) ollama_datetime = ollama_result.get("court_time") if ollama_datetime: logger.info(f"Ollama 提取到时间: {ollama_datetime}") if regex_case_number is None and ollama_result.get("case_number"): logger.info(f"Ollama 提取到案号: {ollama_result.get('case_number')}") except (LLMNetworkError, ConnectionError) as e: logger.warning(f"Ollama 服务不可用，将仅使用正则结果: {e}") except LLMTimeoutError as e: logger.warning(f"Ollama 提取超时，将仅使用正则结果: {e}") except Exception as e: logger.warning(f"Ollama 提取失败，将仅使用正则结果: {e}") best_datetime, extraction_method = self._select_best_datetime(regex_datetimes, ollama_datetime) logger.info(f"最终选择时间: {best_datetime}, 方法: {extraction_method}") final_case_number = regex_case_number if regex_case_number else ollama_result.get("case_number") case_number_source = "regex" if regex_case_number else ("ollama" if ollama_result.get("case_number") else None) result = { "case_number": final_case_number, "court_time": best_datetime, "extraction_method": extraction_method, } logger.info( "传票信息提取完成", extra={ "action": "extract_summons_info", "case_number": result.get("case_number"), "case_number_source": case_number_source, "court_time": str(result.get("court_time")) if result.get("court_time") else None, "extraction_method": extraction_method, }, ) return result def extract_execution_info(self, text: str) -> dict[str, Any]: """ 提取执行裁定书信息（预留扩展） Requirements: 4.4 """ if not text or not text.strip(): logger.warning( "执行裁定书内容为空，无法提取信息", extra={"action": "extract_execution_info", "result": "empty_text"}, ) return {"case_number": None, "preservation_deadline": None} truncated_text = text[:4000] if len(text) > 4000 else text logger.info( "开始提取执行裁定书信息", extra={ "action": "extract_execution_info", "text_length": len(text), "truncated_length": len(truncated_text), }, ) try: prompt = self.EXECUTION_PROMPT.format(text=truncated_text) response = chat( messages=[{"role": "user", "content": prompt}], model=self.ollama_model, llm_service=self.llm_service, ) result = self._parse_execution_response(response) logger.info( "执行裁定书信息提取完成", extra={ "action": "extract_execution_info", "case_number": result.get("case_number"), "preservation_deadline": ( str(result.get("preservation_deadline")) if result.get("preservation_deadline") else None ), }, ) return result except (LLMNetworkError, ConnectionError) as e: logger.error( f"Ollama 服务不可用: {e}", extra={"action": "extract_execution_info", "error_type": "connection_error", "error": str(e)}, ) raise ServiceUnavailableError( message=_("AI 服务暂时不可用，请稍后重试"), code="OLLAMA_SERVICE_UNAVAILABLE", errors={"service": "Ollama 服务连接失败"}, service_name="Ollama", ) from e except LLMTimeoutError as e: logger.error( f"执行裁定书信息提取超时: {e}", extra={"action": "extract_execution_info", "error_type": "timeout_error", "error": str(e)}, ) raise RecognitionTimeoutError( message=_("信息提取超时，请重试"), code="EXTRACTION_TIMEOUT", errors={"timeout": "AI 提取超时"}, ) from e except Exception as e: logger.error( f"执行裁定书信息提取失败: {e!s}", extra={"action": "extract_execution_info", "error_type": type(e).__name__, "error": str(e)}, exc_info=True, ) raise RuntimeError(f"执行裁定书信息提取失败: {e!s}") from e