""" Web API for capability detection """ import asyncio import json from typing import Dict, List, Optional, Any from fastapi import FastAPI, HTTPException, Request, Depends from fastapi.staticfiles import StaticFiles from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse from starlette.middleware.sessions import SessionMiddleware from pydantic import BaseModel from core.capability_detector import CapabilityDetectorFactory from core.openai_detector import OpenAICapabilityDetector from core.anthropic_detector import AnthropicCapabilityDetector from core.gemini_detector import GeminiCapabilityDetector from src.utils.config import ConfigManager, ChannelConfig from src.utils.logger import setup_logger from src.utils.auth import auth_manager from src.utils.security import mask_api_key from api.conversion_api import router as conversion_router from api.unified_api import router as unified_router app = FastAPI(title="AI API统一转换代理系统", version="1.0.0") logger = setup_logger("web_api") # 添加会话中间件 import os import secrets from datetime import datetime, timedelta def get_session_secret_key(): """获取会话密钥，如果未设置则生成并持久化随机密钥""" session_key = os.getenv("SESSION_SECRET_KEY") if not session_key: # 尝试从数据库获取已保存的密钥 from src.utils.database import db_manager stored_key = db_manager.get_config("session_secret_key") if stored_key: session_key = stored_key logger.info("Using stored session secret key from database") else: # 生成新的随机密钥并存储 session_key = secrets.token_hex(32) db_manager.set_config("session_secret_key", session_key) logger.warning("SESSION_SECRET_KEY not set, generated and stored new random key.") logger.info("Sessions will now persist across server restarts.") else: logger.info("Using SESSION_SECRET_KEY from environment variables") return session_key secret_key = get_session_secret_key() app.add_middleware(SessionMiddleware, secret_key=secret_key) # 包含API路由 app.include_router(conversion_router, prefix="/api") # 管理API app.include_router(unified_router) # 统一转换API（直接挂载到根路径） # 注册检测器 CapabilityDetectorFactory.register("openai", OpenAICapabilityDetector) CapabilityDetectorFactory.register("anthropic", AnthropicCapabilityDetector) CapabilityDetectorFactory.register("gemini", GeminiCapabilityDetector) # 静态文件服务 app.mount("/static", StaticFiles(directory="static"), name="static") # 全局状态存储 class CapabilityResultModel(BaseModel): """能力检测结果模型""" capability: str status: str details: Optional[Dict[str, Any]] = None error: Optional[str] = None response_time: Optional[float] = None class ChannelCapabilitiesModel(BaseModel): """渠道能力模型""" provider: str base_url: str models: List[str] capabilities: Dict[str, CapabilityResultModel] detection_time: str class LoginRequest(BaseModel): """登录请求""" password: str class LoginResponse(BaseModel): """登录响应""" success: bool session_token: Optional[str] = None message: str def get_session_user(request: Request): """获取会话用户，验证是否已登录""" if not request.session.get("authenticated"): raise HTTPException(status_code=401, detail="未登录") return True def get_optional_session_user(request: Request): """可选的会话用户获取，用于页面渲染""" return request.session.get("authenticated", False) @app.get("/login", response_class=HTMLResponse) async def login_page(): """返回登录页面""" return """ AI API统一转换代理系统 - 登录 """ @app.get("/") async def index(): """重定向到登录页面""" return RedirectResponse(url="/login", status_code=302) @app.get("/dashboard", response_class=HTMLResponse) async def dashboard(request: Request): """管理仪表板 - 需要认证""" # 检查会话认证状态 if not request.session.get("authenticated"): # 认证失败，重定向到登录页 return RedirectResponse(url="/login", status_code=302) return """ AI API统一转换代理系统

AI API FORMAT CONVERSION

渠道管理

添加新渠道

渠道名称:

提供商:

Base URL:

API密钥:

自定义Key: 用户调用API时使用此key进行身份验证

超时时间(秒):

重试次数:

已配置渠道

正在加载渠道列表...

API使用说明

核心功能：AI API格式统一转换代理系统，支持OpenAI、Anthropic、Gemini三种格式的相互转换

智能路由：根据请求路径自动识别API格式，根据自定义key识别目标渠道，自动进行格式转换

工作原理

格式识别：根据请求路径自动识别源API格式（OpenAI/Anthropic/Gemini）
渠道路由：根据自定义key查找对应的目标渠道配置
格式转换：自动将请求格式转换为目标渠道的API格式
请求转发：调用真实的AI服务API
响应转换：将响应格式转换回源格式并返回给客户端

支持的转换

OpenAI ↔ Anthropic ↔ Gemini（任意格式间相互转换）
流式和非流式请求
函数调用、视觉理解、结构化输出等高级功能
自动模型映射和参数适配

能力检测

AI提供商:

API基础URL:

API密钥:

超时时间(秒):

指定模型 (必填):

必须指定要检测的模型

要检测的能力:

基础聊天流式输出系统消息函数调用结构化输出视觉理解

""" @app.post("/api/login") async def login(login_request: LoginRequest, request: Request): """管理员登录""" # 验证密码 if auth_manager.verify_admin_password(login_request.password): # 设置会话 request.session["authenticated"] = True request.session["login_time"] = datetime.now().isoformat() return JSONResponse( content={ "success": True, "message": "登录成功" } ) else: return JSONResponse( content={ "success": False, "message": "密码错误" }, status_code=401 ) @app.post("/api/logout") async def logout(request: Request): """管理员注销""" # 清除会话 request.session.clear() return JSONResponse( content={ "success": True, "message": "注销成功" } ) @app.get("/api/providers") async def get_providers(): """获取支持的提供商""" return { "providers": [ { "id": "openai", "name": "OpenAI", "description": "OpenAI (GPT-4o, GPT-3.5-turbo 等)", "default_url": "https://api.openai.com/v1" }, { "id": "anthropic", "name": "Anthropic", "description": "Anthropic (Claude-3, Claude-3.5 等)", "default_url": "https://api.anthropic.com" }, { "id": "gemini", "name": "Google Gemini", "description": "Google Gemini (Gemini-1.5, Gemini-2.0 等)", "default_url": "https://generativelanguage.googleapis.com/v1beta" } ] } @app.get("/api/channels") async def get_channels(_: bool = Depends(get_session_user)): """获取所有渠道""" try: from src.channels.channel_manager import ChannelManager manager = ChannelManager() channels = manager.get_all_channels() return { "success": True, "channels": [ { "id": channel.id, "name": channel.name, "provider": channel.provider, "base_url": channel.base_url, "api_key": mask_api_key(channel.api_key), "custom_key": channel.custom_key, "timeout": channel.timeout, "max_retries": channel.max_retries, "enabled": channel.enabled, "models_mapping": channel.models_mapping, "created_at": channel.created_at, "updated_at": channel.updated_at } for channel in channels ] } except Exception as e: logger.error(f"Failed to get channels: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/channels") async def create_channel( channel_data: dict, _: bool = Depends(get_session_user) ): """创建新渠道""" try: from src.channels.channel_manager import ChannelManager manager = ChannelManager() channel_id = manager.add_channel( name=channel_data["name"], provider=channel_data["provider"], base_url=channel_data["base_url"], api_key=channel_data["api_key"], custom_key=channel_data["custom_key"], timeout=channel_data.get("timeout", 30), max_retries=channel_data.get("max_retries", 3), models_mapping=channel_data.get("models_mapping") ) return { "success": True, "channel_id": channel_id, "message": "渠道创建成功" } except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"Failed to create channel: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.put("/api/channels/{channel_id}") async def update_channel( channel_id: str, channel_data: dict, _: bool = Depends(get_session_user) ): """更新渠道""" try: from src.channels.channel_manager import ChannelManager manager = ChannelManager() # 确保空的api_key不会被传递，使用None而不是空字符串 api_key = channel_data.get("api_key") if api_key is not None and api_key.strip() == "": api_key = None success = manager.update_channel( channel_id=channel_id, name=channel_data.get("name"), base_url=channel_data.get("base_url"), api_key=api_key, custom_key=channel_data.get("custom_key"), timeout=channel_data.get("timeout"), max_retries=channel_data.get("max_retries"), enabled=channel_data.get("enabled"), models_mapping=channel_data.get("models_mapping") ) if success: return { "success": True, "message": "渠道更新成功" } else: raise HTTPException(status_code=404, detail="渠道不存在") except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: logger.error(f"Failed to update channel: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.delete("/api/channels/{channel_id}") async def delete_channel( channel_id: str, _: bool = Depends(get_session_user) ): """删除渠道""" try: from src.channels.channel_manager import ChannelManager manager = ChannelManager() success = manager.delete_channel(channel_id) if success: return { "success": True, "message": "渠道删除成功" } else: raise HTTPException(status_code=404, detail="渠道不存在") except Exception as e: logger.error(f"Failed to delete channel: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/capabilities") async def get_capabilities(): """获取所有可检测的能力""" config_manager = ConfigManager() capabilities = config_manager.get_all_capabilities() return { "capabilities": [ { "id": name, "name": name, "description": config.description } for name, config in capabilities.items() ] } @app.post("/api/fetch_models") async def fetch_models(request: dict): """获取模型列表""" try: provider = request.get("provider") base_url = request.get("base_url") api_key = request.get("api_key") if not all([provider, base_url, api_key]): raise HTTPException(status_code=400, detail="缺少必要参数") # 创建配置 config = ChannelConfig( provider=provider, base_url=base_url, api_key=api_key, timeout=30 ) # 创建检测器 detector = CapabilityDetectorFactory.create(config) # 获取模型列表 models = await detector.detect_models() return {"models": models} except Exception as e: logger.error(f"Failed to fetch models: {e}") raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)