update

app/api/endpoints/ai.py

@@ -6,11 +6,20 @@ from app.models.user import UserDB
 import logging
 from app.core.qcloud import qcloud_manager
 from pydantic import BaseModel
+from typing import Optional
+
+from app.core.qwen_client import qwen_client
 router = APIRouter()
 
 class ExtractPickupCodeRequest(BaseModel):
     url: str
 
+class VLRequest(BaseModel):
+    imageData: Optional[bytes] = None
+    url: Optional[str] = None
+    prompt: Optional[str] = None
+    user_content: Optional[str] = None
+
 @router.post("/recognition/pickup_code_with_url", response_model=ResponseModel)
 async def extract_pickup_code_with_url(
     data: ExtractPickupCodeRequest
@@ -58,4 +67,24 @@ async def extract_pickup_code(
         
     except Exception as e:
         logging.exception(f"提取取件码失败: {str(e)}")
-        return error_response(code=500, message=f"提取取件码失败: {str(e)}")
+        return error_response(code=500, message=f"提取取件码失败: {str(e)}")
+    
+@router.post("/qw/vl", response_model=ResponseModel)
+async def extract_pickup_code_with_url(
+    data: VLRequest
+):
+    """从图片 URL 中提取取件码"""
+    
+    try:
+        # 调用 AI 客户端提取取件码  
+        result = await qwen_client.vl(imageData=data.imageData, url=data.url, prompt=data.prompt, user_content=data.user_content)
+        if "error" in result:
+            return error_response(code=500, message=result.get("message", "返回失败"))
+        
+        # 返回原始数据和格式化文本
+        return success_response(data=result)
+        
+    except Exception as e:
+        logging.exception(f"vl失败: {str(e)}")
+        return error_response(code=500, message=f"vl失败: {str(e)}")
+    

app/core/qwen_client.py

@@ -113,6 +113,95 @@ class QwenClient:
         except Exception as e:
             logging.exception(f"调用千问 API 异常: {str(e)}")
             return None
+    
+    async def vl(self, imageData: bytes = None, url: str = None, prompt: str = None, user_content : str = None) -> Dict[str, Any]:
+        """
+        从图片中提取取件码
+        
+        Args:
+            image_content: 图片二进制内容
+            
+        Returns:
+            Dict: 提取结果，包含取件码信息
+        """
+        try:            
+            # 构建消息
+            messages = [
+                {
+                    "role": "system", 
+                    "content": prompt
+                },
+                {
+                    "role": "user", 
+                    "content": [
+                        {
+                            "type": "text", 
+                            "text": user_content
+                        },
+                        {
+                            "type": "image",
+                            "image": f"data:image/jpeg;base64,{base64.b64encode(imageData).decode('utf-8')}" if imageData else url,
+                        }
+                    ]
+                }
+            ]
+            
+            # 使用 SDK 调用 API
+            response = MultiModalConversation.call(
+                model=self.model,
+                messages=messages,
+                api_key=self.api_key,
+                result_format='message',
+                temperature=0.1,
+                max_tokens=1000
+            )
+
+            print(f"response_json: {response}")
+            
+            # 记录响应信息（用于调试）
+            logging.info(f"千问 API 响应状态: {response.status_code}")
+            
+            # 检查响应状态
+            if response.status_code != 200:
+                logging.error(f"千问 API 请求失败: {response.code} - {response.message}")
+                return {"error": "API请求失败", "details": f"{response.code}: {response.message}"}
+            
+            try:
+                output = response.output
+                print(f"千问output: {output}")
+                if output:
+                    choices = output.get('choices', [])
+                    if choices and len(choices) > 0:
+                        message = choices[0].get('message', {})
+                        content = message.get('content', [])
+                        
+                        if isinstance(content, list) and len(content) > 0:
+                            text_content = ""
+                            result = []
+                            for item in content:
+                                if isinstance(item, dict) and 'text' in item:
+                                    text_content = item.get('text', '')
+                                    print(f"提取的文本内容: {text_content}")
+                                    # 只获取 ```json 和 ``` 之间的内容
+                                    text_content = re.search(r'```json(.*)```', text_content, re.DOTALL).group(1)
+
+                                    # 剔除 ```json 和 ```
+                                    # text_content = text_content.replace('```json', '').replace('```', '')
+
+                                    # 尝试直接解析
+                                    pickup_data = json.loads(text_content)
+                                    print(f"pickup_data: {pickup_data}")
+                                    result = pickup_data                                    
+                            
+                            return result
+                            
+            except Exception as e:
+                logging.exception(f"解析千问 API 响应失败: {str(e)}")
+                return None
+                
+        except Exception as e:
+            logging.exception(f"调用千问 API 异常: {str(e)}")
+            return None
 
 # 创建全局实例
 qwen_client = QwenClient()