245 lines
7.4 KiB
Python
245 lines
7.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Doppler配置管理模块
|
||
安全地从Doppler获取配置和密钥
|
||
"""
|
||
|
||
import os
|
||
from typing import Optional, Dict, Any
|
||
|
||
# 新增：优先加载 .env（若存在）
|
||
try:
|
||
from dotenv import load_dotenv, find_dotenv # type: ignore
|
||
_env_path = find_dotenv()
|
||
if _env_path:
|
||
load_dotenv(_env_path)
|
||
else:
|
||
# 尝试从项目根目录加载 .env
|
||
from pathlib import Path
|
||
root_env = Path(__file__).resolve().parents[1] / '.env'
|
||
if root_env.exists():
|
||
load_dotenv(root_env)
|
||
except Exception:
|
||
# 若未安装 python-dotenv 或加载失败，则跳过
|
||
pass
|
||
|
||
def get_secret(key: str, default: Optional[str] = None) -> Optional[str]:
|
||
"""
|
||
从Doppler或环境变量获取密钥
|
||
|
||
Args:
|
||
key: 密钥名称
|
||
default: 默认值
|
||
|
||
Returns:
|
||
密钥值或默认值
|
||
"""
|
||
# 临时的、不安全的解决方案，仅用于测试
|
||
temp_secrets = {
|
||
"RAPIDAPI_KEY": "your_rapidapi_key_here",
|
||
"OPENROUTER_API_KEY_1": "your_openrouter_key_here",
|
||
"GOOGLE_API_KEY": "your_google_api_key_here"
|
||
}
|
||
|
||
# 首先尝试从环境变量获取（Doppler会注入到环境变量，或由 .env 加载）
|
||
value = os.getenv(key)
|
||
|
||
if not value:
|
||
value = temp_secrets.get(key, default)
|
||
|
||
if not value and default is None:
|
||
raise ValueError(f"Required secret '{key}' not found in environment variables or temp_secrets")
|
||
|
||
return value
|
||
|
||
def get_rapidapi_key() -> str:
|
||
"""
|
||
获取RapidAPI密钥
|
||
|
||
Returns:
|
||
RapidAPI密钥
|
||
|
||
Raises:
|
||
ValueError: 如果密钥未找到
|
||
"""
|
||
return get_secret('RAPIDAPI_KEY')
|
||
|
||
def get_openrouter_key() -> str:
|
||
"""
|
||
获取OpenRouter API密钥
|
||
|
||
Returns:
|
||
OpenRouter API密钥
|
||
|
||
Raises:
|
||
ValueError: 如果密钥未找到
|
||
"""
|
||
return get_secret('OPENROUTER_API_KEY_1')
|
||
|
||
def get_google_api_key() -> str:
|
||
"""
|
||
获取Google API密钥 (用于 Gemini/ADK)
|
||
|
||
Returns:
|
||
Google API密钥
|
||
|
||
Raises:
|
||
ValueError: 如果密钥未找到
|
||
"""
|
||
return get_secret('GOOGLE_API_KEY')
|
||
|
||
def get_google_genai_config() -> Dict[str, str]:
|
||
"""
|
||
获取Google GenAI完整配置
|
||
|
||
Returns:
|
||
Google GenAI配置字典
|
||
"""
|
||
use_vertex_ai = get_secret('GOOGLE_GENAI_USE_VERTEXAI', 'FALSE').upper() == 'TRUE'
|
||
api_key = '' if use_vertex_ai else get_secret('GOOGLE_API_KEY', '')
|
||
|
||
return {
|
||
'api_key': api_key,
|
||
'use_vertex_ai': str(use_vertex_ai).upper(),
|
||
'project_id': get_secret('GOOGLE_CLOUD_PROJECT_ID', ''),
|
||
'location': get_secret('GOOGLE_CLOUD_LOCATION', 'us-central1'),
|
||
'memory_bank_enabled': get_secret('VERTEX_MEMORY_BANK_ENABLED', 'TRUE'),
|
||
'service_account_key': get_secret('GOOGLE_SERVICE_ACCOUNT_KEY', '')
|
||
}
|
||
|
||
def get_cloudflare_config() -> Dict[str, str]:
|
||
"""
|
||
获取Cloudflare配置
|
||
|
||
Returns:
|
||
Cloudflare配置字典
|
||
"""
|
||
return {
|
||
# 敏感信息从Doppler获取
|
||
'account_id': get_secret('CLOUDFLARE_ACCOUNT_ID', ''),
|
||
'api_token': get_secret('CLOUDFLARE_API_TOKEN', ''),
|
||
|
||
# 非敏感配置，明文写在代码里
|
||
'vectorize_index': 'autorag-shy-cherry-f1fb',
|
||
'embed_model': '@cf/baai/bge-m3',
|
||
'autorag_domain': 'autorag.seekkey.tech'
|
||
}
|
||
|
||
def get_database_config() -> Dict[str, str]:
|
||
"""
|
||
获取数据库配置
|
||
|
||
Returns:
|
||
数据库配置字典
|
||
"""
|
||
return {
|
||
'postgres_url': get_secret('POSTGRES_URL', ''),
|
||
'mongodb_url': get_secret('MONGODB_URL', ''),
|
||
'zilliz_url': get_secret('ZILLIZ_URL', ''),
|
||
'zilliz_token': get_secret('ZILLIZ_TOKEN', '')
|
||
}
|
||
|
||
def validate_config(mode: str = "hybrid") -> bool:
|
||
"""
|
||
验证必要的配置是否存在
|
||
|
||
Args:
|
||
mode: 验证模式 ("openrouter", "google_adk", "hybrid")
|
||
|
||
Returns:
|
||
配置是否有效
|
||
"""
|
||
print(f"🔧 当前模式: {mode}")
|
||
|
||
required_keys = []
|
||
|
||
# 模式特定配置
|
||
if mode == "openrouter":
|
||
required_keys.extend(['RAPIDAPI_KEY', 'OPENROUTER_API_KEY_1'])
|
||
# 验证 OpenRouter 配置
|
||
openrouter_key = get_secret('OPENROUTER_API_KEY_1', '')
|
||
if not openrouter_key:
|
||
print("❌ OpenRouter API Key 未配置")
|
||
return False
|
||
print("✅ OpenRouter 配置验证通过")
|
||
|
||
elif mode == "google_adk":
|
||
genai_config = get_google_genai_config()
|
||
use_vertex = genai_config.get('use_vertex_ai', 'FALSE').upper() == 'TRUE'
|
||
|
||
if not use_vertex:
|
||
required_keys.extend(['GOOGLE_API_KEY'])
|
||
# 验证 Google ADK 配置
|
||
google_key = get_secret('GOOGLE_API_KEY', '')
|
||
if not google_key:
|
||
print("❌ Google API Key 未配置")
|
||
print("请访问 https://aistudio.google.com/ 获取 API 密钥")
|
||
print("然后运行: doppler secrets set GOOGLE_API_KEY=your_key")
|
||
return False
|
||
print(f"✅ Google ADK 配置验证通过 (密钥长度: {len(google_key)} 字符)")
|
||
else:
|
||
print("✅ Google ADK (Vertex AI) 配置验证通过")
|
||
|
||
# 显示 Google GenAI 配置
|
||
print(f"📱 Google GenAI 配置:")
|
||
if not use_vertex:
|
||
print(f" - API Key: 已配置")
|
||
print(f" - Use Vertex AI: {genai_config.get('use_vertex_ai', False)}")
|
||
if genai_config.get('project_id'):
|
||
print(f" - Project ID: {genai_config['project_id']}")
|
||
if genai_config.get('location'):
|
||
print(f" - Location: {genai_config['location']}")
|
||
|
||
else: # hybrid mode
|
||
required_keys.extend(['RAPIDAPI_KEY'])
|
||
# 检查至少有一个AI API密钥
|
||
ai_keys = ['OPENROUTER_API_KEY_1', 'GOOGLE_API_KEY']
|
||
if not any(os.getenv(key) for key in ai_keys):
|
||
print("❌ 需要至少配置一个AI API密钥:")
|
||
print(" - OPENROUTER_API_KEY_1 (OpenRouter模式)")
|
||
print(" - GOOGLE_API_KEY (Google ADK模式)")
|
||
return False
|
||
|
||
# 验证混合模式配置
|
||
openrouter_key = get_secret('OPENROUTER_API_KEY_1', '')
|
||
google_key = get_secret('GOOGLE_API_KEY', '')
|
||
|
||
available_services = []
|
||
if openrouter_key:
|
||
available_services.append("OpenRouter")
|
||
if google_key:
|
||
available_services.append("Google ADK")
|
||
|
||
print(f"✅ 混合模式配置验证通过，可用服务: {', '.join(available_services)}")
|
||
|
||
missing_keys = []
|
||
for key in required_keys:
|
||
if not os.getenv(key):
|
||
missing_keys.append(key)
|
||
|
||
if missing_keys:
|
||
print(f"❌ 缺少必要的配置: {', '.join(missing_keys)}")
|
||
print("请确保已正确配置Doppler或环境变量")
|
||
return False
|
||
|
||
# 显示配置状态
|
||
print("✅ 配置验证通过")
|
||
print(f"📋 当前模式: {mode}")
|
||
|
||
# 显示可用的AI服务
|
||
ai_services = []
|
||
if os.getenv('OPENROUTER_API_KEY_1'):
|
||
ai_services.append("OpenRouter")
|
||
if os.getenv('GOOGLE_API_KEY'):
|
||
ai_services.append("Google ADK")
|
||
|
||
if ai_services:
|
||
print(f"🤖 可用AI服务: {', '.join(ai_services)}")
|
||
|
||
return True
|
||
|
||
if __name__ == "__main__":
|
||
# 配置验证脚本
|
||
print("🔧 验证配置...")
|
||
validate_config()
|