143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""苍耳本地密钥加载器 · 唯一正式入口。
|
|
|
|
真实密钥只存在苍耳实际运行机器的仓库外文件或进程环境变量中。
|
|
仓库只维护 SC/EPT 编号映射与候选路径，不保存、打印或远程索取密钥。
|
|
|
|
可用 ``VIDEO_AI_SECRETS_FILE`` 显式指定本机密钥文件；未指定时按
|
|
``ENVIRONMENT.hdlp`` 中登记的本机路径查找。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict, Iterable, Optional
|
|
|
|
|
|
_SECRET_MAP = {
|
|
"SC-001": "JIMENG_API_KEY",
|
|
"SC-002": "VOLC_VOICE_API_KEY",
|
|
"SC-003": "VOLC_VOICE_ACCESS_TOKEN",
|
|
"SC-004": "ALIYUN_QWEN_VL_KEY",
|
|
"SC-005": "ALIYUN_WANXIANG_KEY",
|
|
"SC-006": "KLING_API_KEY",
|
|
"SC-007": "ALIYUN_API_KEY",
|
|
"SC-008": "WORKRALLY_API_KEY",
|
|
"SC-009": "VOLC_VOICE_APP_ID",
|
|
"SC-010": "VOLC_VOICE_SECRET_KEY",
|
|
"SC-011": "GITEA_PAT",
|
|
# 旧编号只做本地兼容，不建立第二条密钥路线。
|
|
"ZY-SEC-001": "JIMENG_API_KEY",
|
|
"ZY-SEC-004": "ALIYUN_QWEN_VL_KEY",
|
|
}
|
|
|
|
_ENDPOINT_MAP = {
|
|
"EPT-001": ("JIMENG_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3"),
|
|
"EPT-002": (
|
|
"ALIYUN_QWEN_VL_ENDPOINT",
|
|
"https://ws-umd6xwlovzmshuat.cn-beijing.maas.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation",
|
|
),
|
|
"EPT-003": ("ALIYUN_BAILIAN_BASE_URL", "https://dashscope.aliyuncs.com/api/v1"),
|
|
"EPT-004": ("WORKRALLY_ENDPOINT", "https://workrally.qq.com/zenstudio/api/mcp"),
|
|
}
|
|
|
|
_LOCAL_FILES = (
|
|
Path("/root/guanghulab-local-secrets/cang-ying.env"),
|
|
Path("~/guanghulab-local-secrets/cang-ying.env").expanduser(),
|
|
Path(
|
|
"/Users/bingshuolingdianyuanhe/Documents/guanghulab-local-secrets/"
|
|
"video-ai-system.env"
|
|
),
|
|
)
|
|
|
|
_raw: Optional[Dict[str, str]] = None
|
|
_loaded_file: Optional[Path] = None
|
|
|
|
|
|
def _candidate_files() -> Iterable[Path]:
|
|
explicit = os.environ.get("VIDEO_AI_SECRETS_FILE", "").strip()
|
|
if explicit:
|
|
yield Path(explicit).expanduser()
|
|
yield from _LOCAL_FILES
|
|
|
|
|
|
def _parse_env_file(path: Path) -> Dict[str, str]:
|
|
values: Dict[str, str] = {}
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
for raw_line in handle:
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
key = key.strip()
|
|
if not key:
|
|
continue
|
|
values[key] = value.strip().strip('"').strip("'")
|
|
return values
|
|
|
|
|
|
def _load_local() -> Dict[str, str]:
|
|
global _raw, _loaded_file
|
|
if _raw is not None:
|
|
return _raw
|
|
|
|
_raw = {}
|
|
for path in _candidate_files():
|
|
if path.is_file():
|
|
_raw = _parse_env_file(path)
|
|
_loaded_file = path
|
|
break
|
|
return _raw
|
|
|
|
|
|
def secret(code_or_var: str, default=None):
|
|
"""按 SC 编号或变量名读取本机密钥；环境变量优先。"""
|
|
variable = _SECRET_MAP.get(code_or_var, code_or_var)
|
|
environment_value = os.environ.get(variable)
|
|
if environment_value:
|
|
return environment_value
|
|
return _load_local().get(variable, default)
|
|
|
|
|
|
def endpoint(code_or_var: str, default=None):
|
|
"""按 EPT 编号读取本机覆盖值，未配置时返回公开默认端点。"""
|
|
variable, fallback = _ENDPOINT_MAP.get(code_or_var, (code_or_var, ""))
|
|
value = os.environ.get(variable) or _load_local().get(variable, "")
|
|
return value or fallback or default
|
|
|
|
|
|
def source_status() -> dict:
|
|
"""只报告来源状态和路径，不读取到输出、不展示密钥内容。"""
|
|
_load_local()
|
|
configured = sorted(
|
|
code for code, variable in _SECRET_MAP.items()
|
|
if code.startswith("SC-") and bool(os.environ.get(variable) or _raw.get(variable))
|
|
)
|
|
return {
|
|
"mode": "local-only",
|
|
"source": str(_loaded_file) if _loaded_file else "environment-or-not-found",
|
|
"configured_codes": configured,
|
|
}
|
|
|
|
|
|
def request_approval(*_args, **_kwargs):
|
|
"""旧守门人接口的停用路标；不会联网或发送验证码。"""
|
|
return {
|
|
"ok": False,
|
|
"retired": True,
|
|
"error": "远程邮件守门人已停用；请按 LOCAL-SECRETS-PATH.hdlp 使用苍耳本机密钥入口。",
|
|
}
|
|
|
|
|
|
def confirm_approval(*_args, **_kwargs):
|
|
"""旧守门人接口的停用路标；不会联网或接收验证码。"""
|
|
return request_approval()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
status = source_status()
|
|
print(f"mode={status['mode']}")
|
|
print(f"source={status['source']}")
|
|
print("configured=" + ",".join(status["configured_codes"]))
|