fix: 物流表格识别优先用阿里云OCR,SDK不可用时降级LLM视觉

拆分为三种方法:_parse_table_via_ocr(OCR+LLM)、
_parse_table_via_llm_vision(LLM视觉直读),
自动检测SDK可用性选择路径。

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
wsb1224 2026-07-04 10:16:58 +08:00
parent 94e1b66ec8
commit b6a99ba567

View File

@ -2400,15 +2400,61 @@ class OrderService:
def parse_logistics_table_images(self, image_urls, session): def parse_logistics_table_images(self, image_urls, session):
"""识别物流表格截图,用 LLM 视觉能力直接从图片中提取快递单号和收件人。""" """识别物流表格截图。优先用阿里云OCR+LLMSDK不可用时降级为LLM视觉直接识别。"""
import logging as _log
_logger = _log.getLogger(__name__)
# 检测阿里云 OCR SDK 是否可用
try:
from alibabacloud_ocr_api20210707.models import RecognizeGeneralRequest # noqa: F401
ocr_available = True
except ImportError:
ocr_available = False
_logger.info("[parse_logistics_table_images] 阿里云OCR SDK未安装将使用LLM视觉识别")
if ocr_available:
return self._parse_table_via_ocr(image_urls, session, _logger)
return self._parse_table_via_llm_vision(image_urls, _logger)
def _parse_table_via_ocr(self, image_urls, session, _logger):
"""阿里云OCR + LLM提取。"""
from backend.app.services.ai_service import ai_service
from backend.app.core.config import get_settings
settings = get_settings()
results = []
seen_tracking = set()
for url in image_urls:
try:
_logger.info("[parse_logistics_table_images] OCR处理图片: %s", url)
ctx = ai_service._ocr_and_parse_image(url)
raw_text = ctx.get("raw_text", "")
_logger.info("[parse_logistics_table_images] OCR文本长度: %d", len(raw_text))
if not raw_text.strip():
continue
llm_items = self._llm_extract_logistics(raw_text, settings)
for item in llm_items:
tn = (item.get("tracking_number") or "").strip()
if not tn or len(tn) < 10 or tn in seen_tracking:
continue
seen_tracking.add(tn)
results.append({
"tracking_number": tn,
"express_company": item.get("express_company") or None,
"recipient_name": item.get("recipient_name") or None,
})
except Exception as exc:
_logger.warning("[parse_logistics_table_images] OCR识别失败: %s", exc, exc_info=True)
_logger.info("[parse_logistics_table_images] OCR模式最终结果: %d", len(results))
return results
def _parse_table_via_llm_vision(self, image_urls, _logger):
"""LLM视觉直接识别表格图片。"""
import base64 as _b64 import base64 as _b64
import json as _json import json as _json
import logging as _log
from urllib import request as urllib_request from urllib import request as urllib_request
from backend.app.services.ai_service import read_image_bytes from backend.app.services.ai_service import read_image_bytes
from backend.app.core.config import get_settings from backend.app.core.config import get_settings
_logger = _log.getLogger(__name__)
settings = get_settings() settings = get_settings()
api_key = settings.llm_parse_api_key or settings.aliyun_ai_access_key_id api_key = settings.llm_parse_api_key or settings.aliyun_ai_access_key_id
@ -2425,7 +2471,7 @@ class OrderService:
seen_tracking = set() seen_tracking = set()
for url in image_urls: for url in image_urls:
try: try:
_logger.info("[parse_logistics_table_images] 处理图片: %s", url) _logger.info("[parse_logistics_table_images] LLM视觉处理图片: %s", url)
image_bytes = read_image_bytes(url) image_bytes = read_image_bytes(url)
image_b64 = _b64.b64encode(image_bytes).decode("utf-8") image_b64 = _b64.b64encode(image_bytes).decode("utf-8")
suffix = url.rsplit(".", 1)[-1].lower() if "." in url else "jpeg" suffix = url.rsplit(".", 1)[-1].lower() if "." in url else "jpeg"
@ -2455,7 +2501,6 @@ class OrderService:
content = result["choices"][0]["message"]["content"] content = result["choices"][0]["message"]["content"]
_logger.info("[parse_logistics_table_images] LLM响应: %s", content[:500]) _logger.info("[parse_logistics_table_images] LLM响应: %s", content[:500])
llm_items = self._extract_json_array(content) llm_items = self._extract_json_array(content)
_logger.info("[parse_logistics_table_images] 提取结果: %d", len(llm_items))
for item in llm_items: for item in llm_items:
tn = (item.get("tracking_number") or "").strip() tn = (item.get("tracking_number") or "").strip()
@ -2468,8 +2513,8 @@ class OrderService:
"recipient_name": item.get("recipient_name") or None, "recipient_name": item.get("recipient_name") or None,
}) })
except Exception as exc: except Exception as exc:
_logger.warning("[parse_logistics_table_images] 识别失败: %s", exc, exc_info=True) _logger.warning("[parse_logistics_table_images] LLM视觉识别失败: %s", exc, exc_info=True)
_logger.info("[parse_logistics_table_images] 最终结果: %d", len(results)) _logger.info("[parse_logistics_table_images] LLM视觉模式最终结果: %d", len(results))
return results return results
def _llm_extract_logistics(self, text, settings): def _llm_extract_logistics(self, text, settings):