73 lines
2.3 KiB
Python
73 lines
2.3 KiB
Python
import logging
|
|
import json
|
|
import base64
|
|
from typing import Dict, Any, Optional, List
|
|
import asyncio
|
|
from app.core.config import settings
|
|
from app.core.qwen_client import qwen_client
|
|
|
|
class AIClient:
|
|
"""AI 客户端，统一包装千问和 DeepSeek"""
|
|
|
|
def __init__(self):
|
|
self.timeout = 15 # 请求超时时间（秒）
|
|
|
|
async def extract_pickup_code(self, image_url: str) -> Dict[str, Any]:
|
|
"""
|
|
从图片中提取取件码
|
|
|
|
Args:
|
|
image_content: 图片二进制内容
|
|
|
|
Returns:
|
|
Dict: 提取结果，包含取件码信息
|
|
"""
|
|
try:
|
|
primary_result = await self._extract_with_qwen(image_url)
|
|
|
|
# 检查结果是否有效
|
|
if self._is_valid_result(primary_result):
|
|
return primary_result
|
|
|
|
return {"error": "处理失败", "message": "提取取件码失败"}
|
|
|
|
except Exception as e:
|
|
logging.exception(f"提取取件码异常: {str(e)}")
|
|
return {"error": "处理失败", "message": str(e)}
|
|
|
|
async def _extract_with_qwen(self, image_url: str) -> Dict[str, Any]:
|
|
"""使用千问提取取件码"""
|
|
try:
|
|
# 添加超时控制
|
|
return await asyncio.wait_for(
|
|
qwen_client.extract_pickup_code(image_url),
|
|
timeout=self.timeout
|
|
)
|
|
except asyncio.TimeoutError:
|
|
logging.error("千问 API 请求超时")
|
|
return {"error": "API请求超时", "details": "千问 API 请求超时"}
|
|
except Exception as e:
|
|
logging.exception(f"千问提取异常: {str(e)}")
|
|
return {"error": "处理失败", "message": str(e)}
|
|
|
|
|
|
def _is_valid_result(self, result: Dict[str, Any]) -> bool:
|
|
"""检查结果是否有效"""
|
|
# 检查是否有错误
|
|
if "error" in result:
|
|
return False
|
|
|
|
# 检查是否有站点信息
|
|
stations = result.get("stations", [])
|
|
if not stations:
|
|
return False
|
|
|
|
# 检查是否有取件码
|
|
for station in stations:
|
|
if station.get("pickup_codes") and len(station.get("pickup_codes", [])) > 0:
|
|
return True
|
|
|
|
return False
|
|
|
|
# 创建全局实例
|
|
ai_client = AIClient() |