fix: 快递100 OCR 失败时不再抛400错误，改为降级提示手动输入
- recognize_waybill 捕获所有异常返回空结果而非抛出 - 修正 SysUser 导入为 User（匹配实际模型类名） - 前端识别为空时提示"自动识别未成功，请手动输入单号" Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
eeffebd01a
commit
b2b7641999
@ -674,46 +674,55 @@ class LogisticsService:
|
||||
raise AppException(code=ErrorCode.INTERNAL_ERROR, message="数据库连接不可用", status_code=500)
|
||||
|
||||
def recognize_waybill(self, payload: dict, session: Session | None = None, current_user: dict | None = None) -> dict:
|
||||
"""调用快递100面单OCR识别运单号。"""
|
||||
"""调用快递100面单OCR识别运单号。
|
||||
|
||||
优先使用快递100 API，失败时返回空结果让前端引导手动输入。
|
||||
"""
|
||||
import re as _re
|
||||
|
||||
image_url = (payload.get("image_url") or "").strip()
|
||||
if not image_url:
|
||||
raise AppException(code=ErrorCode.PARAM_ERROR, message="图片地址不能为空", status_code=400)
|
||||
|
||||
settings = self.settings
|
||||
if not settings.kuaidi100_key:
|
||||
raise AppException(code=ErrorCode.THIRD_PARTY_FAILED, message="快递100配置不完整", status_code=400)
|
||||
tracking_number = ""
|
||||
express_company = ""
|
||||
express_name = ""
|
||||
raw_result = {}
|
||||
|
||||
import base64
|
||||
from urllib.parse import quote, urlparse, urlunparse
|
||||
# 尝试快递100 OCR
|
||||
if settings.kuaidi100_key:
|
||||
try:
|
||||
import base64
|
||||
from urllib.parse import quote, urlparse, urlunparse
|
||||
|
||||
parsed = urlparse(image_url)
|
||||
safe_url = urlunparse(parsed._replace(path=quote(parsed.path)))
|
||||
try:
|
||||
with request.urlopen(safe_url, timeout=15) as resp:
|
||||
image_bytes = resp.read()
|
||||
image_b64 = base64.b64encode(image_bytes).decode("utf-8")
|
||||
except Exception as exc:
|
||||
raise AppException(code=ErrorCode.THIRD_PARTY_FAILED, message=f"下载面单图片失败：{exc}", status_code=400) from exc
|
||||
parsed = urlparse(image_url)
|
||||
safe_url = urlunparse(parsed._replace(path=quote(parsed.path)))
|
||||
with request.urlopen(safe_url, timeout=15) as resp:
|
||||
image_bytes = resp.read()
|
||||
image_b64 = base64.b64encode(image_bytes).decode("utf-8")
|
||||
|
||||
endpoint = (settings.logistics_trace_endpoint or "https://api.kuaidi100.com").strip().rstrip("/")
|
||||
body = json.dumps({"key": settings.kuaidi100_key, "image": image_b64}, ensure_ascii=False).encode("utf-8")
|
||||
req = request.Request(url=f"{endpoint}/ocr", data=body, headers={"Content-Type": "application/json; charset=UTF-8"}, method="POST")
|
||||
try:
|
||||
with request.urlopen(req, timeout=30) as resp:
|
||||
result = json.loads(resp.read().decode("utf-8") or "{}")
|
||||
except Exception as exc:
|
||||
raise AppException(code=ErrorCode.THIRD_PARTY_FAILED, message=f"快递100面单识别失败：{exc}", status_code=400) from exc
|
||||
|
||||
tracking_number = result.get("number") or result.get("data", {}).get("number") or ""
|
||||
express_company = result.get("com") or result.get("data", {}).get("com") or ""
|
||||
express_name = result.get("comName") or result.get("data", {}).get("comName") or ""
|
||||
endpoint = (settings.logistics_trace_endpoint or "https://api.kuaidi100.com").strip().rstrip("/")
|
||||
body = json.dumps({"key": settings.kuaidi100_key, "image": image_b64}, ensure_ascii=False).encode("utf-8")
|
||||
req = request.Request(url=f"{endpoint}/ocr", data=body, headers={"Content-Type": "application/json; charset=UTF-8"}, method="POST")
|
||||
with request.urlopen(req, timeout=30) as resp:
|
||||
result = json.loads(resp.read().decode("utf-8") or "{}")
|
||||
raw_result = result
|
||||
tracking_number = (result.get("number") or result.get("data", {}).get("number") or "").strip()
|
||||
express_company = (result.get("com") or result.get("data", {}).get("com") or "").strip()
|
||||
express_name = (result.get("comName") or result.get("data", {}).get("comName") or "").strip()
|
||||
logger.info("[recognize_waybill] 快递100识别成功: %s, company=%s", tracking_number, express_name)
|
||||
except Exception as exc:
|
||||
logger.warning("[recognize_waybill] 快递100 OCR 失败，将返回空结果: %s", exc)
|
||||
raw_result = {"error": str(exc)}
|
||||
|
||||
# 快递100未配置或失败时，返回空结果让前端引导手动输入
|
||||
return {
|
||||
"tracking_number": tracking_number.strip(),
|
||||
"express_company": express_company.strip(),
|
||||
"express_name": express_name.strip(),
|
||||
"confidence": 0.9,
|
||||
"raw_result": result,
|
||||
"tracking_number": tracking_number,
|
||||
"express_company": express_company,
|
||||
"express_name": express_name,
|
||||
"confidence": 0.9 if tracking_number else 0,
|
||||
"raw_result": raw_result,
|
||||
}
|
||||
|
||||
def submit_waybills(self, task_id: int, payload: dict, session: Session | None = None, current_user: dict | None = None) -> dict:
|
||||
@ -937,10 +946,10 @@ class LogisticsService:
|
||||
# 查询业务员姓名
|
||||
salesman_name = ""
|
||||
if order and order.salesman_id:
|
||||
from backend.app.models.system import SysUser
|
||||
from backend.app.models.system import User
|
||||
from sqlalchemy import select as sa_select
|
||||
user = session.execute(
|
||||
sa_select(SysUser).where(SysUser.id == order.salesman_id)
|
||||
sa_select(User).where(User.id == order.salesman_id)
|
||||
).scalar_one_or_none()
|
||||
if user:
|
||||
salesman_name = user.real_name
|
||||
|
||||
@ -238,7 +238,11 @@ Page({
|
||||
var list = that.data.waybillList.slice();
|
||||
list.push(waybill);
|
||||
that.setData({ waybillList: list });
|
||||
wx.showToast({ title: "识别成功", icon: "success" });
|
||||
if (waybill.tracking_number) {
|
||||
wx.showToast({ title: "识别成功", icon: "success" });
|
||||
} else {
|
||||
wx.showToast({ title: "自动识别未成功，请手动输入单号", icon: "none", duration: 2500 });
|
||||
}
|
||||
}).catch(function (err) {
|
||||
wx.hideLoading();
|
||||
that.setData({ recognizing: false });
|
||||
|
||||
Loading…
Reference in New Issue
Block a user