# -*- coding: utf-8 -*- """ AI 提供商配置模型 支持多种 AI 提供商：OpenAI、DeepSeek、Claude、通义千问、智谱GLM、自定义 """ from datetime import datetime from .base import db, get_beijing_now from .user import User import json # AI 提供商常量 class AIProviderType: """AI 提供商类型""" OPENAI = 'openai' DEEPSEEK = 'deepseek' CLAUDE = 'claude' QWEN = 'qwen' # 通义千问 ZHIPU = 'zhipu' # 智谱GLM GROQ = 'groq' # Groq（免费高速推理） TOGETHER = 'together' # Together AI OPENROUTER = 'openrouter' # OpenRouter（聚合多个提供商） GEMINI = 'gemini' # Google Gemini MISTRAL = 'mistral' # Mistral AI SILICONFLOW = 'siliconflow' # SiliconFlow（国内免费平台） ANYROUTER = 'anyrouter' # AnyRouter（Claude Code 共享平台，仅支持 CLI） AGENTROUTER = 'agentrouter' # AgentRouter（Claude Code 共享平台，仅支持 CLI） AUGMENT = 'augment' # Augment CLI CUSTOM = 'custom' # 自定义（兼容 OpenAI 格式） CUSTOM_OPENAI = 'custom_openai' # 自定义 OpenAI 兼容 API CUSTOM_ANTHROPIC = 'custom_anthropic' # 自定义 Anthropic 兼容 API # CLI 模式提供商（使用独立配置界面，不在自定义 API 下拉列表中显示） CLI_PROVIDERS = ['anyrouter', 'agentrouter', 'augment'] # 自定义 API 类型（用于区分 OpenAI 和 Anthropic 兼容格式） CUSTOM_API_TYPES = ['custom', 'custom_openai', 'custom_anthropic'] @staticmethod def get_all(): """获取所有提供商类型""" return [ AIProviderType.OPENAI, AIProviderType.DEEPSEEK, AIProviderType.CLAUDE, AIProviderType.QWEN, AIProviderType.ZHIPU, AIProviderType.GROQ, AIProviderType.TOGETHER, AIProviderType.OPENROUTER, AIProviderType.GEMINI, AIProviderType.MISTRAL, AIProviderType.SILICONFLOW, AIProviderType.ANYROUTER, AIProviderType.AGENTROUTER, AIProviderType.AUGMENT, AIProviderType.CUSTOM, AIProviderType.CUSTOM_OPENAI, AIProviderType.CUSTOM_ANTHROPIC, ] @staticmethod def get_api_providers(): """获取仅支持 API 的提供商（排除 CLI 模式提供商） CLI 模式提供商（AnyRouter、AgentRouter、Augment）使用独立的配置界面， 不应出现在"自定义 API"的提供商下拉列表中。 """ return [p for p in AIProviderType.get_all() if p not in AIProviderType.CLI_PROVIDERS] @staticmethod def get_display_name(provider_type): """获取提供商显示名称""" names = { AIProviderType.OPENAI: 'OpenAI', AIProviderType.DEEPSEEK: 'DeepSeek', AIProviderType.CLAUDE: 'Claude (Anthropic)', AIProviderType.QWEN: '通义千问', AIProviderType.ZHIPU: '智谱GLM', AIProviderType.GROQ: 'Groq (免费高速)', AIProviderType.TOGETHER: 'Together AI', AIProviderType.OPENROUTER: 'OpenRouter', AIProviderType.GEMINI: 'Google Gemini', AIProviderType.MISTRAL: 'Mistral AI', AIProviderType.SILICONFLOW: 'SiliconFlow (硅基流动)', AIProviderType.ANYROUTER: 'AnyRouter', AIProviderType.AGENTROUTER: 'AgentRouter', AIProviderType.AUGMENT: 'Augment CLI', AIProviderType.CUSTOM: '自定义', AIProviderType.CUSTOM_OPENAI: '自定义 (OpenAI 兼容)', AIProviderType.CUSTOM_ANTHROPIC: '自定义 (Anthropic 兼容)', } return names.get(provider_type, provider_type) @staticmethod def get_default_base_url(provider_type): """获取提供商默认 Base URL""" urls = { AIProviderType.OPENAI: 'https://api.openai.com/v1', AIProviderType.DEEPSEEK: 'https://api.deepseek.com/v1', AIProviderType.CLAUDE: 'https://api.anthropic.com/v1', AIProviderType.QWEN: 'https://dashscope.aliyuncs.com/compatible-mode/v1', AIProviderType.ZHIPU: 'https://open.bigmodel.cn/api/paas/v4', AIProviderType.GROQ: 'https://api.groq.com/openai/v1', AIProviderType.TOGETHER: 'https://api.together.xyz/v1', AIProviderType.OPENROUTER: 'https://openrouter.ai/api/v1', AIProviderType.GEMINI: 'https://generativelanguage.googleapis.com/v1beta/openai', AIProviderType.MISTRAL: 'https://api.mistral.ai/v1', AIProviderType.SILICONFLOW: 'https://api.siliconflow.cn/v1', AIProviderType.ANYROUTER: 'https://anyrouter.top/v1', AIProviderType.AGENTROUTER: 'https://agentrouter.org/', AIProviderType.AUGMENT: '', # Augment 使用 Session Token，无需 Base URL AIProviderType.CUSTOM: '', AIProviderType.CUSTOM_OPENAI: '', # 用户自定义 AIProviderType.CUSTOM_ANTHROPIC: '', # 用户自定义 } return urls.get(provider_type, '') @staticmethod def get_default_models(provider_type): """获取提供商默认模型列表""" models = { AIProviderType.OPENAI: [ # GPT-5 系列（最新） 'gpt-5', 'gpt-5-mini', 'gpt-5.1', 'gpt-5.1-mini', 'gpt-5.1-nano', # GPT-4.1 系列 'gpt-4.1', 'gpt-4.1-mini', 'gpt-4.1-nano', # GPT-4o 系列 'gpt-4o', 'gpt-4o-mini', # 推理模型系列（o 系列） 'o3', 'o3-mini', 'o4-mini', 'o1', 'o1-mini', 'o1-pro', # 经典模型 'gpt-4-turbo', 'gpt-4', 'gpt-3.5-turbo', ], AIProviderType.DEEPSEEK: [ # DeepSeek-V3 系列 'deepseek-chat', # V3/V3.1 对话模型 'deepseek-reasoner', # R1 推理模型 'deepseek-coder', # 代码模型 ], AIProviderType.CLAUDE: [ # Claude 4 系列（最新） 'claude-opus-4.5', 'claude-opus-4', 'claude-sonnet-4', # Claude 3.5 系列 'claude-sonnet-4-5-20250514', 'claude-3-5-sonnet-20241022', 'claude-3-5-sonnet-20240620', 'claude-3-5-haiku-20241022', # Claude 3 系列 'claude-3-opus-20240229', 'claude-3-sonnet-20240229', 'claude-3-haiku-20240307', ], AIProviderType.QWEN: [ # Qwen3 系列（最新） 'qwen3-max', 'qwen3-plus', 'qwen3-turbo', # Qwen2.5 系列 'qwen2.5-max', 'qwen2.5-plus', 'qwen2.5-turbo', # 通义千问经典系列 'qwen-max', 'qwen-max-latest', 'qwen-plus', 'qwen-plus-latest', 'qwen-turbo', 'qwen-turbo-latest', 'qwen-long', # 长文本模型 'qwen-coder-plus', # 代码模型 ], AIProviderType.ZHIPU: [ # GLM-4 系列（最新） 'glm-4.6', 'glm-4.5v', # 最新旗舰模型 'glm-4-plus', 'glm-4-long', 'glm-4', 'glm-4-flash', 'glm-4-air', 'glm-4-airx', 'glm-4-9b', # GLM-4V 视觉模型 'glm-4v', 'glm-4v-plus', # 代码模型 'codegeex-4', ], AIProviderType.GROQ: [ # Llama 系列（免费） 'llama-3.3-70b-versatile', 'llama-3.1-70b-versatile', 'llama-3.1-8b-instant', 'llama3-70b-8192', 'llama3-8b-8192', # Mixtral 系列（免费） 'mixtral-8x7b-32768', # Gemma 系列（免费） 'gemma2-9b-it', ], AIProviderType.TOGETHER: [ # Llama 系列 'meta-llama/Llama-3.3-70B-Instruct-Turbo', 'meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo', 'meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo', # Mixtral 系列 'mistralai/Mixtral-8x7B-Instruct-v0.1', 'mistralai/Mistral-7B-Instruct-v0.3', # Qwen 系列 'Qwen/Qwen2.5-72B-Instruct-Turbo', 'Qwen/Qwen2.5-7B-Instruct-Turbo', # DeepSeek 系列 'deepseek-ai/DeepSeek-R1-Distill-Llama-70B', ], AIProviderType.OPENROUTER: [ # 免费模型（带 :free 后缀） 'kwaipilot/kat-coder-pro:free', # 快手 Kat Coder Pro（免费） 'meta-llama/llama-3.2-3b-instruct:free', 'google/gemma-2-9b-it:free', 'mistralai/mistral-7b-instruct:free', 'qwen/qwen-2-7b-instruct:free', # 付费模型 'openai/gpt-4o', 'anthropic/claude-3.5-sonnet', 'google/gemini-pro-1.5', ], AIProviderType.GEMINI: [ # Gemini 2.0 系列（最新） 'gemini-2.0-flash-exp', 'gemini-2.0-flash', # Gemini 1.5 系列 'gemini-1.5-pro', 'gemini-1.5-flash', 'gemini-1.5-flash-8b', # Gemini 1.0 系列 'gemini-pro', ], AIProviderType.MISTRAL: [ # 最新模型 'mistral-large-latest', 'mistral-medium-latest', 'mistral-small-latest', # 开源模型 'open-mixtral-8x22b', 'open-mixtral-8x7b', 'open-mistral-7b', # 代码模型 'codestral-latest', ], AIProviderType.SILICONFLOW: [ # DeepSeek 系列（免费额度） 'deepseek-ai/DeepSeek-V3', 'deepseek-ai/DeepSeek-R1', # Qwen 系列 'Qwen/Qwen2.5-72B-Instruct', 'Qwen/Qwen2.5-32B-Instruct', 'Qwen/Qwen2.5-7B-Instruct', # Llama 系列 'meta-llama/Meta-Llama-3.1-70B-Instruct', 'meta-llama/Meta-Llama-3.1-8B-Instruct', # GLM 系列 'THUDM/glm-4-9b-chat', ], AIProviderType.ANYROUTER: [ # Claude 4 系列（最新） 'claude-opus-4-5-20251101', 'claude-opus-4-1-20250805', 'claude-opus-4-20250514', 'claude-sonnet-4-5-20250929', 'claude-sonnet-4-20250514', 'claude-haiku-4-5-20251001', # Claude 3.7 系列 'claude-3-7-sonnet-20250219', # Claude 3.5 系列 'claude-3-5-sonnet-20241022', 'claude-3-5-haiku-20241022', # GPT 系列 'gpt-5-codex', 'gpt-4o', 'gpt-4o-mini', 'gpt-4-turbo', 'gpt-3.5-turbo', # Gemini 系列 'gemini-2.5-pro', 'gemini-2.0-flash', 'gemini-1.5-pro', 'gemini-1.5-flash', # DeepSeek 系列 'deepseek-chat', 'deepseek-coder', # Llama 系列 'llama-3.1-70b-instruct', 'llama-3.1-8b-instruct', # Mistral 系列 'mistral-large', 'mixtral-8x7b', ], AIProviderType.AGENTROUTER: [ # Claude Sonnet 4.5 系列（推荐） 'claude-sonnet-4-5-20250929', # 最新推荐 'claude-sonnet-4-5-20250514', # Claude Haiku 系列（轻量级快速） 'claude-haiku-4-5-20251001', 'claude-3-5-haiku-20241022', ], AIProviderType.AUGMENT: ['augment-default'], # Augment 使用 Session Token AIProviderType.CUSTOM: [], AIProviderType.CUSTOM_OPENAI: [], # 用户自定义模型 AIProviderType.CUSTOM_ANTHROPIC: [], # 用户自定义模型 } return models.get(provider_type, []) class AIProviderConfig(db.Model): """AI 提供商配置模型 存储用户或系统的 AI 提供商配置，包括 API Key（加密存储） """ __tablename__ = 'ai_provider_configs' id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True) # NULL 表示系统默认配置 provider_type = db.Column(db.String(50), nullable=False) # openai, deepseek, claude, qwen, zhipu, custom name = db.Column(db.String(100)) # 配置名称，如 "我的 OpenAI" api_key_encrypted = db.Column(db.Text, nullable=False) # 加密存储的 API Key base_url = db.Column(db.String(255)) # API Base URL model = db.Column(db.String(100)) # 默认模型 is_active = db.Column(db.Boolean, default=True) # 是否启用 is_default = db.Column(db.Boolean, default=False) # 是否为用户/系统的默认配置 last_used = db.Column(db.DateTime) # 最后使用时间 last_test = db.Column(db.DateTime) # 最后测试时间 test_status = db.Column(db.String(20)) # 测试状态: success, failed, pending test_message = db.Column(db.Text) # 测试结果消息 created_at = db.Column(db.DateTime, default=get_beijing_now) updated_at = db.Column(db.DateTime, default=get_beijing_now, onupdate=get_beijing_now) # 关联 # 明确指定使用 user_id 作为外键，避免与 User.preferred_system_ai_config_id 产生歧义 user = db.relationship( 'User', foreign_keys=[user_id], backref=db.backref('ai_configs', lazy='dynamic') ) def to_dict(self, include_key_preview=False): """转换为字典 Args: include_key_preview: 是否包含 API Key 预览（前4位...后4位） """ data = { 'id': self.id, 'user_id': self.user_id, 'provider_type': self.provider_type, 'provider_name': AIProviderType.get_display_name(self.provider_type), 'name': self.name, 'base_url': self.base_url, 'model': self.model, 'is_active': self.is_active, 'is_default': self.is_default, 'last_used': self.last_used.isoformat() if self.last_used else None, 'last_test': self.last_test.isoformat() if self.last_test else None, 'test_status': self.test_status, 'test_message': self.test_message, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, } if include_key_preview: key = self.get_api_key() if key and len(key) > 8: data['api_key_preview'] = f"{key[:4]}...{key[-4:]}" elif key: data['api_key_preview'] = '***' else: data['api_key_preview'] = None return data def set_api_key(self, api_key): """设置 API Key（加密存储） 注意：这里使用简单的 base64 编码，生产环境建议使用更安全的加密方式 """ import base64 if api_key: # 简单加密：base64 + 反转 encoded = base64.b64encode(api_key.encode('utf-8')).decode('utf-8') self.api_key_encrypted = encoded[::-1] # 反转字符串 else: self.api_key_encrypted = '' def get_api_key(self): """获取解密后的 API Key""" import base64 if self.api_key_encrypted: try: # 解密：反转 + base64 解码 reversed_str = self.api_key_encrypted[::-1] return base64.b64decode(reversed_str.encode('utf-8')).decode('utf-8') except Exception: return None return None def update_test_status(self, status, message=None): """更新测试状态""" self.test_status = status self.test_message = message self.last_test = get_beijing_now() db.session.commit() def mark_used(self): """标记为已使用""" self.last_used = get_beijing_now() db.session.commit() @staticmethod def get_user_config(user_id, provider_type=None): """获取用户的 AI 配置 Args: user_id: 用户 ID provider_type: 可选，指定提供商类型 Returns: AIProviderConfig 或 None """ query = AIProviderConfig.query.filter_by(user_id=user_id, is_active=True) if provider_type: query = query.filter_by(provider_type=provider_type) return query.filter_by(is_default=True).first() or query.first() @staticmethod def get_system_default(): """获取系统默认 AI 配置""" return AIProviderConfig.query.filter_by( user_id=None, is_active=True, is_default=True ).first() @staticmethod def get_effective_config(user_id): """获取用户有效的 AI 配置 优先级规则（用户必须显式选择）： 1. 如果系统强制使用系统 AI（force_system_ai=True）： 1) 如果用户选择了系统池中的某条配置且仍然有效 → 使用该配置 2) 否则 → 返回 None（用户需要从系统池中选择一个） 2. 如果不强制系统 AI： 1) 如果允许用户自定义 AI（allow_user_custom_ai=True）， 且用户有个人默认配置 → 使用用户个人默认配置 2) 否则，如果用户在系统池中选择了某条配置且仍然有效 → 使用该配置 3) 否则 → 返回 None（用户需要选择系统AI或个人配置） 注意：不再使用"系统全局默认"概念，用户必须显式选择。 Args: user_id: 用户 ID Returns: AIProviderConfig 或 None（表示用户需要选择配置） """ from .user import User # 加载系统级策略 force_system = SystemAIConfig.force_system_ai() allow_custom = SystemAIConfig.allow_user_custom_ai() # 加载用户及其系统 AI 偏好（如果存在） user = User.query.get(user_id) if user_id is not None else None preferred_system_id = ( user.preferred_system_ai_config_id if user is not None else None ) def _get_preferred_system_config(): """根据用户偏好获取一条有效的系统级配置""" if not preferred_system_id: return None return AIProviderConfig.query.filter_by( id=preferred_system_id, user_id=None, is_active=True, ).first() # 1) 系统强制使用系统 AI：忽略个人配置，必须从系统池中选择 if force_system: preferred_system_config = _get_preferred_system_config() if preferred_system_config: return preferred_system_config # 用户需要从系统池中选择一个 return None # 2) 如果用户显式选择了系统AI，优先使用系统AI（即使允许自定义） # 这表示用户想要使用系统AI而不是个人配置 preferred_system_config = _get_preferred_system_config() if preferred_system_config: return preferred_system_config # 3) 不强制系统 AI：使用个人配置（如果允许） if allow_custom: user_config = AIProviderConfig.get_user_config(user_id) if user_config: return user_config # 4) 用户需要选择配置（系统AI或个人配置） return None @staticmethod def set_user_default(user_id, config_id): """设置用户默认配置""" # 先取消所有默认 AIProviderConfig.query.filter_by(user_id=user_id).update({'is_default': False}) # 设置新默认 config = AIProviderConfig.query.get(config_id) if config and config.user_id == user_id: config.is_default = True db.session.commit() return True return False @staticmethod def create_config(user_id, provider_type, api_key, model=None, base_url=None, name=None, is_default=False): """创建 AI 配置 Args: user_id: 用户 ID（None 表示系统配置） provider_type: 提供商类型 api_key: API Key model: 模型名称 base_url: API Base URL name: 配置名称 is_default: 是否设为默认 Returns: AIProviderConfig """ config = AIProviderConfig( user_id=user_id, provider_type=provider_type, name=name or AIProviderType.get_display_name(provider_type), base_url=base_url or AIProviderType.get_default_base_url(provider_type), model=model or (AIProviderType.get_default_models(provider_type) or [None])[0], is_active=True, is_default=is_default, test_status='pending' ) config.set_api_key(api_key) # 如果设为默认，先取消其他默认 if is_default: AIProviderConfig.query.filter_by(user_id=user_id).update({'is_default': False}) db.session.add(config) db.session.commit() return config # ============================================================ # 兼容层：SystemAIConfig（使用 SystemConfig 的 category='ai'） # ============================================================ class SystemAIConfig: """系统 AI 配置兼容层（使用 system_configs 表的 category='ai'）""" # 注意：这不是 db.Model，而是兼容层类 @staticmethod def get_config(key, default=None): """获取配置值""" # 延迟导入避免循环依赖 from .base import db from .config import SystemConfig config = SystemConfig.query.filter_by(key=key, category='ai', enabled=True).first() if config: return config.get_value() return default @staticmethod def set_config(key, value, description=None): """设置配置值""" # 延迟导入避免循环依赖 from .base import db from .config import SystemConfig config = SystemConfig.query.filter_by(key=key, category='ai').first() if config: config.set_value(value) if description: config.description = description else: config = SystemConfig( key=key, value='', description=description, category='ai' ) config.set_value(value) db.session.add(config) db.session.commit() return config @staticmethod def get_enabled_providers(): """获取启用的 AI 提供商列表""" enabled = SystemAIConfig.get_config('enabled_providers', AIProviderType.get_all()) # 如果数据库里被清空成空列表/空字符串等，自动回退到默认全量列表 if not enabled: return AIProviderType.get_all() return enabled @staticmethod def set_enabled_providers(providers): """设置启用的 AI 提供商列表""" return SystemAIConfig.set_config( 'enabled_providers', providers, '启用的 AI 提供商列表' ) @staticmethod def is_provider_enabled(provider_type): """检查提供商是否启用""" enabled = SystemAIConfig.get_enabled_providers() return provider_type in enabled @staticmethod def allow_user_custom_ai(): """是否允许用户使用自定义 AI""" return SystemAIConfig.get_config('allow_user_custom_ai', True) @staticmethod def set_allow_user_custom_ai(allow): """设置是否允许用户使用自定义 AI""" return SystemAIConfig.set_config( 'allow_user_custom_ai', allow, '是否允许用户使用自定义 AI 配置' ) @staticmethod def force_system_ai(): """是否强制所有用户使用系统 AI""" return SystemAIConfig.get_config('force_system_ai', False) @staticmethod def set_force_system_ai(force): """设置是否强制所有用户使用系统 AI""" return SystemAIConfig.set_config( 'force_system_ai', force, '是否强制所有用户使用系统 AI（禁用用户自定义）' ) @staticmethod def get_all_provider_info(): """获取所有 API 提供商信息（用于前端"自定义 API"下拉列表） 注意：现在包含 AnyRouter 和 AgentRouter，但 Augment 仍然使用独立配置界面。 """ enabled = SystemAIConfig.get_enabled_providers() providers = [] # 包含所有 API 提供商，以及 anyrouter 和 agentrouter（但不包括 augment） api_providers = AIProviderType.get_api_providers() # 添加 anyrouter 和 agentrouter additional_providers = ['anyrouter', 'agentrouter'] all_providers = api_providers + additional_providers for provider_type in all_providers: providers.append({ 'type': provider_type, 'name': AIProviderType.get_display_name(provider_type), 'base_url': AIProviderType.get_default_base_url(provider_type), 'models': AIProviderType.get_default_models(provider_type), 'enabled': provider_type in enabled, }) return providers # ============================================================ # CLI 模式 Token 模型（Augment、AnyRouter、AgentRouter） # ============================================================ # ============================================================ # 兼容层：AugmentSessionToken（使用新的 CLIToken 表） # ============================================================ class AugmentSessionToken: """Augment Session Token 兼容层（使用 ai_provider_configs 表）""" # 注意：这不是 db.Model，而是兼容层类 # 为了保持向后兼容，我们使用类属性模拟 query @staticmethod def get_global_token(): """获取全局Token""" config = AIProviderConfig.query.filter_by( user_id=None, provider_type='augment', is_active=True, is_default=True ).first() if config: return config.get_api_key() return None @staticmethod def set_global_token(token_str, name='全局Token', admin_user_id=None): """设置全局Token""" # 先将所有全局配置设为非默认 AIProviderConfig.query.filter_by( user_id=None, provider_type='augment' ).update({'is_default': False}) # 查找或创建全局配置 config = AIProviderConfig.query.filter_by( user_id=None, provider_type='augment', is_active=True ).first() if config: config.set_api_key(token_str) config.is_default = True config.name = name or '全局 Augment Token' else: config = AIProviderConfig( user_id=None, provider_type='augment', name=name or '全局 Augment Token', model='augment-cli', base_url='', is_active=True, is_default=True ) config.set_api_key(token_str) db.session.add(config) db.session.commit() return config @staticmethod def delete_global_token(): """删除全局Token""" configs = AIProviderConfig.query.filter_by( user_id=None, provider_type='augment' ).all() for config in configs: config.set_api_key('') config.is_active = False db.session.commit() return len(configs) > 0 @classmethod def query(cls): """查询接口（兼容）""" # 返回一个模拟的查询对象，用于兼容旧代码 class MockQuery: def filter_by(self, **kwargs): return self def first(self): return None return MockQuery() # ============================================================ # 兼容层：AnyRouterToken（使用新的 CLIToken 表） # ============================================================ class AnyRouterToken: """AnyRouter Token 兼容层（使用 ai_provider_configs 表）""" @staticmethod def get_user_token(user_id): """获取用户的 Token""" config = AIProviderConfig.query.filter_by( user_id=user_id, provider_type='anyrouter', is_active=True ).first() if config: return config.get_api_key() return None @staticmethod def get_global_token(): """获取全局 Token""" config = AIProviderConfig.query.filter_by( user_id=None, provider_type='anyrouter', is_active=True, is_default=True ).first() if config: return config.get_api_key() return None @staticmethod def set_user_token(user_id, token_str, name='用户 Token'): """设置用户的 Token""" config = AIProviderConfig.query.filter_by( user_id=user_id, provider_type='anyrouter', is_active=True ).first() if config: config.set_api_key(token_str) config.name = name or 'AnyRouter Token' else: config = AIProviderConfig( user_id=user_id, provider_type='anyrouter', name=name or 'AnyRouter Token', model='claude-sonnet-4', base_url='https://anyrouter.top', is_active=True, is_default=False ) config.set_api_key(token_str) db.session.add(config) db.session.commit() return config @staticmethod def delete_user_token(user_id): """删除用户的 Token""" config = AIProviderConfig.query.filter_by( user_id=user_id, provider_type='anyrouter', is_active=True ).first() if config: config.set_api_key('') config.is_active = False db.session.commit() return True return False @classmethod def query(cls): """查询接口（兼容）""" class MockQuery: def filter_by(self, **kwargs): return self def first(self): return None return MockQuery() # ============================================================ # 兼容层：AgentRouterToken（使用新的 CLIToken 表） # ============================================================ class AgentRouterToken: """AgentRouter Token 兼容层（使用 ai_provider_configs 表） API 文档: https://docs.agentrouter.org/start.html """ @staticmethod def get_user_token(user_id): """获取用户的 Token""" config = AIProviderConfig.query.filter_by( user_id=user_id, provider_type='agentrouter', is_active=True ).first() if config: return config.get_api_key() return None @staticmethod def get_global_token(): """获取全局 Token""" config = AIProviderConfig.query.filter_by( user_id=None, provider_type='agentrouter', is_active=True, is_default=True ).first() if config: return config.get_api_key() return None @staticmethod def set_user_token(user_id, token_str, name='用户 Token'): """设置用户的 Token""" config = AIProviderConfig.query.filter_by( user_id=user_id, provider_type='agentrouter', is_active=True ).first() if config: config.set_api_key(token_str) config.name = name or 'AgentRouter Token' else: config = AIProviderConfig( user_id=user_id, provider_type='agentrouter', name=name or 'AgentRouter Token', model='claude-sonnet-4', base_url='https://agentrouter.org/', is_active=True, is_default=False ) config.set_api_key(token_str) db.session.add(config) db.session.commit() return config @staticmethod def delete_user_token(user_id): """删除用户的 Token""" config = AIProviderConfig.query.filter_by( user_id=user_id, provider_type='agentrouter', is_active=True ).first() if config: config.set_api_key('') config.is_active = False db.session.commit() return True return False @classmethod def query(cls): """查询接口（兼容）""" class MockQuery: def filter_by(self, **kwargs): return self def first(self): return None return MockQuery()