guanghulab/video-ai-system/tools/qwen-vision.py

#!/usr/bin/env python3
"""铸渊之眼 · 通义千问视觉分析器
用阿里百炼 qwen-vl 模型看图片，输出风格/色调/构图分析

用法:
  python3 qwen-vision.py <image.jpg>                     # 单图分析
  python3 qwen-vision.py <image1.jpg> <image2.jpg>       # 双图对比
"""

import sys, os, json, base64
from urllib.request import Request, urlopen
from urllib.error import URLError

# === 配置 ===
# 从 .env 读 key
env_path = os.path.expanduser("~/guanghulab/video-ai-system/.env")
api_key = None
if os.path.exists(env_path):
    for line in open(env_path):
        line = line.strip()
        if line.startswith("ALIYUN_API_KEY="):
            api_key = line.split("=", 1)[1].strip()
            break

if not api_key:
    print(json.dumps({"error": "未找到ALIYUN_API_KEY"}))
    sys.exit(1)

# 端点：先试公网，再试北京
ENDPOINTS = [
    "https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation",
]
MODELS = ["qwen-vl-max", "qwen3-vl-plus", "qwen-vl-plus"]

def encode_image(path):
    """读取图片并转为base64 data URI"""
    with open(path, "rb") as f:
        b64 = base64.b64encode(f.read()).decode()
    ext = path.rsplit(".", 1)[-1].lower()
    mime = {"jpg": "jpeg", "jpeg": "jpeg", "png": "png", "webp": "webp"}.get(ext, "jpeg")
    return f"data:image/{mime};base64,{b64}"

def call_vision(images, prompt, model, endpoint):
    """调用视觉模型"""
    content = []
    for img in images:
        content.append({"image": img})
    content.append({"text": prompt})
    
    body = {
        "model": model,
        "input": {"messages": [{"role": "user", "content": content}]}
    }
    
    req = Request(
        endpoint,
        data=json.dumps(body).encode(),
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    )
    
    resp = urlopen(req, timeout=60)
    return json.loads(resp.read())

def extract_content(response):
    """从响应中提取文本内容"""
    try:
        return response["output"]["choices"][0]["message"]["content"][0]["text"]
    except:
        return json.dumps(response, ensure_ascii=False)

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: qwen-vision.py <image> [image2]")
        sys.exit(1)
    
    images = [encode_image(p) for p in sys.argv[1:]]
    
    if len(images) == 1:
        prompt = """请详细分析这张图片的视觉特征，输出JSON格式：
{
  "style": "渲染风格（如3D动漫/2D手绘/真人写实/UE5游戏等）",
  "color_palette": ["主色调1", "主色调2", "主色调3"],
  "lighting": "光影风格描述",
  "composition": "构图方式（特写/中景/全景/俯视/平视等）",
  "key_elements": ["画面中的关键元素"],
  "text_content": "画面中出现的所有文字内容",
  "mood": "氛围感受"
}
只输出JSON，不要其他文字。"""
    else:
        prompt = """请对比这两张图片，输出JSON格式：
{
  "style_match": true或false,
  "style_match_detail": "两张图渲染风格是否一致的具体说明",
  "color_consistency": "色调是否一致，给出0-100分",
  "composition_match": "构图方式是否协调",
  "key_differences": ["主要差异点"],
  "recommendation": "如果要让第二张图匹配第一张图的风格，建议修改什么"
}
只输出JSON，不要其他文字。"""
    
    # 尝试不同模型和端点
    result = None
    for model in MODELS:
        for ep in ENDPOINTS:
            try:
                print(f"[尝试] {model} @ {ep[:50]}...", file=sys.stderr)
                resp = call_vision(images, prompt, model, ep)
                content = extract_content(resp)
                # 尝试解析JSON
                try:
                    # 提取JSON（可能被markdown包裹）
                    if "```json" in content:
                        content = content.split("```json")[1].split("```")[0]
                    elif "```" in content:
                        content = content.split("```")[1].split("```")[0]
                    parsed = json.loads(content.strip())
                    parsed["_model"] = model
                    parsed["_endpoint"] = ep
                    print(json.dumps(parsed, ensure_ascii=False, indent=2))
                    sys.exit(0)
                except json.JSONDecodeError:
                    print(content)
                    sys.exit(0)
            except URLError as e:
                print(f"[失败] {model}: {e}", file=sys.stderr)
                continue
            except Exception as e:
                print(f"[异常] {model}: {e}", file=sys.stderr)
                continue
    
    print(json.dumps({"error": "所有模型/端点都失败了"}, ensure_ascii=False))
    sys.exit(1)