公司git
fec3426bad
feat(config): 添加API备用地址和密钥配置 - 新增 AI_BASE_URL 和 AI_BACKUP_BASE_URL 配置项 - 为试用、高级、视频和Gemini Flash等API添加备用密钥 - 使用f-string格式化API端点URL以支持动态基地址 - 统一API端点构建方式，提高配置灵活性 feat(services): 实现API故障自动切换机制 - 在聊天生成服务中集成多线路候选和故障转移逻辑 - 重构图像和视频生成服务以支持备用API线路 - 实现智能路由，根据状态码自动切换到备用线路 - 增强错误处理和日志记录功能 feat(utils): 新增API线路管理和切换工具函数 - 实现 get_backup_api_url() 函数用于URL备用地址转换 - 创建 get_backup_api_key() 函数管理备用密钥映射 - 开发 get_api_candidates() 函数生成主备线路候选列表 - 添加 should_switch_to_backup() 函数判断是否需要切换线路 refactor(services): 优化视频生成任务的API密钥配置 - 将视频生成任务独立使用VIDEO_KEY而非通用密钥 - 确保视频服务使用专门的API密钥进行身份验证 ```
64 lines
1.8 KiB
Python
64 lines
1.8 KiB
Python
from urllib.parse import quote, urlparse, urlunparse
|
||
from config import Config
|
||
|
||
def get_proxied_url(url):
|
||
"""
|
||
如果启用了代理，则返回经过代理包装的 URL。
|
||
否则返回原 URL。
|
||
"""
|
||
if Config.USE_PROXY and url:
|
||
return f"{Config.PROXY_URL}{quote(url)}"
|
||
return url
|
||
|
||
def get_backup_api_url(url):
|
||
"""
|
||
将已有接口地址切换到备用 AI Base URL，并保留原始路径。
|
||
"""
|
||
if not url or not Config.AI_BACKUP_BASE_URL:
|
||
return url
|
||
|
||
source = urlparse(url)
|
||
backup = urlparse(Config.AI_BACKUP_BASE_URL)
|
||
if not backup.scheme or not backup.netloc:
|
||
return url
|
||
|
||
return urlunparse((
|
||
backup.scheme,
|
||
backup.netloc,
|
||
source.path,
|
||
source.params,
|
||
source.query,
|
||
source.fragment,
|
||
))
|
||
|
||
def get_backup_api_key(api_key):
|
||
"""
|
||
返回内置主 Key 对应的备用 Key；自定义 Key 默认沿用原值。
|
||
"""
|
||
fallback_map = {
|
||
Config.TRIAL_KEY: Config.TRIAL_BACKUP_KEY,
|
||
Config.PREMIUM_KEY: Config.PREMIUM_BACKUP_KEY,
|
||
Config.VIDEO_KEY: Config.VIDEO_BACKUP_KEY,
|
||
Config.GEMINI_FLASH_IMAGE_PREMIUM_KEY: Config.GEMINI_FLASH_IMAGE_PREMIUM_BACKUP_KEY,
|
||
}
|
||
return fallback_map.get(api_key, api_key)
|
||
|
||
def get_api_candidates(url, api_key):
|
||
"""
|
||
生成主线路与备用线路候选。
|
||
"""
|
||
candidates = [{"url": url, "api_key": api_key, "label": "primary"}]
|
||
backup_url = get_backup_api_url(url)
|
||
backup_key = get_backup_api_key(api_key)
|
||
|
||
if backup_url and backup_url != url:
|
||
candidates.append({"url": backup_url, "api_key": backup_key, "label": "backup"})
|
||
|
||
return candidates
|
||
|
||
def should_switch_to_backup(status_code):
|
||
"""
|
||
遇到鉴权、限流或服务端异常时，尝试切换备用线路。
|
||
"""
|
||
return status_code in (401, 403, 408, 429) or status_code >= 500
|