Upload 8 files

Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+# Install dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application
+COPY . .
+
+# Create data directory
+RUN mkdir -p /app/data
+
+# Expose port
+EXPOSE 7860
+
+# Run application
+CMD ["python", "app.py"]

account_manager.py

@@ -0,0 +1,247 @@
+"""
+账号池管理器 - 支持 OAuth Token 的增删改查和自动刷新
+"""
+import json
+import os
+import httpx
+from typing import List, Optional
+from datetime import datetime, timedelta
+from models import Account, OAuthToken, AccountStats
+
+# 数据文件路径 (HF Spaces 持久化目录)
+DATA_DIR = os.environ.get("DATA_DIR", "./data")
+ACCOUNTS_FILE = os.path.join(DATA_DIR, "accounts.json")
+CONFIG_FILE = os.path.join(DATA_DIR, "config.json")
+
+# Google OAuth 配置 (Antigravity 使用的 Client ID)
+OAUTH_CLIENT_ID = os.environ.get(
+    "OAUTH_CLIENT_ID", 
+    "595848968694-r5ng3t6qb9elhe1u1h1hqgq4j2r3hgvk.apps.googleusercontent.com"
+)
+# 默认使用 AI Studio 的公开 Client Secret
+OAUTH_CLIENT_SECRET = os.environ.get(
+    "OAUTH_CLIENT_SECRET", 
+    "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"
+)
+OAUTH_TOKEN_URL = "https://oauth2.googleapis.com/token"
+
+
+class AccountManager:
+    """账号管理器 - 支持 OAuth Token"""
+    
+    def __init__(self):
+        self._accounts: dict[str, Account] = {}
+        self._current_index = 0
+        self._ensure_data_dir()
+        self._load_accounts()
+    
+    def _ensure_data_dir(self):
+        """确保数据目录存在"""
+        os.makedirs(DATA_DIR, exist_ok=True)
+    
+    def _load_accounts(self):
+        """从文件加载账号"""
+        if os.path.exists(ACCOUNTS_FILE):
+            try:
+                with open(ACCOUNTS_FILE, "r", encoding="utf-8") as f:
+                    data = json.load(f)
+                    for item in data:
+                        account = Account(**item)
+                        self._accounts[account.id] = account
+            except Exception as e:
+                print(f"加载账号失败: {e}")
+    
+    def _save_accounts(self):
+        """保存账号到文件"""
+        try:
+            data = [acc.model_dump(mode="json") for acc in self._accounts.values()]
+            with open(ACCOUNTS_FILE, "w", encoding="utf-8") as f:
+                json.dump(data, f, ensure_ascii=False, indent=2, default=str)
+        except Exception as e:
+            print(f"保存账号失败: {e}")
+    
+    def add_account(
+        self, 
+        email: str, 
+        access_token: str, 
+        refresh_token: str,
+        expires_in: int = 3600,
+        project_id: Optional[str] = None
+    ) -> Account:
+        """添加新账号"""
+        now = int(datetime.now().timestamp())
+        token = OAuthToken(
+            access_token=access_token,
+            refresh_token=refresh_token,
+            expires_in=expires_in,
+            expiry_timestamp=now + expires_in,
+            project_id=project_id
+        )
+        account = Account(email=email, token=token)
+        self._accounts[account.id] = account
+        self._save_accounts()
+        return account
+    
+    def remove_account(self, account_id: str) -> bool:
+        """删除账号"""
+        if account_id in self._accounts:
+            del self._accounts[account_id]
+            self._save_accounts()
+            return True
+        return False
+    
+    def get_account(self, account_id: str) -> Optional[Account]:
+        """获取单个账号"""
+        return self._accounts.get(account_id)
+    
+    def get_all_accounts(self) -> List[Account]:
+        """获取所有账号"""
+        return list(self._accounts.values())
+    
+    def get_available_accounts(self) -> List[Account]:
+        """获取所有可用账号"""
+        return [acc for acc in self._accounts.values() if acc.is_available()]
+    
+    async def get_next_token(self) -> Optional[Account]:
+        """
+        获取下一个可用的 Token（轮询机制）
+        自动刷新过期的 Token
+        """
+        available = self.get_available_accounts()
+        if not available:
+            return None
+        
+        # Round Robin
+        self._current_index = self._current_index % len(available)
+        account = available[self._current_index]
+        self._current_index += 1
+        
+        # 检查并刷新过期 Token
+        if account.is_token_expired():
+            print(f"账号 {account.email} 的 token 即将过期，正在刷新...")
+            try:
+                await self._refresh_token(account)
+            except Exception as e:
+                print(f"刷新 token 失败: {e}")
+                # 继续使用可能过期的 token，让 API 返回错误
+        
+        return account
+    
+    async def _refresh_token(self, account: Account):
+        """刷新 OAuth Token"""
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                OAUTH_TOKEN_URL,
+                data={
+                    "client_id": OAUTH_CLIENT_ID,
+                    "client_secret": OAUTH_CLIENT_SECRET,
+                    "refresh_token": account.token.refresh_token,
+                    "grant_type": "refresh_token"
+                }
+            )
+            
+            if response.status_code != 200:
+                raise Exception(f"刷新失败: {response.text}")
+            
+            data = response.json()
+            now = int(datetime.now().timestamp())
+            
+            account.token.access_token = data["access_token"]
+            account.token.expires_in = data.get("expires_in", 3600)
+            account.token.expiry_timestamp = now + account.token.expires_in
+            
+            self._save_accounts()
+            print(f"Token 刷新成功！有效期: {account.token.expires_in} 秒")
+    
+    def update_account_stats(self, account_id: str, success: bool, error: str = None):
+        """更新账号统计信息"""
+        account = self._accounts.get(account_id)
+        if account:
+            account.total_requests += 1
+            account.last_used = datetime.now()
+            if success:
+                account.successful_requests += 1
+                account.last_error = None
+            else:
+                account.failed_requests += 1
+                account.last_error = error
+            self._save_accounts()
+    
+    def set_account_cooldown(self, account_id: str, duration_seconds: int):
+        """设置账号冷却时间"""
+        account = self._accounts.get(account_id)
+        if account:
+            account.cooldown_until = datetime.now() + timedelta(seconds=duration_seconds)
+            self._save_accounts()
+    
+    def toggle_account(self, account_id: str) -> bool:
+        """切换账号启用状态"""
+        account = self._accounts.get(account_id)
+        if account:
+            account.enabled = not account.enabled
+            self._save_accounts()
+            return account.enabled
+        return False
+    
+    def get_stats(self) -> AccountStats:
+        """获取统计汇总"""
+        accounts = list(self._accounts.values())
+        total_requests = sum(acc.total_requests for acc in accounts)
+        successful = sum(acc.successful_requests for acc in accounts)
+        
+        return AccountStats(
+            total_accounts=len(accounts),
+            available_accounts=len([a for a in accounts if a.is_available()]),
+            total_requests=total_requests,
+            success_rate=successful / total_requests if total_requests > 0 else 0.0
+        )
+
+
+class ConfigManager:
+    """配置管理器 - 管理 API Key 等可变配置"""
+    
+    def __init__(self):
+        self._config = {
+            "api_key": "sk-antigravity"
+        }
+        self._ensure_data_dir()
+        self._load_config()
+    
+    def _ensure_data_dir(self):
+        """确保数据目录存在"""
+        os.makedirs(DATA_DIR, exist_ok=True)
+    
+    def _load_config(self):
+        """从文件加载配置"""
+        if os.path.exists(CONFIG_FILE):
+            try:
+                with open(CONFIG_FILE, "r", encoding="utf-8") as f:
+                    self._config.update(json.load(f))
+            except Exception as e:
+                print(f"加载配置失败: {e}")
+    
+    def _save_config(self):
+        """保存配置到文件"""
+        try:
+            with open(CONFIG_FILE, "w", encoding="utf-8") as f:
+                json.dump(self._config, f, ensure_ascii=False, indent=2)
+        except Exception as e:
+            print(f"保存配置失败: {e}")
+    
+    def get_api_key(self) -> str:
+        """获取 API Key"""
+        return self._config.get("api_key", "sk-antigravity")
+    
+    def set_api_key(self, api_key: str) -> bool:
+        """设置 API Key"""
+        if not api_key or len(api_key.strip()) == 0:
+            return False
+        self._config["api_key"] = api_key.strip()
+        self._save_config()
+        return True
+
+
+# 全局单例
+account_manager = AccountManager()
+config_manager = ConfigManager()
+

app.py

@@ -0,0 +1,467 @@
+"""
+API 反代服务 - 主应用程序
+Gradio 管理界面 + FastAPI API 端点
+"""
+import os
+import json
+import gradio as gr
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import StreamingResponse, JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+from contextlib import asynccontextmanager
+
+from models import OpenAIChatRequest, OpenAIMessage, SUPPORTED_MODELS, ServiceConfig
+from account_manager import account_manager, config_manager
+from load_balancer import load_balancer
+from proxy_service import stream_chat_completion, chat_completion
+
+# ============ 服务配置 ============
+config = ServiceConfig()
+
+# 从环境变量获取管理员凭据
+ADMIN_USERNAME = os.environ.get("ADMIN_USERNAME", "admin")
+ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "antigravity")
+
+
+# ============ FastAPI 应用 ============
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """应用生命周期管理"""
+    print("🚀 API 反代服务启动中...")
+    yield
+    print("👋 API 反代服务关闭")
+
+
+app = FastAPI(
+    title="Antigravity API Proxy",
+    description="OpenAI 兼容的 API 反代服务",
+    version="1.0.0",
+    lifespan=lifespan
+)
+
+# CORS 中间件
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# ============ API 端点 ============
+
+@app.get("/v1/models")
+async def list_models():
+    """列出支持的模型"""
+    return {
+        "object": "list",
+        "data": [
+            {
+                "id": model["id"],
+                "object": "model",
+                "owned_by": "antigravity",
+                "permission": []
+            }
+            for model in SUPPORTED_MODELS
+        ]
+    }
+
+
+@app.post("/v1/chat/completions")
+async def chat_completions(request: Request):
+    """
+    OpenAI 兼容的 Chat Completion API
+    """
+    # 验证 API Key
+    auth_header = request.headers.get("Authorization", "")
+    if not auth_header.startswith("Bearer "):
+        raise HTTPException(status_code=401, detail="Missing API key")
+    
+    # 验证 API Key 是否匹配
+    provided_key = auth_header[7:]  # 去掉 "Bearer " 前缀
+    expected_key = config_manager.get_api_key()
+    if provided_key != expected_key:
+        raise HTTPException(status_code=401, detail="Invalid API key")
+    
+    # 获取请求体
+    body = await request.json()
+    
+    try:
+        openai_request = OpenAIChatRequest(
+            model=body.get("model", "gemini-2.5-flash"),
+            messages=[OpenAIMessage(**msg) for msg in body.get("messages", [])],
+            temperature=body.get("temperature", 1.0),
+            top_p=body.get("top_p", 0.95),
+            max_tokens=body.get("max_tokens", 8192),
+            stream=body.get("stream", False)
+        )
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=f"Invalid request: {e}")
+    
+    # 获取可用账号
+    account = await load_balancer.get_next_account()
+    if not account:
+        raise HTTPException(status_code=503, detail="No available accounts")
+    
+    # 处理请求
+    if openai_request.stream:
+        return StreamingResponse(
+            stream_chat_completion(openai_request, account),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+            }
+        )
+    else:
+        result = await chat_completion(openai_request, account)
+        if "error" in result:
+            raise HTTPException(status_code=500, detail=result["error"])
+        return JSONResponse(content=result)
+
+
+@app.get("/health")
+async def health_check():
+    """健康检查"""
+    stats = account_manager.get_stats()
+    return {
+        "status": "healthy",
+        "accounts": stats.total_accounts,
+        "available": stats.available_accounts
+    }
+
+
+@app.post("/admin/import-accounts")
+async def import_accounts(request: Request):
+    """
+    批量导入账号 (需要管理员密码)
+    请求体格式:
+    {
+        "password": "admin_password",
+        "accounts": [
+            {"email": "xxx@gmail.com", "refresh_token": "1//xxx..."},
+            ...
+        ]
+    }
+    """
+    body = await request.json()
+    
+    # 验证管理员密码
+    if body.get("password") != ADMIN_PASSWORD:
+        raise HTTPException(status_code=401, detail="Invalid admin password")
+    
+    accounts_data = body.get("accounts", [])
+    if not accounts_data:
+        raise HTTPException(status_code=400, detail="No accounts provided")
+    
+    results = []
+    for acc in accounts_data:
+        email = acc.get("email")
+        refresh_token = acc.get("refresh_token")
+        
+        if not email or not refresh_token:
+            results.append({"email": email, "status": "error", "message": "Missing email or refresh_token"})
+            continue
+        
+        try:
+            account = account_manager.add_account(
+                email=email,
+                access_token="pending",  # 将在首次使用时自动刷新
+                refresh_token=refresh_token,
+                expires_in=0  # 立即过期，强制刷新
+            )
+            results.append({"email": email, "status": "success", "id": account.id})
+        except Exception as e:
+            results.append({"email": email, "status": "error", "message": str(e)})
+    
+    return {
+        "imported": len([r for r in results if r["status"] == "success"]),
+        "failed": len([r for r in results if r["status"] == "error"]),
+        "results": results
+    }
+
+
+# ============ Gradio 管理界面 ============
+
+def get_accounts_table():
+    """获取账号列表表格数据"""
+    accounts = account_manager.get_all_accounts()
+    if not accounts:
+        return [["暂无账号", "-", "-", "-", "-", "-"]]
+    
+    return [
+        [
+            acc.id,
+            acc.email,
+            "✅ 正常" if acc.is_available() else "❌ 冷却中",
+            str(acc.total_requests),
+            f"{acc.successful_requests / acc.total_requests * 100:.1f}%" if acc.total_requests > 0 else "-",
+            acc.last_used.strftime("%Y-%m-%d %H:%M") if acc.last_used else "-"
+        ]
+        for acc in accounts
+    ]
+
+
+def add_account(email: str, access_token: str, refresh_token: str, project_id: str):
+    """添加账号"""
+    if not email or not access_token or not refresh_token:
+        return "❌ 请填写完整信息", get_accounts_table()
+    
+    try:
+        account = account_manager.add_account(
+            email=email,
+            access_token=access_token,
+            refresh_token=refresh_token,
+            project_id=project_id if project_id else None
+        )
+        return f"✅ 账号 {email} 添加成功！", get_accounts_table()
+    except Exception as e:
+        return f"❌ 添加失败: {e}", get_accounts_table()
+
+
+def delete_account(email_to_delete: str):
+    """删除账号"""
+    if not email_to_delete or email_to_delete == "暂无账号":
+        return "❌ 请选择要删除的账号", get_accounts_table()
+    
+    # 查找账号 ID
+    accounts = account_manager.get_all_accounts()
+    target_account = None
+    for acc in accounts:
+        if acc.email == email_to_delete:
+            target_account = acc
+            break
+    
+    if not target_account:
+        return f"❌ 账号 {email_to_delete} 不存在", get_accounts_table()
+    
+    try:
+        if account_manager.remove_account(target_account.id):
+            return f"✅ 账号 {email_to_delete} 已删除", get_accounts_table()
+        else:
+            return f"❌ 删除账号 {email_to_delete} 失败", get_accounts_table()
+    except Exception as e:
+        return f"❌ 删除失败: {e}", get_accounts_table()
+
+
+def get_account_emails():
+    """获取所有账号邮箱列表 (用于更新 Dropdown)"""
+    accounts = account_manager.get_all_accounts()
+    if not accounts:
+        return gr.update(choices=["暂无账号"], value=None)
+    emails = [acc.email for acc in accounts]
+    return gr.update(choices=emails, value=None)
+
+
+def _get_account_emails_list():
+    """获取账号邮箱列表 (用于初始化)"""
+    accounts = account_manager.get_all_accounts()
+    if not accounts:
+        return ["暂无账号"]
+    return [acc.email for acc in accounts]
+
+
+def update_api_key(new_api_key: str):
+    """更新 API Key"""
+    if not new_api_key or len(new_api_key.strip()) < 3:
+        return "❌ API Key 不能为空且至少需要 3 个字符", config_manager.get_api_key()
+    
+    if config_manager.set_api_key(new_api_key.strip()):
+        return f"✅ API Key 已更新为: {new_api_key.strip()}", new_api_key.strip()
+    else:
+        return "❌ 更新 API Key 失败", config_manager.get_api_key()
+
+
+def get_service_info():
+    """获取服务信息"""
+    stats = account_manager.get_stats()
+    current_api_key = config_manager.get_api_key()
+    
+    # 获取当前 Space URL
+    space_url = os.environ.get("SPACE_HOST", "localhost:7860")
+    if not space_url.startswith("http"):
+        space_url = f"https://{space_url}"
+    
+    return f"""
+## 🔗 API 配置
+
+**Base URL:** `{space_url}/v1`
+
+**API Key:** `{current_api_key}`
+
+**支持的模型:**
+- `gemini-2.5-pro` / `gemini-2.5-flash` / `gemini-2.5-flash-lite`
+- `gemini-3-flash` / `gemini-3-pro` / `gemini-3-pro-high`
+- `claude-sonnet-4-5-thinking` / `claude-sonnet-4-5` / `claude-opus-4-5`
+
+---
+
+## 📊 服务状态
+
+| 指标 | 值 |
+|------|-----|
+| 总账号数 | {stats.total_accounts} |
+| 可用账号 | {stats.available_accounts} |
+| 总请求数 | {stats.total_requests} |
+| 成功率 | {stats.success_rate * 100:.1f}% |
+
+---
+
+## 🛠️ 使用方法
+
+### cURL 示例
+```bash
+curl {space_url}/v1/chat/completions \\
+  -H "Content-Type: application/json" \\
+  -H "Authorization: Bearer {current_api_key}" \\
+  -d '{{
+    "model": "gemini-2.5-flash",
+    "messages": [{{"role": "user", "content": "Hello"}}],
+    "stream": true
+  }}'
+```
+
+### NextChat / Cherry Studio 配置
+1. API 类型: OpenAI
+2. Base URL: `{space_url}/v1`
+3. API Key: `{current_api_key}`
+"""
+
+
+# 创建 Gradio 界面
+with gr.Blocks(
+    title="Antigravity API Proxy",
+    theme=gr.themes.Soft(
+        primary_hue="purple",
+        secondary_hue="blue",
+    ),
+    css="""
+    .container { max-width: 1200px; margin: auto; }
+    .logo { font-size: 2em; font-weight: bold; }
+    .danger-btn { background-color: #dc3545 !important; }
+    """
+) as demo:
+    gr.Markdown(
+        """
+        # 🚀 Antigravity API Proxy
+        ### 将 Gemini/Claude 转换为 OpenAI 兼容 API
+        """,
+        elem_classes="logo"
+    )
+    
+    with gr.Tabs():
+        # 仪表盘
+        with gr.Tab("📊 仪表盘"):
+            service_info = gr.Markdown(get_service_info)
+            refresh_btn = gr.Button("🔄 刷新状态")
+            refresh_btn.click(fn=get_service_info, outputs=service_info)
+        
+        # 账号管理
+        with gr.Tab("👤 账号管理"):
+            gr.Markdown("### 添加新账号")
+            gr.Markdown("> 需要从 Antigravity 桌面应用导出账号信息，或手动获取 OAuth Token")
+            
+            with gr.Row():
+                email_input = gr.Textbox(label="邮箱", placeholder="example@gmail.com")
+                project_id_input = gr.Textbox(label="Project ID (选填)", placeholder="可留空")
+            
+            access_token_input = gr.Textbox(
+                label="Access Token", 
+                placeholder="ya29.xxx... 或填 pending",
+                lines=2
+            )
+            refresh_token_input = gr.Textbox(
+                label="Refresh Token",
+                placeholder="1//xxx...",
+                lines=2
+            )
+            
+            add_btn = gr.Button("➕ 添加账号", variant="primary")
+            result_text = gr.Textbox(label="结果", interactive=False)
+            
+            gr.Markdown("---")
+            gr.Markdown("### 账号列表")
+            accounts_table = gr.Dataframe(
+                headers=["ID", "邮箱", "状态", "请求数", "成功率", "最后使用"],
+                value=get_accounts_table,
+                interactive=False
+            )
+            
+            gr.Markdown("### 删除账号")
+            with gr.Row():
+                delete_email_dropdown = gr.Dropdown(
+                    label="选择要删除的账号",
+                    choices=_get_account_emails_list(),
+                    interactive=True
+                )
+                delete_btn = gr.Button("🗑️ 删除账号", variant="stop")
+            delete_result = gr.Textbox(label="删除结果", interactive=False)
+            
+            # 绑定事件
+            add_btn.click(
+                fn=add_account,
+                inputs=[email_input, access_token_input, refresh_token_input, project_id_input],
+                outputs=[result_text, accounts_table]
+            ).then(
+                fn=get_account_emails,
+                outputs=delete_email_dropdown
+            )
+            
+            delete_btn.click(
+                fn=delete_account,
+                inputs=[delete_email_dropdown],
+                outputs=[delete_result, accounts_table]
+            ).then(
+                fn=get_account_emails,
+                outputs=delete_email_dropdown
+            )
+        
+        # 设置
+        with gr.Tab("⚙️ 设置"):
+            gr.Markdown("### API Key 配置")
+            gr.Markdown("> 修改后，所有 API 请求需要使用新的 API Key")
+            
+            with gr.Row():
+                api_key_input = gr.Textbox(
+                    label="当前 API Key",
+                    value=config_manager.get_api_key,
+                    placeholder="输入新的 API Key",
+                    interactive=True
+                )
+                save_api_key_btn = gr.Button("💾 保存 API Key", variant="primary")
+            
+            api_key_result = gr.Textbox(label="结果", interactive=False)
+            
+            save_api_key_btn.click(
+                fn=update_api_key,
+                inputs=[api_key_input],
+                outputs=[api_key_result, api_key_input]
+            )
+            
+            gr.Markdown("---")
+            gr.Markdown("### 环境变量配置")
+            gr.Markdown("""
+            > 以下配置在 HF Spaces 环境下通过 Secrets 设置
+            
+            | 配置项 | 环境变量 | 默认值 |
+            |--------|----------|--------|
+            | 数据目录 | `DATA_DIR` | `./data` |
+            | 管理员用户名 | `ADMIN_USERNAME` | `admin` |
+            | 管理员密码 | `ADMIN_PASSWORD` | `antigravity` |
+            """)
+
+
+# 挂载 Gradio 到 FastAPI (带密码保护)
+app = gr.mount_gradio_app(
+    app, 
+    demo, 
+    path="/",
+    auth=(ADMIN_USERNAME, ADMIN_PASSWORD),
+    auth_message="🔐 请输入管理员凭据登录 Antigravity API Proxy"
+)
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)

load_balancer.py

@@ -0,0 +1,50 @@
+"""
+负载均衡器 - Round Robin + 自动故障转移
+"""
+from typing import Optional
+from datetime import datetime, timedelta
+from account_manager import account_manager
+from models import Account
+
+
+class LoadBalancer:
+    """智能负载均衡器"""
+    
+    def __init__(self):
+        self.cooldown_duration = 60  # 默认冷却时间（秒）
+        self.max_retries = 3
+    
+    async def get_next_account(self) -> Optional[Account]:
+        """获取下一个可用账号"""
+        return await account_manager.get_next_token()
+    
+    def mark_account_error(self, account_id: str, error_code: int, error_msg: str = ""):
+        """
+        标记账号错误，根据错误类型决定是否冷却
+        - 429: Too Many Requests - 立即冷却
+        - 400: Bad Request - 可能是请求格式问题，不冷却
+        - 401: Unauthorized - Token 失效，冷却
+        - 403: Forbidden - 配额耗尽，长时间冷却
+        """
+        account_manager.update_account_stats(account_id, success=False, error=error_msg)
+        
+        if error_code == 429:
+            # 速率限制，冷却 60 秒
+            account_manager.set_account_cooldown(account_id, 60)
+            print(f"账号 {account_id} 触发速率限制，冷却 60 秒")
+        elif error_code == 401:
+            # Token 失效，冷却 30 秒（等待刷新）
+            account_manager.set_account_cooldown(account_id, 30)
+            print(f"账号 {account_id} Token 失效，冷却 30 秒")
+        elif error_code == 403:
+            # 配额耗尽，冷却 5 分钟
+            account_manager.set_account_cooldown(account_id, 300)
+            print(f"账号 {account_id} 配额耗尽，冷却 5 分钟")
+    
+    def mark_account_success(self, account_id: str):
+        """标记请求成功"""
+        account_manager.update_account_stats(account_id, success=True)
+
+
+# 全局单例
+load_balancer = LoadBalancer()

models.py

@@ -0,0 +1,146 @@
+"""
+账号数据模型定义 - 支持 Web Session Token (OAuth) 方式
+"""
+from pydantic import BaseModel, Field
+from typing import Optional, Literal, Dict, Any
+from datetime import datetime
+import uuid
+
+
+class OAuthToken(BaseModel):
+    """OAuth Token 信息"""
+    access_token: str
+    refresh_token: str
+    expires_in: int = 3600
+    expiry_timestamp: int = 0  # Unix 时间戳
+    project_id: Optional[str] = None
+    session_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+
+
+class Account(BaseModel):
+    """API 账号模型 - Web Session Token 方式"""
+    id: str = Field(default_factory=lambda: str(uuid.uuid4())[:8])
+    email: str
+    token: OAuthToken
+    enabled: bool = True
+    created_at: datetime = Field(default_factory=datetime.now)
+    
+    # 统计信息
+    total_requests: int = 0
+    successful_requests: int = 0
+    failed_requests: int = 0
+    last_used: Optional[datetime] = None
+    last_error: Optional[str] = None
+    
+    # 冷却状态
+    cooldown_until: Optional[datetime] = None
+    
+    def is_available(self) -> bool:
+        """检查账号是否可用"""
+        if not self.enabled:
+            return False
+        if self.cooldown_until and datetime.now() < self.cooldown_until:
+            return False
+        return True
+    
+    def is_token_expired(self) -> bool:
+        """检查 token 是否过期（提前5分钟刷新）"""
+        now = int(datetime.now().timestamp())
+        return now >= self.token.expiry_timestamp - 300
+    
+    def display_name(self) -> str:
+        """显示名称"""
+        return self.email
+
+
+class AccountStats(BaseModel):
+    """账号统计汇总"""
+    total_accounts: int = 0
+    available_accounts: int = 0
+    total_requests: int = 0
+    success_rate: float = 0.0
+
+
+class ServiceConfig(BaseModel):
+    """服务配置"""
+    api_key: str = "sk-antigravity"
+    admin_username: str = "admin"
+    admin_password: str = "antigravity"
+    port: int = 7860
+    enable_cors: bool = True
+    max_retries: int = 3
+    retry_delay: float = 1.0
+    cooldown_duration: int = 60  # 秒
+
+
+# ============ OpenAI 兼容格式 ============
+
+class OpenAIMessage(BaseModel):
+    """OpenAI 消息格式"""
+    role: Literal["system", "user", "assistant"]
+    content: str
+
+
+class OpenAIChatRequest(BaseModel):
+    """OpenAI Chat Completion 请求"""
+    model: str
+    messages: list[OpenAIMessage]
+    temperature: Optional[float] = 1.0
+    top_p: Optional[float] = 0.95
+    max_tokens: Optional[int] = 8192
+    stream: bool = False
+
+
+class OpenAIChatChoice(BaseModel):
+    """OpenAI 响应选项"""
+    index: int = 0
+    message: Optional[Dict[str, Any]] = None
+    delta: Optional[Dict[str, Any]] = None
+    finish_reason: Optional[str] = None
+
+
+class OpenAIChatResponse(BaseModel):
+    """OpenAI Chat Completion 响应"""
+    id: str = Field(default_factory=lambda: f"chatcmpl-{uuid.uuid4().hex[:8]}")
+    object: str = "chat.completion"
+    created: int = Field(default_factory=lambda: int(datetime.now().timestamp()))
+    model: str
+    choices: list[OpenAIChatChoice]
+    usage: Optional[Dict[str, int]] = None
+
+
+# ============ 模型映射 ============
+
+MODEL_MAPPING: Dict[str, str] = {
+    # Claude 到 Gemini 映射
+    "claude-sonnet-4-5-thinking": "gemini-2.5-pro-preview",
+    "claude-sonnet-4-5": "gemini-2.5-flash-preview", 
+    "claude-opus-4-5": "gemini-2.5-pro-preview",
+    "claude-3-5-sonnet": "gemini-2.5-flash-preview",
+    "claude-3-5-haiku": "gemini-2.5-flash-lite-preview",
+    
+    # Gemini 模型直通
+    "gemini-3-flash": "gemini-3-flash-preview",
+    "gemini-3-pro": "gemini-3-pro-preview",
+    "gemini-3-pro-high": "gemini-3-pro-preview",
+    "gemini-2.5-pro": "gemini-2.5-pro-preview",
+    "gemini-2.5-flash": "gemini-2.5-flash-preview",
+    "gemini-2.5-flash-lite": "gemini-2.5-flash-lite-preview",
+}
+
+SUPPORTED_MODELS = [
+    # Gemini 3 系列
+    {"id": "gemini-3-flash", "name": "Gemini 3 Flash (极速预览)"},
+    {"id": "gemini-3-pro", "name": "Gemini 3 Pro (经典版)"},
+    {"id": "gemini-3-pro-high", "name": "Gemini 3 Pro (最强推理)"},
+    
+    # Gemini 2.5 系列
+    {"id": "gemini-2.5-pro", "name": "Gemini 2.5 Pro (极致推理)"},
+    {"id": "gemini-2.5-flash", "name": "Gemini 2.5 Flash (极速响应)"},
+    {"id": "gemini-2.5-flash-lite", "name": "Gemini 2.5 Flash Lite (轻量极速)"},
+    
+    # Claude 映射
+    {"id": "claude-sonnet-4-5-thinking", "name": "Claude Sonnet 4.5 - 思维链"},
+    {"id": "claude-sonnet-4-5", "name": "Claude Sonnet 4.5"},
+    {"id": "claude-opus-4-5", "name": "Claude Opus 4.5"},
+]

protocol_converter.py

@@ -0,0 +1,237 @@
+"""
+协议转换器 - OpenAI 格式 <-> Gemini 格式
+"""
+from typing import Dict, Any, List
+from models import OpenAIChatRequest, MODEL_MAPPING
+
+
+def convert_openai_to_gemini(request: OpenAIChatRequest) -> Dict[str, Any]:
+    """
+    将 OpenAI Chat Completion 请求转换为 Gemini 格式
+    
+    OpenAI 格式:
+    {
+        "model": "gpt-4",
+        "messages": [
+            {"role": "system", "content": "You are..."},
+            {"role": "user", "content": "Hello"}
+        ]
+    }
+    
+    Gemini 格式:
+    {
+        "contents": [{"role": "user", "parts": [{"text": "Hello"}]}],
+        "systemInstruction": {"role": "user", "parts": [{"text": "You are..."}]},
+        "generationConfig": {...}
+    }
+    """
+    contents = []
+    system_instruction = None
+    
+    for msg in request.messages:
+        if msg.role == "system":
+            system_instruction = {
+                "role": "user",
+                "parts": [{"text": msg.content}]
+            }
+        elif msg.role == "user":
+            contents.append({
+                "role": "user",
+                "parts": [{"text": msg.content}]
+            })
+        elif msg.role == "assistant":
+            contents.append({
+                "role": "model",
+                "parts": [{"text": msg.content}]
+            })
+    
+    # 如果没有 system instruction，使用空字符串
+    if system_instruction is None:
+        system_instruction = {"role": "user", "parts": [{"text": ""}]}
+    
+    # Generation Config
+    generation_config = {
+        "temperature": request.temperature or 1.0,
+        "topP": request.top_p or 0.95,
+        "maxOutputTokens": request.max_tokens or 8192,
+        "candidateCount": 1,
+    }
+    
+    # 检查是否需要启用思维链（thinking）
+    model_lower = request.model.lower()
+    if "thinking" in model_lower or "sonnet-3-7" in model_lower:
+        generation_config["thinkingConfig"] = {
+            "includeThoughts": True,
+            "thinkingBudget": 8191,  # Google Protocol Limit < 8192
+        }
+    
+    return {
+        "contents": contents,
+        "systemInstruction": system_instruction,
+        "generationConfig": generation_config,
+    }
+
+
+def map_model_name(model: str) -> str:
+    """
+    映射模型名称到 Gemini API 支持的名称
+    
+    支持灵活匹配：
+    - 精确匹配: claude-sonnet-4-5 -> gemini-2.5-flash-preview
+    - 模糊匹配: 包含 opus -> gemini-2.5-pro-preview
+    """
+    # 先尝试精确匹配
+    if model in MODEL_MAPPING:
+        return MODEL_MAPPING[model]
+    
+    # 模糊匹配
+    model_lower = model.lower()
+    
+    # Gemini 模型直通（添加 -preview 后缀如需要）
+    if model_lower.startswith("gemini-"):
+        if not model_lower.endswith("-preview"):
+            # 某些模型需要 -preview 后缀
+            if model_lower in ["gemini-3-flash", "gemini-3-pro", "gemini-2.5-pro", "gemini-2.5-flash"]:
+                return model + "-preview"
+        return model
+    
+    # Claude 模型映射
+    if "opus" in model_lower:
+        return "gemini-2.5-pro-preview"
+    if "sonnet" in model_lower:
+        if "thinking" in model_lower:
+            return "gemini-2.5-pro-preview"
+        return "gemini-2.5-flash-preview"
+    if "haiku" in model_lower:
+        return "gemini-2.5-flash-lite-preview"
+    
+    # 默认返回原模型名
+    return model
+
+
+def convert_gemini_to_openai_chunk(gemini_data: Dict[str, Any], model: str) -> Dict[str, Any]:
+    """
+    将 Gemini 流式响应转换为 OpenAI chunk 格式
+    
+    Gemini 格式:
+    {
+        "candidates": [{
+            "content": {"parts": [{"text": "Hello"}]},
+            "finishReason": "STOP"
+        }]
+    }
+    
+    OpenAI 格式:
+    {
+        "id": "chatcmpl-xxx",
+        "object": "chat.completion.chunk",
+        "choices": [{
+            "index": 0,
+            "delta": {"content": "Hello"},
+            "finish_reason": null
+        }]
+    }
+    """
+    import uuid
+    from datetime import datetime
+    
+    # 解析 Gemini 响应
+    candidates = gemini_data.get("candidates", [])
+    if not candidates:
+        # 可能是嵌套在 response 中
+        response = gemini_data.get("response", {})
+        candidates = response.get("candidates", [])
+    
+    text = ""
+    finish_reason = None
+    is_thought = False
+    thought_signature = None
+    
+    if candidates:
+        candidate = candidates[0]
+        content = candidate.get("content", {})
+        parts = content.get("parts", [])
+        
+        if parts:
+            part = parts[0]
+            text = part.get("text", "")
+            is_thought = part.get("thought", False)
+            thought_signature = part.get("thoughtSignature")
+        
+        # 转换结束原因
+        gemini_reason = candidate.get("finishReason")
+        if gemini_reason == "STOP":
+            finish_reason = "stop"
+        elif gemini_reason == "MAX_TOKENS":
+            finish_reason = "length"
+        elif gemini_reason == "SAFETY":
+            finish_reason = "content_filter"
+    
+    # 构建 OpenAI chunk
+    delta = {"content": text}
+    if is_thought:
+        delta["thought"] = True
+    if thought_signature:
+        delta["thoughtSignature"] = thought_signature
+    
+    return {
+        "id": gemini_data.get("responseId", f"chatcmpl-{uuid.uuid4().hex[:8]}"),
+        "object": "chat.completion.chunk",
+        "created": int(datetime.now().timestamp()),
+        "model": model,
+        "choices": [{
+            "index": 0,
+            "delta": delta,
+            "finish_reason": finish_reason
+        }]
+    }
+
+
+def convert_gemini_to_openai_response(gemini_data: Dict[str, Any], model: str) -> Dict[str, Any]:
+    """
+    将 Gemini 非流式响应转换为 OpenAI 格式
+    """
+    import uuid
+    from datetime import datetime
+    
+    candidates = gemini_data.get("candidates", [])
+    if not candidates:
+        response = gemini_data.get("response", {})
+        candidates = response.get("candidates", [])
+    
+    text = ""
+    finish_reason = "stop"
+    
+    if candidates:
+        candidate = candidates[0]
+        content = candidate.get("content", {})
+        parts = content.get("parts", [])
+        
+        if parts:
+            text = parts[0].get("text", "")
+        
+        gemini_reason = candidate.get("finishReason")
+        if gemini_reason == "MAX_TOKENS":
+            finish_reason = "length"
+        elif gemini_reason == "SAFETY":
+            finish_reason = "content_filter"
+    
+    return {
+        "id": f"chatcmpl-{uuid.uuid4().hex[:8]}",
+        "object": "chat.completion",
+        "created": int(datetime.now().timestamp()),
+        "model": model,
+        "choices": [{
+            "index": 0,
+            "message": {
+                "role": "assistant",
+                "content": text
+            },
+            "finish_reason": finish_reason
+        }],
+        "usage": {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0
+        }
+    }

proxy_service.py

@@ -0,0 +1,159 @@
+"""
+API 代理服务 - 调用 Google Antigravity API
+"""
+import httpx
+import json
+import uuid
+from typing import AsyncGenerator
+from models import OpenAIChatRequest, Account
+from protocol_converter import convert_openai_to_gemini, map_model_name, convert_gemini_to_openai_chunk
+from load_balancer import load_balancer
+
+# Google Antigravity 内部 API 端点
+GEMINI_API_URL = "https://daily-cloudcode-pa.sandbox.googleapis.com/v1internal:streamGenerateContent"
+
+
+async def stream_chat_completion(
+    request: OpenAIChatRequest,
+    account: Account
+) -> AsyncGenerator[str, None]:
+    """
+    流式 Chat Completion
+    
+    1. 将 OpenAI 请求转换为 Gemini 格式
+    2. 调用 Google Antigravity API
+    3. 将 Gemini 响应转换回 OpenAI 格式
+    """
+    # 转换请求
+    gemini_request = convert_openai_to_gemini(request)
+    
+    # 映射模型名
+    upstream_model = map_model_name(request.model)
+    
+    # 构建请求体
+    request_body = {
+        "project": account.token.project_id or "default-project",
+        "requestId": str(uuid.uuid4()),
+        "model": upstream_model,
+        "userAgent": "antigravity-hf",
+        "request": {
+            "contents": gemini_request["contents"],
+            "systemInstruction": gemini_request["systemInstruction"],
+            "generationConfig": gemini_request["generationConfig"],
+            "sessionId": account.token.session_id
+        }
+    }
+    
+    # 发送请求
+    url = f"{GEMINI_API_URL}?alt=sse"
+    
+    async with httpx.AsyncClient(timeout=120.0) as client:
+        async with client.stream(
+            "POST",
+            url,
+            json=request_body,
+            headers={
+                "Authorization": f"Bearer {account.token.access_token}",
+                "Host": "daily-cloudcode-pa.sandbox.googleapis.com",
+                "User-Agent": "antigravity-hf/1.0",
+                "Content-Type": "application/json"
+            }
+        ) as response:
+            if response.status_code != 200:
+                error_text = await response.aread()
+                error_msg = f"上游服务错误 ({response.status_code}): {error_text.decode()}"
+                
+                # 标记错误
+                load_balancer.mark_account_error(
+                    account.id, 
+                    response.status_code, 
+                    error_msg
+                )
+                
+                # 返回错误
+                yield f"data: {json.dumps({'error': error_msg})}\n\n"
+                return
+            
+            # 处理 SSE 流
+            buffer = ""
+            async for chunk in response.aiter_text():
+                buffer += chunk
+                
+                # 按行解析 SSE 事件
+                while "\n" in buffer:
+                    line, buffer = buffer.split("\n", 1)
+                    line = line.strip()
+                    
+                    if not line:
+                        continue
+                    
+                    if line.startswith("data: "):
+                        data = line[6:]
+                        
+                        if data == "[DONE]":
+                            yield "data: [DONE]\n\n"
+                            break
+                        
+                        try:
+                            gemini_data = json.loads(data)
+                            openai_chunk = convert_gemini_to_openai_chunk(
+                                gemini_data, 
+                                request.model
+                            )
+                            yield f"data: {json.dumps(openai_chunk)}\n\n"
+                        except json.JSONDecodeError:
+                            continue
+            
+            # 标记成功
+            load_balancer.mark_account_success(account.id)
+
+
+async def chat_completion(
+    request: OpenAIChatRequest,
+    account: Account
+) -> dict:
+    """
+    非流式 Chat Completion
+    """
+    from protocol_converter import convert_gemini_to_openai_response
+    
+    # 转换请求
+    gemini_request = convert_openai_to_gemini(request)
+    upstream_model = map_model_name(request.model)
+    
+    request_body = {
+        "project": account.token.project_id or "default-project",
+        "requestId": str(uuid.uuid4()),
+        "model": upstream_model,
+        "userAgent": "antigravity-hf",
+        "request": {
+            "contents": gemini_request["contents"],
+            "systemInstruction": gemini_request["systemInstruction"],
+            "generationConfig": gemini_request["generationConfig"],
+            "sessionId": account.token.session_id
+        }
+    }
+    
+    # 非流式 URL
+    url = "https://daily-cloudcode-pa.sandbox.googleapis.com/v1internal:generateContent"
+    
+    async with httpx.AsyncClient(timeout=120.0) as client:
+        response = await client.post(
+            url,
+            json=request_body,
+            headers={
+                "Authorization": f"Bearer {account.token.access_token}",
+                "Host": "daily-cloudcode-pa.sandbox.googleapis.com",
+                "User-Agent": "antigravity-hf/1.0",
+                "Content-Type": "application/json"
+            }
+        )
+        
+        if response.status_code != 200:
+            error_msg = f"上游服务错误 ({response.status_code}): {response.text}"
+            load_balancer.mark_account_error(account.id, response.status_code, error_msg)
+            return {"error": error_msg}
+        
+        load_balancer.mark_account_success(account.id)
+        gemini_data = response.json()
+        return convert_gemini_to_openai_response(gemini_data, request.model)

requirements.txt

@@ -0,0 +1,7 @@
+gradio==4.44.0
+huggingface_hub==0.25.0
+fastapi>=0.100.0
+uvicorn>=0.23.0
+httpx>=0.25.0
+pydantic>=2.0.0
+python-dotenv>=1.0.0