app.py

# 启动事件
@app.on_event("startup")
async def startup_event():
    """服务启动时初始化"""
    logger.info(f"OpenAI API代理服务已启动，可以接受请求")
    logger.info(f"支持多token轮询，请在Authorization头中使用英文逗号分隔多个token")
    logger.info(f"服务地址: http://127.0.0.1:7860")
    logger.info(f"OpenAI API格式请求示例: POST http://127.0.0.1:7860import json
import time
import asyncio
import uvicorn
from fastapi import FastAPI, Request, HTTPException, Header, Depends
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any, Union
import requests
from datetime import datetime
import logging
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("openai-proxy")

# 创建FastAPI应用
app = FastAPI(
    title="OpenAI API Proxy",
    description="将OpenAI API请求代理到DeepSider API",
    version="1.0.0"
)

# 增加日志输出级别
if os.getenv("DEBUG", "false").lower() == "true":
    logging.getLogger("openai-proxy").setLevel(logging.DEBUG)

# 添加CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 配置
DEEPSIDER_API_BASE = "https://api.chargpt.ai/api/v1"
TOKEN_INDEX = 0

# 模型映射表
MODEL_MAPPING = {
    "gpt-3.5-turbo": "anthropic/claude-3.5-sonnet",
    "gpt-4": "anthropic/claude-3.7-sonnet",
    "gpt-4o": "openai/gpt-4o",
    "gpt-4-turbo": "openai/gpt-4o",
    "gpt-4o-mini": "openai/gpt-4o-mini",
    "claude-3-sonnet-20240229": "anthropic/claude-3.5-sonnet",
    "claude-3-opus-20240229": "anthropic/claude-3.7-sonnet",
    "claude-3.5-sonnet": "anthropic/claude-3.5-sonnet",
    "claude-3.7-sonnet": "anthropic/claude-3.7-sonnet",
}

def get_headers(api_key):
    global TOKEN_INDEX
    # 检查是否包含多个token（用逗号分隔）
    tokens = api_key.split(',')
    
    if len(tokens) > 0:
        # 轮询选择token
        current_token = tokens[TOKEN_INDEX % len(tokens)]
        TOKEN_INDEX = (TOKEN_INDEX + 1) % len(tokens)
    else:
        current_token = api_key
    
    return {
        "accept": "application/json",
        "content-type": "application/json",
        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
        "authorization": f"Bearer {current_token.strip()}",
        "i-version": "1.1.64"
    }

# OpenAI API请求模型
class ChatMessage(BaseModel):
    role: str
    content: str
    name: Optional[str] = None

class ChatCompletionRequest(BaseModel):
    model: str
    messages: List[ChatMessage]
    temperature: Optional[float] = 1.0
    top_p: Optional[float] = 1.0
    n: Optional[int] = 1
    stream: Optional[bool] = False
    stop: Optional[Union[List[str], str]] = None
    max_tokens: Optional[int] = None
    presence_penalty: Optional[float] = 0
    frequency_penalty: Optional[float] = 0
    user: Optional[str] = None
    
# 账户余额查询函数
async def check_account_balance(api_key, token_index=None):
    """检查账户余额信息"""
    tokens = api_key.split(',')
    
    # 如果提供了token_index并且有效，则使用指定的token
    if token_index is not None and len(tokens) > token_index:
        current_token = tokens[token_index].strip()
    else:
        # 否则使用第一个token
        current_token = tokens[0].strip() if tokens else api_key
        
    headers = {
        "accept": "*/*",
        "content-type": "application/json",
        "authorization": f"Bearer {current_token}"
    }
    
    try:
        # 获取账户余额信息
        response = requests.get(
            f"{DEEPSIDER_API_BASE.replace('/v2', '')}/quota/retrieve",
            headers=headers
        )
        
        if response.status_code == 200:
            data = response.json()
            if data.get('code') == 0:
                quota_list = data.get('data', {}).get('list', [])
                
                # 解析余额信息
                quota_info = {}
                for item in quota_list:
                    item_type = item.get('type', '')
                    available = item.get('available', 0)
                    
                    quota_info[item_type] = {
                        "total": item.get('total', 0),
                        "available": available,
                        "title": item.get('title', '')
                    }
                
                return True, quota_info
        
        return False, {}
            
    except Exception as e:
        logger.warning(f"检查账户余额出错：{str(e)}")
        return False, {}

# 工具函数
def verify_api_key(api_key: str = Header(..., alias="Authorization")):
    """验证API密钥"""
    if not api_key.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid API key format")
    return api_key.replace("Bearer ", "")

def map_openai_to_deepsider_model(model: str) -> str:
    """将OpenAI模型名称映射到DeepSider模型名称"""
    return MODEL_MAPPING.get(model, "anthropic/claude-3.7-sonnet")

def format_messages_for_deepsider(messages: List[ChatMessage]) -> str:
    """格式化消息列表为DeepSider API所需的提示格式"""
    # 直接合并所有消息内容，无需特殊格式化
    combined_prompt = ""
    for msg in messages:
        if msg.role == "system":
            combined_prompt = msg.content + "\n\n" + combined_prompt
        else:
            combined_prompt += msg.content + "\n\n"
    
    return combined_prompt.strip()

async def generate_openai_response(full_response: str, request_id: str, model: str) -> Dict:
    """生成符合OpenAI API响应格式的完整响应"""
    timestamp = int(time.time())
    return {
        "id": f"chatcmpl-{request_id}",
        "object": "chat.completion",
        "created": timestamp,
        "model": model,
        "choices": [
            {
                "index": 0,
                "message": {
                    "role": "assistant",
                    "content": full_response
                },
                "finish_reason": "stop"
            }
        ],
        "usage": {
            "prompt_tokens": 0,  # 无法准确计算
            "completion_tokens": 0,  # 无法准确计算
            "total_tokens": 0  # 无法准确计算
        }
    }

async def stream_openai_response(response, request_id: str, model: str, api_key, token_index):
    """流式返回OpenAI API格式的响应"""
    timestamp = int(time.time())
    
    try:
        # 直接传递原始响应
        for chunk in response.iter_lines():
            if chunk:
                yield chunk.decode('utf-8') + "\n"
                
    except Exception as e:
        logger.error(f"流式响应处理出错: {str(e)}")
        
        # 返回错误信息
        error_chunk = {
            "id": f"chatcmpl-{request_id}",
            "object": "chat.completion.chunk",
            "created": timestamp,
            "model": model,
            "choices": [
                {
                    "index": 0,
                    "delta": {
                        "content": f"\n\n[处理响应时出错: {str(e)}]"
                    },
                    "finish_reason": "stop"
                }
            ]
        }
        yield f"data: {json.dumps(error_chunk)}\n\n"
        yield "data: [DONE]\n\n"

# 路由定义
@app.get("/")
async def root():
    """返回简单的HTML页面，展示使用说明"""
    return {
        "message": "OpenAI API Proxy服务已启动 连接至DeepSider API",
        "usage": {
            "模型列表": "GET /v1/models",
            "聊天完成": "POST /v1/chat/completions",
            "账户余额": "GET /admin/balance (需要X-Admin-Key头)"
        },
        "说明": "请在Authorization头中使用Bearer token格式，支持使用英文逗号分隔多个token实现轮询"
    }

@app.get("/v1/models")
async def list_models(api_key: str = Depends(verify_api_key)):
    """列出可用的模型"""
    models = []
    for openai_model, _ in MODEL_MAPPING.items():
        models.append({
            "id": openai_model,
            "object": "model",
            "created": int(time.time()),
            "owned_by": "openai-proxy"
        })
    
    return {
        "object": "list",
        "data": models
    }

@app.post("/v1/chat/completions")
async def create_chat_completion(
    request: Request,
    api_key: str = Depends(verify_api_key)
):
    """创建聊天完成API - 支持普通请求和流式请求"""
    # 解析请求体
    body = await request.json()
    chat_request = ChatCompletionRequest(**body)
    
    # 生成唯一请求ID
    request_id = datetime.now().strftime("%Y%m%d%H%M%S") + str(time.time_ns())[-6:]
    
    # 映射模型
    deepsider_model = map_openai_to_deepsider_model(chat_request.model)
    
    # 准备DeepSider API所需的提示
    prompt = format_messages_for_deepsider(chat_request.messages)
    
    # 准备请求体
    payload = {
        "model": deepsider_model,
        "prompt": prompt,
        "stream": chat_request.stream
    }
    
    # 获取请求头（包含选择的token）
    headers = get_headers(api_key)
    # 获取当前使用的token
    tokens = api_key.split(',')
    current_token_index = (TOKEN_INDEX - 1) % len(tokens) if len(tokens) > 0 else 0
    
    try:
        # 记录请求信息
        logger.info(f"发送请求到DeepSider API - 模型: {deepsider_model}, Prompt长度: {len(prompt)}")
        logger.debug(f"请求正文: {json.dumps(payload)}")
        
        # 发送请求到DeepSider API
        response = requests.post(
            f"{DEEPSIDER_API_BASE}/chat/completions",
            headers=headers,
            json=payload,
            stream=chat_request.stream,
            timeout=60  # 设置60秒超时
        )
        
        # 检查响应状态
        if response.status_code != 200:
            error_msg = f"DeepSider API请求失败: {response.status_code}"
            try:
                error_data = response.json()
                error_msg += f" - {error_data.get('message', '')}"
            except:
                try:
                    error_msg += f" - {response.text[:200]}"
                except:
                    pass
                
            logger.error(error_msg)
            return {
                "error": {
                    "message": error_msg,
                    "type": "api_error",
                    "code": response.status_code
                }
            }
        
        # 处理流式或非流式响应
        if chat_request.stream:
            # 返回流式响应
            return StreamingResponse(
                stream_openai_response(response, request_id, chat_request.model, api_key, current_token_index),
                media_type="text/event-stream"
            )
        else:
            try:
                # 非流式请求，直接返回响应
                json_response = response.json()
                # 记录响应，有助于调试
                logger.debug(f"非流式响应: {json.dumps(json_response)}")
                return json_response
                
            except Exception as e:
                logger.exception(f"非流式响应处理出错: {str(e)}")
                return {
                    "error": {
                        "message": f"处理响应时出错: {str(e)}",
                        "type": "processing_error",
                        "code": "internal_error"
                    }
                }
            
    except HTTPException:
        raise
    except Exception as e:
        logger.exception("处理请求时出错")
        raise HTTPException(status_code=500, detail=f"内部服务器错误: {str(e)}")

@app.get("/admin/balance")
async def get_account_balance(request: Request, admin_key: str = Header(None, alias="X-Admin-Key")):
    """查看账户余额"""
    # 简单的管理密钥检查
    expected_admin_key = os.getenv("ADMIN_KEY", "admin")
    if not admin_key or admin_key != expected_admin_key:
        raise HTTPException(status_code=403, detail="Unauthorized")
    
    # 从请求头中获取API密钥
    auth_header = request.headers.get("Authorization", "")
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
    
    api_key = auth_header.replace("Bearer ", "")
    tokens = api_key.split(',')
    
    result = {}
    
    # 获取所有token的余额信息
    for i, token in enumerate(tokens):
        token_display = f"token_{i+1}"
        success, quota_info = await check_account_balance(api_key, i)
        
        if success:
            result[token_display] = {
                "status": "success",
                "quota": quota_info
            }
        else:
            result[token_display] = {
                "status": "error",
                "message": "无法获取账户余额信息"
            }
    
    return result

# 错误处理器
@app.exception_handler(404)
async def not_found_handler(request, exc):
    return {
        "error": {
            "message": f"未找到资源: {request.url.path}",
            "type": "not_found_error",
            "code": "not_found"
        }
    }

@app.exception_handler(500)
async def server_error_handler(request, exc):
    return {
        "error": {
            "message": f"服务器内部错误: {str(exc)}",
            "type": "server_error",
            "code": "internal_server_error"
        }
    }

# 启动事件
@app.on_event("startup")
async def startup_event():
    """服务启动时初始化"""
    logger.info(f"OpenAI API代理服务已启动，可以接受请求")
    logger.info(f"支持多token轮询，请在Authorization头中使用英文逗号分隔多个token")
    logger.info(f"服务地址: http://127.0.0.1:7860")
    logger.info(f"OpenAI API格式请求示例: POST http://127.0.0.1:7860/v1/chat/completions")
    logger.info(f"可用模型查询: GET http://127.0.0.1:7860/v1/models")

# 主程序
if __name__ == "__main__":
    # 启动服务器
    port = int(os.getenv("PORT", "7860"))
    logger.info(f"启动OpenAI API代理服务 端口: {port}")
    uvicorn.run(app, host="0.0.0.0", port=port)