59 lines
2.0 KiB
Python
59 lines
2.0 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_data: bytes = None, url: str = None) -> Dict[str, Any]:
|
|
"""
|
|
从图片中提取取件码
|
|
|
|
Args:
|
|
image_content: 图片二进制内容
|
|
|
|
Returns:
|
|
Dict: 提取结果，包含取件码信息
|
|
"""
|
|
try:
|
|
primary_result = await self._extract_with_qwen(image_data, 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_data: bytes = None, url: str = None) -> Dict[str, Any]:
|
|
"""使用千问提取取件码"""
|
|
try:
|
|
# 添加超时控制
|
|
return await asyncio.wait_for(
|
|
qwen_client.extract_pickup_code(image_data, 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: List[Dict[str, Any]]) -> bool:
|
|
|
|
return isinstance(result, list)
|
|
|
|
# 创建全局实例
|
|
ai_client = AIClient() |