866 lines
33 KiB
Python
866 lines
33 KiB
Python
"""场外基金附件 OCR 与字段识别适配层。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import base64
|
||
import json
|
||
import re
|
||
import time
|
||
import zipfile
|
||
from collections.abc import Mapping
|
||
from dataclasses import dataclass
|
||
from decimal import Decimal, InvalidOperation
|
||
from importlib import import_module, util
|
||
from io import BytesIO
|
||
from pathlib import Path
|
||
from typing import Any, Literal, cast
|
||
from xml.etree import ElementTree
|
||
|
||
import httpx
|
||
|
||
from app.core.config import Settings
|
||
from app.core.offsite_fund_contracts import DocumentType
|
||
|
||
RecognitionStatus = Literal["mock", "success", "disabled", "misconfigured", "error"]
|
||
|
||
REQUIRED_FIELDS: dict[str, tuple[str, ...]] = {
|
||
"subscription": (
|
||
"基金代码",
|
||
"基金名称",
|
||
"账户标识",
|
||
"申请编号",
|
||
"申请日期",
|
||
"代销机构",
|
||
"申购金额",
|
||
"金额单位",
|
||
),
|
||
"redemption": (
|
||
"基金代码",
|
||
"基金名称",
|
||
"账户标识",
|
||
"申请编号",
|
||
"申请日期",
|
||
"代销机构",
|
||
"赎回份额",
|
||
),
|
||
}
|
||
|
||
EXTRACTED_FIELD_NAMES: tuple[str, ...] = (
|
||
"基金代码",
|
||
"基金名称",
|
||
"账户标识",
|
||
"投资者名称",
|
||
"客户标识",
|
||
"申请编号",
|
||
"申请日期",
|
||
"代销机构",
|
||
"申购金额",
|
||
"金额单位",
|
||
"赎回份额",
|
||
"最新净值",
|
||
"基金最新总份额",
|
||
"申请前持有份额",
|
||
"当前最新可用份额",
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class RecognitionSourceFile:
|
||
filename: str
|
||
media_type: str
|
||
payload: bytes
|
||
file_hash: str
|
||
original_file_path: str = ""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OcrRecognitionResult:
|
||
status: RecognitionStatus
|
||
ocr_text: str
|
||
tables: list[object]
|
||
page_evidence: dict[str, object]
|
||
raw_summary: dict[str, object]
|
||
error_message: str | None = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StructuredRecognitionResult:
|
||
document_type: DocumentType
|
||
extracted_fields: dict[str, object]
|
||
field_confidence: dict[str, Decimal]
|
||
missing_fields: tuple[str, ...]
|
||
low_confidence_fields: tuple[str, ...]
|
||
page_evidence: dict[str, object]
|
||
ocr_text: str
|
||
ocr_status: RecognitionStatus
|
||
llm_status: RecognitionStatus
|
||
error_message: str | None = None
|
||
|
||
|
||
class OffsiteDocumentRecognitionAdapter:
|
||
"""先 OCR，再用 DeepSeek 或本地 Mock 生成统一字段。"""
|
||
|
||
def __init__(
|
||
self,
|
||
settings: Settings,
|
||
*,
|
||
client: httpx.AsyncClient | None = None,
|
||
aliyun_client: object | None = None,
|
||
) -> None:
|
||
self.settings = settings
|
||
self.client = client
|
||
self.aliyun_client = aliyun_client
|
||
self._owns_client = client is None
|
||
|
||
def health_check(self) -> dict[str, object]:
|
||
ocr_missing = self._ocr_missing_config() if self.settings.offsite_ocr_enabled else []
|
||
if (
|
||
self.settings.offsite_ocr_enabled
|
||
and self._uses_official_docmind()
|
||
and self.aliyun_client is None
|
||
and not self._docmind_sdk_available()
|
||
):
|
||
ocr_missing.append("ALIBABACLOUD_DOCMIND_SDK")
|
||
llm_missing = (
|
||
self._deepseek_missing_config() if self.settings.offsite_deepseek_enabled else []
|
||
)
|
||
return {
|
||
"ocr": (
|
||
{"status": "misconfigured", "missing": ocr_missing}
|
||
if ocr_missing
|
||
else {"status": "ok" if self.settings.offsite_ocr_enabled else "disabled"}
|
||
),
|
||
"deepseek": (
|
||
{"status": "misconfigured", "missing": llm_missing}
|
||
if llm_missing
|
||
else {
|
||
"status": "ok" if self.settings.offsite_deepseek_enabled else "disabled",
|
||
"model": self.settings.offsite_deepseek_model,
|
||
}
|
||
),
|
||
}
|
||
|
||
async def recognize(self, source: RecognitionSourceFile) -> StructuredRecognitionResult:
|
||
ocr = await self._recognize_ocr(source)
|
||
ocr = self._with_docx_text_fallback(source, ocr)
|
||
return await self._extract_fields(source, ocr)
|
||
|
||
async def close(self) -> None:
|
||
"""收工时关闭自己创建的 HTTP 客户端，由调用方一次性调用。
|
||
|
||
为什么不能放在单次 recognize() 结尾：同一个 Worker 进程会连续处理多封邮件，
|
||
客户端在第一次识别结束时被关闭后，`_client()` 仍会把这个已关闭的对象交回调用方，
|
||
于是从第二封邮件起每次都抛 "Cannot send a request, as the client has been closed."，
|
||
重试到阈值后把收件游标拖成 blocked，整个收件箱停止收信。
|
||
关闭动作集中在这里，保证"建一次、用一路、收工关一次"。
|
||
"""
|
||
if not self._owns_client or self.client is None:
|
||
return
|
||
client, self.client = self.client, None
|
||
await client.aclose()
|
||
|
||
async def _recognize_ocr(self, source: RecognitionSourceFile) -> OcrRecognitionResult:
|
||
if not self.settings.offsite_ocr_enabled:
|
||
text = _mock_text(source)
|
||
return OcrRecognitionResult(
|
||
status="mock",
|
||
ocr_text=text,
|
||
tables=[],
|
||
page_evidence={},
|
||
raw_summary={"provider": "mock", "filename": source.filename},
|
||
)
|
||
missing = self._ocr_missing_config()
|
||
if missing:
|
||
return OcrRecognitionResult(
|
||
status="misconfigured",
|
||
ocr_text="",
|
||
tables=[],
|
||
page_evidence={},
|
||
raw_summary={},
|
||
error_message=f"OCR配置缺失：{', '.join(missing)}",
|
||
)
|
||
if self._uses_official_docmind():
|
||
return await self._recognize_official_docmind(source)
|
||
client = self._client()
|
||
payload = {
|
||
"filename": source.filename,
|
||
"media_type": source.media_type,
|
||
"file_hash": source.file_hash,
|
||
"content_base64": base64.b64encode(source.payload).decode("ascii"),
|
||
}
|
||
try:
|
||
response = await client.post(
|
||
self.settings.offsite_aliyun_ocr_endpoint,
|
||
headers={
|
||
"X-Acs-AccessKey-Id": self.settings.offsite_aliyun_access_key_id,
|
||
"X-Acs-Signed-Gateway": "offsite-fund",
|
||
},
|
||
json=payload,
|
||
timeout=httpx.Timeout(self.settings.offsite_ocr_timeout_seconds),
|
||
)
|
||
response.raise_for_status()
|
||
body = _json_mapping(response)
|
||
data = _mapping_value(body, "data")
|
||
text = _text_from_mapping(data) or _text_from_mapping(body)
|
||
tables = _list_value(data, "tables") or _list_value(body, "tables")
|
||
page_evidence = _dict_value(data, "page_evidence") or _dict_value(body, "page_evidence")
|
||
return OcrRecognitionResult(
|
||
status="success",
|
||
ocr_text=text,
|
||
tables=tables,
|
||
page_evidence=page_evidence,
|
||
raw_summary={
|
||
"provider": "aliyun_document_ai",
|
||
"response_keys": sorted(str(key) for key in body.keys()),
|
||
},
|
||
)
|
||
except (httpx.HTTPError, ValueError) as exc:
|
||
return OcrRecognitionResult(
|
||
status="error",
|
||
ocr_text="",
|
||
tables=[],
|
||
page_evidence={},
|
||
raw_summary={"provider": "aliyun_document_ai"},
|
||
error_message=type(exc).__name__,
|
||
)
|
||
|
||
async def _recognize_official_docmind(
|
||
self, source: RecognitionSourceFile
|
||
) -> OcrRecognitionResult:
|
||
deadline = time.monotonic() + self.settings.offsite_ocr_timeout_seconds
|
||
file_stream: Any | None = None
|
||
try:
|
||
client = self._aliyun_client()
|
||
sdk_client = cast(Any, client)
|
||
if source.original_file_path:
|
||
file_stream = open(source.original_file_path, "rb")
|
||
else:
|
||
file_stream = BytesIO(source.payload)
|
||
submit_request = self._aliyun_request(
|
||
"SubmitDocParserJobAdvanceRequest",
|
||
file_url_object=file_stream,
|
||
file_name=Path(source.filename).name,
|
||
)
|
||
submit_response = await self._aliyun_call(
|
||
sdk_client.submit_doc_parser_job_advance,
|
||
submit_request,
|
||
deadline,
|
||
self._aliyun_runtime_options(),
|
||
)
|
||
job_id = self._aliyun_value(submit_response, "body", "data", "id")
|
||
if not isinstance(job_id, str) or not job_id:
|
||
raise ValueError("阿里云文档解析未返回任务 ID")
|
||
await self._wait_for_docmind(client, job_id, deadline)
|
||
result_response = await self._aliyun_call(
|
||
sdk_client.get_doc_parser_result,
|
||
self._aliyun_request(
|
||
"GetDocParserResultRequest",
|
||
id=job_id,
|
||
layout_num=0,
|
||
layout_step_size=3000,
|
||
),
|
||
deadline,
|
||
)
|
||
text, tables, evidence = self._docmind_result(result_response)
|
||
return OcrRecognitionResult(
|
||
status="success",
|
||
ocr_text=text,
|
||
tables=tables,
|
||
page_evidence=evidence,
|
||
raw_summary={
|
||
"provider": "aliyun_docmind",
|
||
"job_id": job_id,
|
||
},
|
||
)
|
||
except (OSError, ImportError, TimeoutError, ValueError, TypeError, AttributeError) as exc:
|
||
return OcrRecognitionResult(
|
||
status="error",
|
||
ocr_text="",
|
||
tables=[],
|
||
page_evidence={},
|
||
raw_summary={"provider": "aliyun_docmind"},
|
||
error_message=type(exc).__name__,
|
||
)
|
||
finally:
|
||
if file_stream is not None:
|
||
file_stream.close()
|
||
|
||
async def _wait_for_docmind(
|
||
self, client: object, job_id: str, deadline: float
|
||
) -> None:
|
||
sdk_client = cast(Any, client)
|
||
while True:
|
||
response = await self._aliyun_call(
|
||
sdk_client.query_doc_parser_status,
|
||
self._aliyun_request("QueryDocParserStatusRequest", id=job_id),
|
||
deadline,
|
||
)
|
||
status = str(self._aliyun_value(response, "body", "data", "status") or "")
|
||
if status.lower() == "success":
|
||
return
|
||
if status.lower() == "fail":
|
||
message = self._aliyun_value(response, "body", "message")
|
||
raise ValueError(f"阿里云文档解析失败：{message or '未知错误'}")
|
||
remaining = deadline - time.monotonic()
|
||
if remaining <= 0:
|
||
raise TimeoutError("阿里云文档解析轮询超时")
|
||
await asyncio.sleep(min(self.settings.offsite_aliyun_poll_interval_seconds, remaining))
|
||
|
||
async def _aliyun_call(
|
||
self, method: object, request: object, deadline: float, *extra: object
|
||
) -> object:
|
||
remaining = deadline - time.monotonic()
|
||
if remaining <= 0:
|
||
raise TimeoutError("阿里云文档解析调用超时")
|
||
callable_method = cast(Any, method)
|
||
return await asyncio.wait_for(
|
||
asyncio.to_thread(callable_method, request, *extra),
|
||
timeout=remaining,
|
||
)
|
||
|
||
def _aliyun_runtime_options(self) -> object:
|
||
try:
|
||
tea_util_models = import_module("alibabacloud_tea_util.models")
|
||
except ImportError as exc:
|
||
raise ImportError("缺少阿里云 Tea 运行时依赖，请安装 alibabacloud_tea_util") from exc
|
||
runtime_options = cast(Any, tea_util_models.RuntimeOptions)
|
||
timeout_milliseconds = max(
|
||
1, int(self.settings.offsite_ocr_timeout_seconds * 1000)
|
||
)
|
||
return runtime_options(
|
||
read_timeout=timeout_milliseconds,
|
||
connect_timeout=timeout_milliseconds,
|
||
)
|
||
|
||
def _aliyun_client(self) -> object:
|
||
if self.aliyun_client is not None:
|
||
return self.aliyun_client
|
||
try:
|
||
docmind_module = import_module("alibabacloud_docmind_api20220711.client")
|
||
openapi_models = import_module("alibabacloud_tea_openapi.models")
|
||
except ImportError as exc:
|
||
raise ImportError(
|
||
"缺少阿里云文档智能 SDK，请安装 alibabacloud_docmind_api20220711"
|
||
) from exc
|
||
docmind_client = cast(Any, docmind_module.Client)
|
||
open_api_models = cast(Any, openapi_models)
|
||
endpoint = self.settings.offsite_aliyun_ocr_endpoint.removeprefix("https://")
|
||
endpoint = endpoint.removeprefix("http://").rstrip("/")
|
||
config = open_api_models.Config(
|
||
access_key_id=self.settings.offsite_aliyun_access_key_id,
|
||
access_key_secret=self.settings.offsite_aliyun_access_key_secret,
|
||
)
|
||
config.endpoint = endpoint
|
||
config.region_id = self.settings.offsite_aliyun_region_id
|
||
self.aliyun_client = docmind_client(config)
|
||
return self.aliyun_client
|
||
|
||
def _aliyun_request(self, class_name: str, **values: object) -> object:
|
||
filtered_values = {key: value for key, value in values.items() if value is not None}
|
||
if self.aliyun_client is not None and not self._docmind_sdk_available():
|
||
return _LocalDocmindRequest(class_name=class_name, values=filtered_values)
|
||
try:
|
||
models_module = import_module("alibabacloud_docmind_api20220711.models")
|
||
except ImportError as exc:
|
||
raise ImportError(
|
||
"缺少阿里云文档智能 SDK，请安装 alibabacloud_docmind_api20220711"
|
||
) from exc
|
||
models = cast(Any, models_module)
|
||
request_class = getattr(models, class_name, None)
|
||
if request_class is None:
|
||
raise ValueError(f"阿里云文档智能 SDK 不支持请求模型：{class_name}")
|
||
return request_class(**filtered_values)
|
||
|
||
@staticmethod
|
||
def _docmind_sdk_available() -> bool:
|
||
try:
|
||
return all(
|
||
util.find_spec(module_name) is not None
|
||
for module_name in (
|
||
"alibabacloud_docmind_api20220711",
|
||
"alibabacloud_tea_openapi",
|
||
)
|
||
)
|
||
except (ImportError, ModuleNotFoundError):
|
||
return False
|
||
|
||
@staticmethod
|
||
def _aliyun_value(value: object, *path: str) -> object:
|
||
current = value
|
||
for key in path:
|
||
if current is None:
|
||
return None
|
||
if hasattr(current, "to_map"):
|
||
current = current.to_map()
|
||
if isinstance(current, Mapping):
|
||
current = OffsiteDocumentRecognitionAdapter._mapping_value(current, key)
|
||
else:
|
||
current = getattr(current, key, None)
|
||
return current
|
||
|
||
@staticmethod
|
||
def _mapping_value(value: Mapping[object, object], key: str) -> object:
|
||
camel = re.sub(r"_([a-z])", lambda match: match.group(1).upper(), key)
|
||
for candidate in (key, camel, key[:1].upper() + key[1:], camel[:1].upper() + camel[1:]):
|
||
if candidate in value:
|
||
return value[candidate]
|
||
return None
|
||
|
||
def _docmind_result(
|
||
self, response: object
|
||
) -> tuple[str, list[object], dict[str, object]]:
|
||
raw_layouts = self._aliyun_value(response, "body", "data", "layouts")
|
||
layouts = raw_layouts if isinstance(raw_layouts, list) else []
|
||
text_parts: list[str] = []
|
||
tables: list[object] = []
|
||
evidence: dict[str, object] = {}
|
||
for layout_index, layout in enumerate(layouts, start=1):
|
||
markdown = self._aliyun_value(layout, "markdown_content")
|
||
if isinstance(markdown, str) and markdown:
|
||
text_parts.append(markdown)
|
||
blocks = self._aliyun_value(layout, "blocks")
|
||
if isinstance(blocks, list):
|
||
for block_index, block in enumerate(blocks, start=1):
|
||
block_text = self._aliyun_value(block, "text")
|
||
if isinstance(block_text, str) and block_text:
|
||
text_parts.append(block_text)
|
||
if self._aliyun_value(block, "type") in {"table", "form"}:
|
||
tables.append(self._plain_value(block))
|
||
evidence[f"layout_{layout_index}_block_{block_index}"] = (
|
||
self._plain_value(block)
|
||
)
|
||
return "\n".join(text_parts), tables, evidence
|
||
|
||
@staticmethod
|
||
def _plain_value(value: object) -> object:
|
||
if hasattr(value, "to_map"):
|
||
return OffsiteDocumentRecognitionAdapter._plain_value(value.to_map())
|
||
if isinstance(value, Mapping):
|
||
return {
|
||
str(key): OffsiteDocumentRecognitionAdapter._plain_value(item)
|
||
for key, item in value.items()
|
||
}
|
||
if isinstance(value, list):
|
||
return [OffsiteDocumentRecognitionAdapter._plain_value(item) for item in value]
|
||
return value
|
||
|
||
def _uses_official_docmind(self) -> bool:
|
||
endpoint = self.settings.offsite_aliyun_ocr_endpoint.lower()
|
||
return "docmind-api." in endpoint
|
||
|
||
@staticmethod
|
||
def _with_docx_text_fallback(
|
||
source: RecognitionSourceFile, result: OcrRecognitionResult
|
||
) -> OcrRecognitionResult:
|
||
if not _is_docx_source(source):
|
||
return result
|
||
fallback = _extract_docx_text(source.payload)
|
||
if not fallback or fallback in result.ocr_text:
|
||
return result
|
||
summary = {**result.raw_summary, "docx_text_fallback": True}
|
||
return OcrRecognitionResult(
|
||
status=result.status,
|
||
ocr_text=f"{result.ocr_text}\n{fallback}".strip(),
|
||
tables=result.tables,
|
||
page_evidence=result.page_evidence,
|
||
raw_summary=summary,
|
||
error_message=result.error_message,
|
||
)
|
||
|
||
async def _extract_fields(
|
||
self, source: RecognitionSourceFile, ocr: OcrRecognitionResult
|
||
) -> StructuredRecognitionResult:
|
||
if ocr.status in {"error", "misconfigured"} or (
|
||
ocr.status == "success"
|
||
and not ocr.ocr_text.strip()
|
||
and not ocr.tables
|
||
):
|
||
return _recognition_error_result(ocr)
|
||
if not self.settings.offsite_deepseek_enabled:
|
||
return _mock_structured_result(source, ocr, "mock")
|
||
missing = self._deepseek_missing_config()
|
||
if missing:
|
||
result = _mock_structured_result(source, ocr, "misconfigured")
|
||
return _with_error(result, f"DeepSeek配置缺失：{', '.join(missing)}")
|
||
try:
|
||
body = await self._call_deepseek(source, ocr)
|
||
document_type = _document_type(body.get("document_type"))
|
||
fields = _recognized_fields(body)
|
||
confidence = _confidence_map(_mapping_dict(body, "field_confidence"))
|
||
page_evidence = _mapping_dict(body, "page_evidence") or ocr.page_evidence
|
||
missing_fields = tuple(
|
||
str(item) for item in _mapping_list(body, "missing_fields")
|
||
)
|
||
low_fields = tuple(
|
||
str(item) for item in _mapping_list(body, "low_confidence_fields")
|
||
)
|
||
if document_type in ("subscription", "redemption"):
|
||
missing_fields = _relevant_missing_fields(
|
||
document_type, fields, missing_fields
|
||
)
|
||
if not low_fields:
|
||
low_fields = _low_confidence_fields(confidence)
|
||
return StructuredRecognitionResult(
|
||
document_type=document_type,
|
||
extracted_fields=fields,
|
||
field_confidence=confidence,
|
||
missing_fields=missing_fields,
|
||
low_confidence_fields=low_fields,
|
||
page_evidence=page_evidence,
|
||
ocr_text=ocr.ocr_text,
|
||
ocr_status=ocr.status,
|
||
llm_status="success",
|
||
error_message=ocr.error_message,
|
||
)
|
||
except (httpx.HTTPError, ValueError, KeyError, TypeError) as exc:
|
||
result = _mock_structured_result(source, ocr, "error")
|
||
return _with_error(result, type(exc).__name__)
|
||
|
||
async def _call_deepseek(
|
||
self, source: RecognitionSourceFile, ocr: OcrRecognitionResult
|
||
) -> Mapping[str, object]:
|
||
client = self._client()
|
||
response = await client.post(
|
||
self.settings.offsite_deepseek_base_url.rstrip("/") + "/chat/completions",
|
||
headers={
|
||
"Authorization": f"Bearer {self.settings.offsite_deepseek_api_key}",
|
||
"Content-Type": "application/json",
|
||
},
|
||
json={
|
||
"model": self.settings.offsite_deepseek_model,
|
||
"messages": [
|
||
{
|
||
"role": "system",
|
||
"content": (
|
||
"你只输出一个JSON对象，不要输出Markdown、解释或代码围栏。"
|
||
"JSON必须严格包含document_type、extracted_fields、"
|
||
"field_confidence、missing_fields、low_confidence_fields、"
|
||
"page_evidence六个顶层字段。"
|
||
"extracted_fields必须是对象，字段名只能使用下面列出的中文字段；"
|
||
"无法从原文确认的字段填null，不得猜测或编造。"
|
||
"文档中的产品代码、产品编号、基金产品代码均映射为基金代码；"
|
||
"基金代码不要求固定六位，必须按原文保留。"
|
||
"document_type只能是summary、subscription、redemption、other。"
|
||
"模板："
|
||
'{"document_type":"other","extracted_fields":'
|
||
'{"基金代码":null,"基金名称":null,"账户标识":null,'
|
||
'"投资者名称":null,"客户标识":null,"申请编号":null,'
|
||
'"申请日期":null,"代销机构":null,"申购金额":null,'
|
||
'"金额单位":null,"赎回份额":null,"最新净值":null,'
|
||
'"基金最新总份额":null,"申请前持有份额":null,'
|
||
'"当前最新可用份额":null},'
|
||
'"field_confidence":{},"missing_fields":[],'
|
||
'"low_confidence_fields":[],"page_evidence":{}}'
|
||
),
|
||
},
|
||
{
|
||
"role": "user",
|
||
"content": json.dumps(
|
||
{
|
||
"filename": source.filename,
|
||
"media_type": source.media_type,
|
||
"ocr_text": ocr.ocr_text[:12000],
|
||
"tables": ocr.tables,
|
||
"page_evidence": ocr.page_evidence,
|
||
},
|
||
ensure_ascii=False,
|
||
),
|
||
},
|
||
],
|
||
"temperature": 0,
|
||
"response_format": {"type": "json_object"},
|
||
},
|
||
timeout=httpx.Timeout(self.settings.offsite_deepseek_timeout_seconds),
|
||
)
|
||
response.raise_for_status()
|
||
content = _deepseek_content(_json_mapping(response))
|
||
parsed = json.loads(content)
|
||
if not isinstance(parsed, dict):
|
||
raise ValueError("DeepSeek响应不是JSON对象")
|
||
return cast(Mapping[str, object], parsed)
|
||
|
||
def _client(self) -> httpx.AsyncClient:
|
||
if self.client is None:
|
||
self.client = httpx.AsyncClient()
|
||
return self.client
|
||
|
||
def _ocr_missing_config(self) -> list[str]:
|
||
missing = []
|
||
if not self.settings.offsite_aliyun_ocr_endpoint:
|
||
missing.append("OFFSITE_ALIYUN_OCR_ENDPOINT")
|
||
if not self.settings.offsite_aliyun_access_key_id:
|
||
missing.append("OFFSITE_ALIYUN_ACCESS_KEY_ID")
|
||
if not self.settings.offsite_aliyun_access_key_secret:
|
||
missing.append("OFFSITE_ALIYUN_ACCESS_KEY_SECRET")
|
||
return missing
|
||
|
||
def _deepseek_missing_config(self) -> list[str]:
|
||
missing = []
|
||
if not self.settings.offsite_deepseek_base_url:
|
||
missing.append("OFFSITE_DEEPSEEK_BASE_URL")
|
||
if not self.settings.offsite_deepseek_api_key:
|
||
missing.append("OFFSITE_DEEPSEEK_API_KEY")
|
||
if not self.settings.offsite_deepseek_model:
|
||
missing.append("OFFSITE_DEEPSEEK_MODEL")
|
||
return missing
|
||
|
||
|
||
def _mock_text(source: RecognitionSourceFile) -> str:
|
||
if _is_docx_source(source):
|
||
return f"文件名：{source.filename}"
|
||
text = source.payload.decode("utf-8", errors="ignore").strip()
|
||
return text or f"文件名：{source.filename}"
|
||
|
||
|
||
def _is_docx_source(source: RecognitionSourceFile) -> bool:
|
||
return (
|
||
source.filename.lower().endswith(".docx")
|
||
or source.media_type
|
||
== "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class _LocalDocmindRequest:
|
||
"""供注入式假客户端使用的轻量请求对象。"""
|
||
|
||
class_name: str
|
||
values: dict[str, object]
|
||
|
||
|
||
def _mock_structured_result(
|
||
source: RecognitionSourceFile, ocr: OcrRecognitionResult, llm_status: RecognitionStatus
|
||
) -> StructuredRecognitionResult:
|
||
document_type = _classify(source.filename + "\n" + ocr.ocr_text)
|
||
fields = _extract_simple_fields(ocr.ocr_text)
|
||
confidence = {key: Decimal("1") for key in fields}
|
||
missing_fields: tuple[str, ...] = ()
|
||
if document_type in ("subscription", "redemption"):
|
||
missing_fields = _missing_fields(document_type, fields)
|
||
return StructuredRecognitionResult(
|
||
document_type=document_type,
|
||
extracted_fields=fields,
|
||
field_confidence=confidence,
|
||
missing_fields=missing_fields,
|
||
low_confidence_fields=(),
|
||
page_evidence=ocr.page_evidence,
|
||
ocr_text=ocr.ocr_text,
|
||
ocr_status=ocr.status,
|
||
llm_status=llm_status,
|
||
error_message=ocr.error_message,
|
||
)
|
||
|
||
|
||
def _recognition_error_result(ocr: OcrRecognitionResult) -> StructuredRecognitionResult:
|
||
return StructuredRecognitionResult(
|
||
document_type="other",
|
||
extracted_fields={},
|
||
field_confidence={},
|
||
missing_fields=(),
|
||
low_confidence_fields=(),
|
||
page_evidence=ocr.page_evidence,
|
||
ocr_text=ocr.ocr_text,
|
||
ocr_status="error",
|
||
llm_status="error",
|
||
error_message=ocr.error_message or "OCR未返回可用内容",
|
||
)
|
||
|
||
|
||
def _with_error(result: StructuredRecognitionResult, message: str) -> StructuredRecognitionResult:
|
||
return StructuredRecognitionResult(
|
||
document_type=result.document_type,
|
||
extracted_fields=result.extracted_fields,
|
||
field_confidence=result.field_confidence,
|
||
missing_fields=result.missing_fields,
|
||
low_confidence_fields=result.low_confidence_fields,
|
||
page_evidence=result.page_evidence,
|
||
ocr_text=result.ocr_text,
|
||
ocr_status=result.ocr_status,
|
||
llm_status=result.llm_status,
|
||
error_message=message,
|
||
)
|
||
|
||
|
||
def _classify(text: str) -> DocumentType:
|
||
if "汇总" in text:
|
||
return "summary"
|
||
if "赎回" in text:
|
||
return "redemption"
|
||
if "申购" in text:
|
||
return "subscription"
|
||
return "other"
|
||
|
||
|
||
def _extract_simple_fields(text: str) -> dict[str, object]:
|
||
fields: dict[str, object] = {}
|
||
for name in (
|
||
"基金代码",
|
||
"基金名称",
|
||
"账户标识",
|
||
"投资者名称",
|
||
"客户标识",
|
||
"申请编号",
|
||
"申请日期",
|
||
"代销机构",
|
||
"申购金额",
|
||
"金额单位",
|
||
"赎回份额",
|
||
):
|
||
labels = ("基金代码", "产品代码") if name == "基金代码" else (name,)
|
||
value = next(
|
||
(
|
||
found
|
||
for label in labels
|
||
if (found := _find_label_value(text, label)) is not None
|
||
),
|
||
None,
|
||
)
|
||
if value:
|
||
fields[name] = value
|
||
return fields
|
||
|
||
|
||
def _find_label_value(text: str, label: str) -> str | None:
|
||
pattern = rf"{re.escape(label)}\s*[:：]?\s*([^\s,，;；]+)"
|
||
matched = re.search(pattern, text)
|
||
if not matched:
|
||
return None
|
||
return matched.group(1).strip()
|
||
|
||
|
||
def _extract_docx_text(payload: bytes) -> str:
|
||
try:
|
||
with zipfile.ZipFile(BytesIO(payload)) as archive:
|
||
xml = archive.read("word/document.xml")
|
||
root = ElementTree.fromstring(xml)
|
||
except (OSError, KeyError, ValueError, zipfile.BadZipFile, ElementTree.ParseError):
|
||
return ""
|
||
namespace = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
|
||
paragraphs = []
|
||
for paragraph in root.findall(".//w:p", namespace):
|
||
text = "".join(node.text or "" for node in paragraph.findall(".//w:t", namespace))
|
||
if text.strip():
|
||
paragraphs.append(text.strip())
|
||
return "\n".join(paragraphs)
|
||
|
||
|
||
def _missing_fields(document_type: str, fields: Mapping[str, object]) -> tuple[str, ...]:
|
||
required = list(REQUIRED_FIELDS.get(document_type, ()))
|
||
if not (fields.get("投资者名称") or fields.get("客户标识")):
|
||
required.append("投资者名称/客户标识")
|
||
return tuple(name for name in required if not str(fields.get(name) or "").strip())
|
||
|
||
|
||
def _relevant_missing_fields(
|
||
document_type: str,
|
||
fields: Mapping[str, object],
|
||
reported: tuple[str, ...],
|
||
) -> tuple[str, ...]:
|
||
required = set(REQUIRED_FIELDS.get(document_type, ()))
|
||
required.add("投资者名称/客户标识")
|
||
filtered = tuple(
|
||
name for name in reported
|
||
if name in required and (
|
||
name == "投资者名称/客户标识"
|
||
or not str(fields.get(name) or "").strip()
|
||
)
|
||
)
|
||
return tuple(dict.fromkeys((*filtered, *_missing_fields(document_type, fields))))
|
||
|
||
|
||
def _low_confidence_fields(confidence: Mapping[str, Decimal]) -> tuple[str, ...]:
|
||
return tuple(name for name, value in confidence.items() if value < Decimal("0.95"))
|
||
|
||
|
||
def _confidence_map(value: Mapping[str, object]) -> dict[str, Decimal]:
|
||
result: dict[str, Decimal] = {}
|
||
for key, raw in value.items():
|
||
try:
|
||
result[str(key)] = Decimal(str(raw))
|
||
except (InvalidOperation, ValueError):
|
||
continue
|
||
return result
|
||
|
||
|
||
def _document_type(value: object) -> DocumentType:
|
||
if value in ("summary", "subscription", "redemption", "other"):
|
||
return cast(DocumentType, value)
|
||
return "other"
|
||
|
||
|
||
def _json_mapping(response: httpx.Response) -> Mapping[str, object]:
|
||
body: Any = response.json()
|
||
if not isinstance(body, dict):
|
||
raise ValueError("响应不是JSON对象")
|
||
return cast(Mapping[str, object], body)
|
||
|
||
|
||
def _mapping_value(value: Mapping[str, object], key: str) -> Mapping[str, object]:
|
||
nested = value.get(key)
|
||
if isinstance(nested, dict):
|
||
return cast(Mapping[str, object], nested)
|
||
return {}
|
||
|
||
|
||
def _dict_value(value: Mapping[str, object], key: str) -> dict[str, object]:
|
||
nested = value.get(key)
|
||
if isinstance(nested, dict):
|
||
return dict(cast(Mapping[str, object], nested))
|
||
return {}
|
||
|
||
|
||
def _mapping_dict(value: Mapping[str, object], key: str) -> dict[str, object]:
|
||
nested = value.get(key)
|
||
if isinstance(nested, dict):
|
||
return dict(cast(Mapping[str, object], nested))
|
||
return {}
|
||
|
||
|
||
def _mapping_list(value: Mapping[str, object], key: str) -> list[object]:
|
||
nested = value.get(key)
|
||
return list(nested) if isinstance(nested, list) else []
|
||
|
||
|
||
def _recognized_fields(body: Mapping[str, object]) -> dict[str, object]:
|
||
nested = _mapping_dict(body, "extracted_fields")
|
||
fields = {
|
||
name: value
|
||
for name, value in nested.items()
|
||
if name in EXTRACTED_FIELD_NAMES and value is not None
|
||
}
|
||
for name in EXTRACTED_FIELD_NAMES:
|
||
value = body.get(name)
|
||
if name not in fields and value is not None:
|
||
fields[name] = value
|
||
return fields
|
||
|
||
|
||
def _list_value(value: Mapping[str, object], key: str) -> list[object]:
|
||
nested = value.get(key)
|
||
return list(nested) if isinstance(nested, list) else []
|
||
|
||
|
||
def _text_from_mapping(value: Mapping[str, object]) -> str:
|
||
for key in ("ocr_text", "text", "content", "markdown"):
|
||
candidate = value.get(key)
|
||
if isinstance(candidate, str):
|
||
return candidate
|
||
return ""
|
||
|
||
|
||
def _deepseek_content(body: Mapping[str, object]) -> str:
|
||
choices = body.get("choices")
|
||
if not isinstance(choices, list) or not choices:
|
||
raise ValueError("DeepSeek响应缺少choices")
|
||
first = choices[0]
|
||
if not isinstance(first, dict):
|
||
raise ValueError("DeepSeek响应格式错误")
|
||
message = first.get("message")
|
||
if not isinstance(message, dict):
|
||
raise ValueError("DeepSeek响应缺少message")
|
||
content = message.get("content")
|
||
if not isinstance(content, str) or not content.strip():
|
||
raise ValueError("DeepSeek响应缺少content")
|
||
return content
|