Guanghu Domestic Migration d1e47f4565
Some checks are pending
自动更新代码和重启 / update-and-restart (push) Waiting to run
CI检查 + 自动部署 / check (push) Waiting to run
CI检查 + 自动部署 / deploy (push) Blocked by required conditions
重启聊天服务 / restart (push) Waiting to run
chore: import sanitized domestic snapshot for REPO-002
Source snapshot: ca48d3ddf926d79aa138306164169baf764bb829
2026-07-17 15:54:41 +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)