R360-UAVmodelTool/ali_image_validation.py

import os
import base64
import json
import time
from PIL import Image
from io import BytesIO
import logging
from openai import OpenAI
from retry import retry
import requests

class AliImageValidator:
    def __init__(self, api_key="sk-ccfcdd12fd434d0dab1406958663df9d"):
        self.client = OpenAI(
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
            api_key=api_key or os.environ.get("DASHSCOPE_API_KEY")
        )
        if not self.client.api_key:
            raise ValueError("未找到API密钥，请设置环境变量DASHSCOPE_API_KEY或通过参数传递")

        logging.basicConfig(
            format='%(asctime)s - %(levelname)s - %(message)s',
            level=logging.DEBUG
        )


    @retry(tries=3, delay=1, backoff=2, max_delay=5, exceptions=(Exception,))
    def analyze_image(self, image_url):
        if not image_url.startswith(('http://', 'https://')):
            return {
                'is_false_positive': False,
                'uav_detected': False,
                'probability': 0.0,
                'reason': '无效的图片URL',
                'response_time': 0.0
            }

        system_prompt = """
你是一个专业的图像分析专家，请严格按照以下要求响应：
1. 输出必须为合法的JSON格式，不能是纯自然语言段落
2. 包含三个字段：
   probability（误报概率0-1）
   is_uav（是否是无人机 bool）
   reason（分析理由）
3. 不要包含任何额外说明

请分析该监控画面：
- 判断是否为安全误报
- 识别画面中是否存在无人机
- 给出详细分析理由
"""

        try:
            start_time = time.time()
            model_name = "qwen-vl-max-latest"
            logging.info(f"发送请求到阿里云 | 端点: {self.client.base_url} | 模型: {model_name}")
            response = self.client.chat.completions.create(
                model=model_name,
                messages=[
                    {"role": "system", "content": [{"type": "text", "text": system_prompt}]},
                    {
                        "role": "user",
                        "content": [
                            {
                                "type": "image_url",
                                "image_url": {
                                    "url": image_url
                                }
                            },
                            {"type": "text", "text": "请分析该监控画面是否为误报"}
                        ]
                    }
                ]
            )
            end_time = time.time()
            response_time = end_time - start_time
            
            try:
                result = response.choices[0].message.content
                logging.debug(f"原始响应内容: {result}")
                try:
                    result_json = json.loads(result.split('```json')[1].split('```')[0].strip())  # 提取markdown代码块中的JSON
                    
                    return {
                        'is_false_positive': result_json.get('is_false_positive', False),
                        'uav_detected': result_json.get('is_uav', False),
                        'probability': result_json.get('probability', 0.0),
                        'reason': result_json.get('reason', '无分析结果'),
                        'response_time': response_time
                    }
                except (IndexError, json.JSONDecodeError, KeyError) as e:
                    logging.error(f"响应解析失败 | 错误类型: {type(e).__name__} | 原始响应: {result}")
                    return {
                        'is_false_positive': "误报" in result,
                        'uav_detected': False,
                        'probability': 0.0,
                        'reason': '响应格式异常',
                        'response_time': response_time
                    }
            except (json.JSONDecodeError, ValueError) as e:
                logging.error(f"响应解析失败: {str(e)}")
                return {
                    'is_false_positive': "误报" in result,
                    'uav_detected': False,
                    'probability': 0.0,
                    'reason': '解析失败',
                    'response_time': response_time
                }

        except Exception as e:
            end_time = time.time()
            response_time = end_time - start_time
            logging.error(f"API调用失败 | 端点: {self.client.base_url} | 模型: {model_name} | 错误类型: {type(e).__name__} | 错误详情: {str(e)}")
            return {
                'is_false_positive': False,
                'uav_detected': False,
                'probability': 0.0,
                'reason': 'API调用失败',
                'response_time': response_time
            }

if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description='阿里云图像误报分析工具')
    parser.add_argument('image_url', help='需要分析的图片URL地址')
    args = parser.parse_args()

    validator = AliImageValidator()
    result = validator.analyze_image(args.image_url)
    
    print(f"综合分析结果：\n误报概率: {result['probability']:.2f}\n无人机识别: {'是' if result['uav_detected'] else '否'}\n判定结果: {'误报' if result['is_false_positive'] else '真实威胁'}\n响应时间: {result['response_time']:.2f}秒\n分析理由: {result['reason']}")