From 7c05b1da93d8e447f29ff3b361ed9bc5468f8146 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=B0=E6=9C=94?= <565183519@qq.com> Date: Fri, 26 Jun 2026 13:59:30 +0800 Subject: [PATCH] =?UTF-8?q?D146:=20=E7=BB=9F=E4=B8=80=E5=AF=86=E9=92=A5?= =?UTF-8?q?=E5=8A=A0=E8=BD=BD=E5=99=A8=20=C2=B7=20secrets-loader.py=20?= =?UTF-8?q?=C2=B7=20=E4=B8=8D=E5=86=8D=E6=95=A3=E8=90=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - secrets-loader.py: 按LOCAL-SECRETS-PATH.hdlp顺序统一加载所有密钥 - qwen-vision.py: 改用secrets-loader.get()获取密钥，删除重复的加载逻辑 - 所有工具脚本今后统一import secrets_loader → 一个入口管理所有密钥 --- video-ai-system/tools/qwen-vision.py | 119 +++++------------------- video-ai-system/tools/secrets-loader.py | 58 ++++++++++++ 2 files changed, 82 insertions(+), 95 deletions(-) create mode 100644 video-ai-system/tools/secrets-loader.py diff --git a/video-ai-system/tools/qwen-vision.py b/video-ai-system/tools/qwen-vision.py index 795c2a9..7759ca7 100644 --- a/video-ai-system/tools/qwen-vision.py +++ b/video-ai-system/tools/qwen-vision.py @@ -2,100 +2,52 @@ """铸渊之眼 · 通义千问视觉分析器 用阿里百炼 qwen-vl 模型看图片，输出风格/色调/构图分析 -⚠️ 密钥不在代码仓库里。读取规则见 LOCAL-SECRETS-PATH.hdlp。 -密钥变量: ALIYUN_QWEN_VL_KEY + ALIYUN_QWEN_VL_ENDPOINT -读取顺序: - 1. /Users/bingshuolingdianyuanhe/Documents/guanghulab-local-secrets/video-ai-system.env - 2. ~/guanghulab/video-ai-system/.env - 3. 当前进程环境变量 +⚠️ 密钥通过 secrets-loader.py 统一加载。路径见 LOCAL-SECRETS-PATH.hdlp。 用法: python3 qwen-vision.py # 单图分析 python3 qwen-vision.py # 双图对比 """ -import sys, os, json, base64 -from urllib.request import Request, urlopen -from urllib.error import URLError +import sys, os, json, base64, subprocess -# === 密钥加载 === -# 按 LOCAL-SECRETS-PATH.hdlp 规定的顺序加载 -SECRET_PATHS = [ - "/Users/bingshuolingdianyuanhe/Documents/guanghulab-local-secrets/video-ai-system.env", - os.path.expanduser("~/guanghulab/video-ai-system/.env"), -] +# 统一密钥加载 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from secrets_loader import get -api_key = None -endpoint = None - -# 先试环境变量 -api_key = os.environ.get("ALIYUN_QWEN_VL_KEY") or os.environ.get("ALIYUN_BAILIAN_API_KEY") -endpoint = os.environ.get("ALIYUN_QWEN_VL_ENDPOINT") or "https://ws-umd6xwlovzmshuat.cn-beijing.maas.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation" - -# 再从文件读 -if not api_key: - for secret_path in SECRET_PATHS: - if os.path.exists(secret_path): - for line in open(secret_path): - line = line.strip() - if not line or line.startswith("#"): - continue - if "=" in line: - k, v = line.split("=", 1) - k = k.strip() - v = v.strip() - if k in ("ALIYUN_QWEN_VL_KEY", "ALIYUN_API_KEY", "ALIYUN_BAILIAN_API_KEY") and v and not api_key: - api_key = v - if k == "ALIYUN_QWEN_VL_ENDPOINT" and v and not os.environ.get("ALIYUN_QWEN_VL_ENDPOINT"): - endpoint = v - if api_key: - break +api_key = get("ALIYUN_QWEN_VL_KEY") +endpoint = get("ALIYUN_QWEN_VL_ENDPOINT", + "https://ws-umd6xwlovzmshuat.cn-beijing.maas.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation") if not api_key: - print(json.dumps({"error": "未找到ALIYUN_QWEN_VL_KEY。请确认密钥文件存在。路径: LOCAL-SECRETS-PATH.hdlp"})) + print(json.dumps({"error": "未找到ALIYUN_QWEN_VL_KEY。密钥通过 secrets-loader.py 加载。→ LOCAL-SECRETS-PATH.hdlp"})) sys.exit(1) -# 端点 -ENDPOINTS = [ - endpoint, - "https://dashscope.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation", -] +ENDPOINTS = [endpoint, "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): - """调用视觉模型""" +def call_vision(images, prompt, model, ep): 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()) + 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: @@ -110,56 +62,33 @@ if __name__ == "__main__": if len(images) == 1: prompt = """请详细分析这张图片的视觉特征，输出JSON格式： -{ - "style": "渲染风格（如3D动漫/2D手绘/真人写实/UE5游戏等）", - "color_palette": ["主色调1", "主色调2", "主色调3"], - "lighting": "光影风格描述", - "composition": "构图方式（特写/中景/全景/俯视/平视等）", - "key_elements": ["画面中的关键元素"], - "text_content": "画面中出现的所有文字内容", - "mood": "氛围感受" -} +{"style":"渲染风格","color_palette":["主色调"],"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": "如果要让第二张图匹配第一张图的风格，建议修改什么" -} + prompt = """对比两张图，输出JSON： +{"style_match":true/false,"style_diff":"风格差异描述","color_consistency":"色调一致性0-100","composition_coherence":"构图连贯性","key_differences":["差异"],"recommendation":"修改建议"} 只输出JSON，不要其他文字。""" - # 尝试不同模型和端点 - result = None for model in MODELS: for ep in ENDPOINTS: try: - print(f"[尝试] {model} @ {ep[:50]}...", file=sys.stderr) + print(f"[{model}]", 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) + print(f" fail: {e}", file=sys.stderr) continue print(json.dumps({"error": "所有模型/端点都失败了"}, ensure_ascii=False)) diff --git a/video-ai-system/tools/secrets-loader.py b/video-ai-system/tools/secrets-loader.py new file mode 100644 index 0000000..f099a21 --- /dev/null +++ b/video-ai-system/tools/secrets-loader.py @@ -0,0 +1,58 @@ +#!/usr/bin/env python3 +"""视频AI系统 · 统一密钥加载器 +所有工具脚本通过本模块获取密钥，不各自散落读取逻辑。 + +用法: + from tools.secrets_loader import secrets + key = secrets["JIMENG_API_KEY"] + +密钥来源: video-ai-system/LOCAL-SECRETS-PATH.hdlp +读取顺序: 本地secrets文件 → 仓库.env → 环境变量 → 空值不覆盖 +""" + +import os + +# 密钥文件路径（按优先级） +_SECRET_FILES = [ + "/Users/bingshuolingdianyuanhe/Documents/guanghulab-local-secrets/video-ai-system.env", + os.path.expanduser("~/guanghulab/video-ai-system/.env"), +] + +def _load(): + """加载所有密钥到 dict，高优先级文件的同名 key 不被低优先级覆盖""" + result = {} + + # 先从低优先级加载 + for path in reversed(_SECRET_FILES): + if os.path.exists(path): + with open(path) as f: + for line in f: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + k, v = line.split("=", 1) + k, v = k.strip(), v.strip() + if k not in result: + result[k] = v + + # 最后加载环境变量（最高优先级） + for key in list(result.keys()): + env_val = os.environ.get(key) + if env_val: + result[key] = env_val + + return result + +secrets = _load() + +# 快速访问 +def get(key, default=None): + return secrets.get(key, default) + +if __name__ == "__main__": + # 诊断模式：列出所有已加载的密钥变量名（不显示值） + print("=== 已加载密钥变量 ===") + for k in sorted(secrets.keys()): + v = secrets[k] + masked = v[:8] + "..." + v[-4:] if len(v) > 12 else "***" + print(f" {k} = {masked}")