cang-ying/tools/qwen-vision.py
Guanghu Domestic Migration a2fa7d57d8 chore: import sanitized domestic snapshot for REPO-005
Source snapshot: 23037fa3a2e9b0597162735755e92fc763bf4d94
2026-07-17 15:55:48 +08:00

95 lines
4.0 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""铸渊之眼 · 通义千问视觉分析器
用阿里百炼 qwen-vl 模型看图片,输出风格/色调/构图分析
⚠️ 密钥通过 secrets-loader.py 统一加载。路径见 LOCAL-SECRETS-PATH.hdlp。
用法:
python3 qwen-vision.py <image.jpg> # 单图分析
python3 qwen-vision.py <image1.jpg> <image2.jpg> # 双图对比
"""
import sys, os, json, base64, subprocess
# 统一密钥加载 · 编号路由
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from secrets_loader import secret, endpoint
api_key = secret("SC-004") # 阿里千问VL视觉
ep = endpoint("EPT-002") # 业务空间端点
if not api_key:
print(json.dumps({"error": "未找到ALIYUN_QWEN_VL_KEY。密钥通过 secrets-loader.py 加载。→ LOCAL-SECRETS-PATH.hdlp"}))
sys.exit(1)
ENDPOINTS = [ep, "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):
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, ep):
content = []
for img in images:
content.append({"image": img})
content.append({"text": prompt})
body = json.dumps({"model": model, "input": {"messages": [{"role": "user", "content": content}]}})
proc = subprocess.run(["curl", "-s", "-X", "POST", ep,
"-H", f"Authorization: Bearer {api_key}",
"-H", "Content-Type: application/json",
"--data-binary", "@-"], input=body, capture_output=True, text=True, timeout=120)
if proc.returncode != 0 or not proc.stdout.strip():
raise Exception(f"curl失败: {proc.stderr[:200]}")
return json.loads(proc.stdout)
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":"渲染风格","color_palette":["主色调"],"lighting":"光影风格","composition":"构图方式","key_elements":["关键元素"],"text_content":"画面文字","mood":"氛围感受"}
只输出JSON不要其他文字。"""
else:
prompt = """对比两张图输出JSON
{"style_match":true/false,"style_diff":"风格差异描述","color_consistency":"色调一致性0-100","composition_coherence":"构图连贯性","key_differences":["差异"],"recommendation":"修改建议"}
只输出JSON不要其他文字。"""
for model in MODELS:
for ep in ENDPOINTS:
try:
print(f"[{model}]", file=sys.stderr)
resp = call_vision(images, prompt, model, ep)
content = extract_content(resp)
try:
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
print(json.dumps(parsed, ensure_ascii=False, indent=2))
sys.exit(0)
except json.JSONDecodeError:
print(content)
sys.exit(0)
except Exception as e:
print(f" fail: {e}", file=sys.stderr)
continue
print(json.dumps({"error": "所有模型/端点都失败了"}, ensure_ascii=False))
sys.exit(1)