cang-ying/tools/qwen-vision.py

95 lines
4.0 KiB
Python
Raw Permalink Normal View History

#!/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)