441 lines
16 KiB
Python
441 lines
16 KiB
Python
|
"""场外基金附件 OCR 与字段识别适配层。"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import base64
|
|||
|
|
import json
|
|||
|
|
import re
|
|||
|
|
from collections.abc import Mapping
|
|||
|
|
from dataclasses import dataclass
|
|||
|
|
from decimal import Decimal, InvalidOperation
|
|||
|
|
from typing import Any, Literal, cast
|
|||
|
|
|
|||
|
|
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": (
|
|||
|
|
"基金代码",
|
|||
|
|
"基金名称",
|
|||
|
|
"账户标识",
|
|||
|
|
"申请编号",
|
|||
|
|
"申请日期",
|
|||
|
|
"代销机构",
|
|||
|
|
"赎回份额",
|
|||
|
|
),
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
@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) -> None:
|
|||
|
|
self.settings = settings
|
|||
|
|
self.client = 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 []
|
|||
|
|
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)
|
|||
|
|
structured = await self._extract_fields(source, ocr)
|
|||
|
|
if self._owns_client and self.client is not None:
|
|||
|
|
await self.client.aclose()
|
|||
|
|
return structured
|
|||
|
|
|
|||
|
|
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)}",
|
|||
|
|
)
|
|||
|
|
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 _extract_fields(
|
|||
|
|
self, source: RecognitionSourceFile, ocr: OcrRecognitionResult
|
|||
|
|
) -> StructuredRecognitionResult:
|
|||
|
|
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 = _dict_value(body, "extracted_fields")
|
|||
|
|
confidence = _confidence_map(_dict_value(body, "field_confidence"))
|
|||
|
|
page_evidence = _dict_value(body, "page_evidence") or ocr.page_evidence
|
|||
|
|
missing_fields = tuple(str(item) for item in _list_value(body, "missing_fields"))
|
|||
|
|
low_fields = tuple(str(item) for item in _list_value(body, "low_confidence_fields"))
|
|||
|
|
if document_type in ("subscription", "redemption") and not missing_fields:
|
|||
|
|
missing_fields = _missing_fields(document_type, 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。字段名必须使用申请日期和申请编号，"
|
|||
|
|
"document_type只能是summary、subscription、redemption、other。"
|
|||
|
|
),
|
|||
|
|
},
|
|||
|
|
{
|
|||
|
|
"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:
|
|||
|
|
text = source.payload.decode("utf-8", errors="ignore").strip()
|
|||
|
|
return text or f"文件名：{source.filename}"
|
|||
|
|
|
|||
|
|
|
|||
|
|
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 _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 (
|
|||
|
|
"基金代码",
|
|||
|
|
"基金名称",
|
|||
|
|
"账户标识",
|
|||
|
|
"投资者名称",
|
|||
|
|
"客户标识",
|
|||
|
|
"申请编号",
|
|||
|
|
"申请日期",
|
|||
|
|
"代销机构",
|
|||
|
|
"申购金额",
|
|||
|
|
"金额单位",
|
|||
|
|
"赎回份额",
|
|||
|
|
):
|
|||
|
|
value = _find_label_value(text, name)
|
|||
|
|
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 _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 _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 _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
|