D146: 统一密钥加载器 · secrets-loader.py · 不再散落
- secrets-loader.py: 按LOCAL-SECRETS-PATH.hdlp顺序统一加载所有密钥 - qwen-vision.py: 改用secrets-loader.get()获取密钥，删除重复的加载逻辑 - 所有工具脚本今后统一import secrets_loader → 一个入口管理所有密钥
This commit is contained in:
冰朔
parent
e122ac92e3
commit
7c05b1da93
@ -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 <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
|
||||
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))
|
||||
|
||||
58
video-ai-system/tools/secrets-loader.py
Normal file
58
video-ai-system/tools/secrets-loader.py
Normal file
@ -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}")
|
||||
Loading…
x
Reference in New Issue
Block a user