cang-ying/tools/secrets_loader.py

#!/usr/bin/env python3
"""
光湖语言系统 · 统一密钥加载器 v2.0 · API守门人版
Guanghu Language System · Secrets Loader with CA-API-Guard

设计哲学:
  所有API密钥通过苍耳API守门人(ca-api-guard)获取。
  调用API前 → 守门人发送验证码到苍耳QQ邮箱 → 苍耳确认 → 密钥释放。
  密钥本身永不暴露给AI本地文件。

用法:
  from tools.secrets_loader import secret, endpoint, request_approval
  key = secret("SC-001")       # 获取已授权的API密钥
  url = endpoint("EPT-001")    # 获取端点
  cid = request_approval("SC-001", "EED-PROJ-001", "第1集第3镜视频生成")  # 请求授权

编号→变量名映射见 LOCAL-SECRETS-PATH.hdlp
"""

import os, json, time, urllib.request, sys

# ═══════════════════════════════════════════
# 配置
# ═══════════════════════════════════════════

# 苍耳API守门人地址（SG-001 大脑服务器）
API_GUARD_URL = os.environ.get("CA_API_GUARD_URL", "http://10.3.4.11:8923")
# 如果从外部访问，使用公网地址
API_GUARD_PUBLIC_URL = os.environ.get("CA_API_GUARD_PUBLIC_URL", "http://43.156.237.110:8923")

# 本地缓存（已验证的密钥，短期有效）
_CACHE_FILE = os.path.expanduser("~/.ca-api-guard/cache.json")
_CACHE = {}
_CACHE_TTL = 600  # 10分钟缓存

# ═══════════════════════════════════════════
# 编号→变量名映射
# ═══════════════════════════════════════════
_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",
}

_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"),
}


def _load_cache():
    """加载本地缓存"""
    global _CACHE
    if _CACHE:
        return
    try:
        if os.path.exists(_CACHE_FILE):
            with open(_CACHE_FILE) as f:
                _CACHE = json.load(f)
    except:
        _CACHE = {}


def _save_cache():
    """保存本地缓存"""
    try:
        os.makedirs(os.path.dirname(_CACHE_FILE), exist_ok=True)
        with open(_CACHE_FILE, "w") as f:
            json.dump(_CACHE, f)
    except:
        pass


def _call_guard(endpoint, data=None, public=False):
    """调用苍耳API守门人"""
    url = (API_GUARD_PUBLIC_URL if public else API_GUARD_URL) + endpoint
    try:
        req = urllib.request.Request(url)
        req.add_header("Content-Type", "application/json")
        if data:
            req.data = json.dumps(data).encode()
        resp = urllib.request.urlopen(req, timeout=10)
        return json.loads(resp.read())
    except Exception as e:
        return {"ok": False, "error": str(e)}


def request_approval(api_code, caller_id, description):
    """
    请求API调用授权 → 发送验证码到苍耳邮箱
    返回: {"ok": true, "challenge_id": "xxx", "message": "..."}
    """
    return _call_guard("/api/request", {
        "api_code": api_code,
        "caller_id": caller_id,
        "description": description
    }, public=True)


def confirm_approval(challenge_id, code):
    """
    确认验证码 → 获取API密钥并缓存
    返回: {"ok": true, "api_code": "xxx", "api_key": "xxx"}
    """
    result = _call_guard("/api/confirm", {
        "challenge_id": challenge_id,
        "code": code
    }, public=True)
    if result.get("ok") and result.get("api_key"):
        _load_cache()
        _CACHE[result["api_code"]] = {
            "key": result["api_key"],
            "cached_at": time.time()
        }
        _save_cache()
    return result


def secret(code_or_var, default=None):
    """
    按编号获取API密钥。
    优先从本地缓存读取，缓存未命中则尝试从守门人获取。
    
    如果密钥未授权 → 返回空字符串。
    调用者应检查返回值，如果为空 → 调用 request_approval() 请求授权。
    """
    # 先查本地缓存
    _load_cache()
    api_code = code_or_var if code_or_var in _SECRET_MAP else None
    if not api_code:
        # 反向查找（变量名→编号）
        for sc, var in _SECRET_MAP.items():
            if var == code_or_var:
                api_code = sc
                break
    if not api_code:
        api_code = code_or_var

    if api_code in _CACHE:
        cached = _CACHE[api_code]
        if time.time() - cached.get("cached_at", 0) < _CACHE_TTL:
            return cached.get("key", default)

    # 尝试从守门人获取（如果有已授权的缓存）
    result = _call_guard("/api/request", {
        "api_code": api_code,
        "caller_id": "auto",
        "description": "自动获取缓存密钥"
    })
    
    # 如果守门人返回了密钥（预授权模式），缓存并返回
    if result.get("ok") and result.get("api_key"):
        _CACHE[api_code] = {
            "key": result["api_key"],
            "cached_at": time.time()
        }
        _save_cache()
        return result["api_key"]

    return default or ""


def endpoint(code_or_var, default=None):
    """按编号获取端点URL"""
    var_name, fallback = _ENDPOINT_MAP.get(code_or_var, (code_or_var, ""))
    # 从环境变量获取（如果有）
    val = os.environ.get(var_name, "")
    return val if val else (fallback if fallback else default)


if __name__ == "__main__":
    print("=== 苍耳API守门人 · 密钥加载器 v2.0 ===")
    print(f"守门人地址: {API_GUARD_URL}")
    print()
    
    # 检查健康状态
    health = _call_guard("/health")
    print(f"守门人状态: {json.dumps(health, ensure_ascii=False)}")
    print()
    
    # 列出可用API
    print("可用API编号:")
    for code, var in _SECRET_MAP.items():
        key = secret(code)
        status = "✅ 已缓存" if key else "⏳ 需授权"
        print(f"  {code} → {var} [{status}]")