Upload 3 files

config.py

@@ -0,0 +1,457 @@
+"""
+Configuration constants for the Geminicli2api proxy server.
+Centralizes all configuration to avoid duplication across modules.
+"""
+import os
+from typing import Any, Optional
+
+# Client Configuration
+
+# 需要自动封禁的错误码 (默认值，可通过环境变量或配置覆盖)
+AUTO_BAN_ERROR_CODES = [401, 403]
+
+# Default Safety Settings for Google API
+DEFAULT_SAFETY_SETTINGS = [
+    {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_NONE"},
+    {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_NONE"},
+    {"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_NONE"},
+    {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_NONE"},
+    {"category": "HARM_CATEGORY_CIVIC_INTEGRITY", "threshold": "BLOCK_NONE"}
+]
+
+# Helper function to get base model name from any variant
+def get_base_model_name(model_name):
+    """Convert variant model name to base model name."""
+    # Remove all possible suffixes in order
+    suffixes = ["-maxthinking", "-nothinking", "-search"]
+    for suffix in suffixes:
+        if model_name.endswith(suffix):
+            return model_name[:-len(suffix)]
+    return model_name
+
+# Helper function to check if model uses search grounding
+def is_search_model(model_name):
+    """Check if model name indicates search grounding should be enabled."""
+    return "-search" in model_name
+
+# Helper function to check if model uses no thinking
+def is_nothinking_model(model_name):
+    """Check if model name indicates thinking should be disabled."""
+    return "-nothinking" in model_name
+
+# Helper function to check if model uses max thinking
+def is_maxthinking_model(model_name):
+    """Check if model name indicates maximum thinking budget should be used."""
+    return "-maxthinking" in model_name
+
+# Helper function to get thinking budget for a model
+def get_thinking_budget(model_name):
+    """Get the appropriate thinking budget for a model based on its name and variant."""
+    
+    if is_nothinking_model(model_name):
+        return 128  # Limited thinking for pro
+    elif is_maxthinking_model(model_name):
+        return 32768
+    else:
+        # Default thinking budget for regular models
+        return -1  # Default for all models
+
+# Helper function to check if thinking should be included in output
+def should_include_thoughts(model_name):
+    """Check if thoughts should be included in the response."""
+    if is_nothinking_model(model_name):
+        # For nothinking mode, still include thoughts if it's a pro model
+        base_model = get_base_model_name(model_name)
+        return "gemini-2.5-pro" in base_model
+    else:
+        # For all other modes, include thoughts
+        return True
+
+# Dynamic Configuration System - Optimized for memory efficiency
+async def get_config_value(key: str, default: Any = None, env_var: Optional[str] = None) -> Any:
+    """Get configuration value with priority: ENV > Storage > default."""
+    # Priority 1: Environment variable
+    if env_var and os.getenv(env_var):
+        return os.getenv(env_var)
+    
+    # Priority 2: Storage system
+    try:
+        from src.storage_adapter import get_storage_adapter
+        storage_adapter = await get_storage_adapter()
+        value = await storage_adapter.get_config(key)
+        # 检查值是否存在（不是None），允许空字符串、0、False等有效值
+        if value is not None:
+            return value
+    except Exception as e:
+        # Debug: print import/storage errors
+        # print(f"Config storage error for key {key}: {e}")
+        pass
+    
+    return default
+
+
+# Configuration getters - all async
+async def get_proxy_config():
+    """Get proxy configuration."""
+    proxy_url = await get_config_value("proxy", env_var="PROXY")
+    return proxy_url if proxy_url else None
+
+async def get_calls_per_rotation() -> int:
+    """Get calls per rotation setting."""
+    env_value = os.getenv("CALLS_PER_ROTATION")
+    if env_value:
+        try:
+            return int(env_value)
+        except ValueError:
+            pass
+    
+    return int(await get_config_value("calls_per_rotation", 100))
+
+async def get_auto_ban_enabled() -> bool:
+    """Get auto ban enabled setting."""
+    env_value = os.getenv("AUTO_BAN")
+    if env_value:
+        return env_value.lower() in ("true", "1", "yes", "on")
+    
+    return bool(await get_config_value("auto_ban_enabled", False))
+
+async def get_auto_ban_error_codes() -> list:
+    """
+    Get auto ban error codes.
+    
+    Environment variable: AUTO_BAN_ERROR_CODES (comma-separated, e.g., "400,403")
+    TOML config key: auto_ban_error_codes
+    Default: [400, 403]
+    """
+    env_value = os.getenv("AUTO_BAN_ERROR_CODES")
+    if env_value:
+        try:
+            return [int(code.strip()) for code in env_value.split(",") if code.strip()]
+        except ValueError:
+            pass
+    
+    codes = await get_config_value("auto_ban_error_codes")
+    if codes and isinstance(codes, list):
+        return codes
+    return AUTO_BAN_ERROR_CODES
+
+async def get_retry_429_max_retries() -> int:
+    """Get max retries for 429 errors."""
+    env_value = os.getenv("RETRY_429_MAX_RETRIES")
+    if env_value:
+        try:
+            return int(env_value)
+        except ValueError:
+            pass
+    
+    return int(await get_config_value("retry_429_max_retries", 5))
+
+async def get_retry_429_enabled() -> bool:
+    """Get 429 retry enabled setting."""
+    env_value = os.getenv("RETRY_429_ENABLED")
+    if env_value:
+        return env_value.lower() in ("true", "1", "yes", "on")
+    
+    return bool(await get_config_value("retry_429_enabled", True))
+
+async def get_retry_429_interval() -> float:
+    """Get 429 retry interval in seconds."""
+    env_value = os.getenv("RETRY_429_INTERVAL")
+    if env_value:
+        try:
+            return float(env_value)
+        except ValueError:
+            pass
+    
+    return float(await get_config_value("retry_429_interval", 1))
+
+
+# Model name lists for different features
+BASE_MODELS = [
+    "gemini-2.5-pro-preview-06-05",
+    "gemini-2.5-pro", 
+    "gemini-2.5-pro-preview-05-06",
+    "gemini-2.5-flash",
+]
+
+def get_available_models(router_type="openai"):
+    """
+    Get available models with feature prefixes.
+    
+    Args:
+        router_type: "openai" or "gemini"
+        
+    Returns:
+        List of model names with feature prefixes
+    """
+    models = []
+    
+    for base_model in BASE_MODELS:
+        # 基础模型
+        models.append(base_model)
+        
+        # 假流式模型 (前缀格式)
+        models.append(f"假流式/{base_model}")
+        
+        # 流式抗截断模型 (仅在流式传输时有效，前缀格式)
+        models.append(f"流式抗截断/{base_model}")
+        
+        # 支持thinking模式后缀与功能前缀组合
+        for thinking_suffix in ["-maxthinking", "-nothinking", "-search"]:
+            # 基础模型 + thinking后缀
+            models.append(f"{base_model}{thinking_suffix}")
+            
+            # 假流式 + thinking后缀
+            models.append(f"假流式/{base_model}{thinking_suffix}")
+            
+            # 流式抗截断 + thinking后缀
+            models.append(f"流式抗截断/{base_model}{thinking_suffix}")
+    
+    return models
+
+def is_fake_streaming_model(model_name: str) -> bool:
+    """Check if model name indicates fake streaming should be used."""
+    return model_name.startswith("假流式/")
+
+def is_anti_truncation_model(model_name: str) -> bool:
+    """Check if model name indicates anti-truncation should be used."""
+    return model_name.startswith("流式抗截断/")
+
+def get_base_model_from_feature_model(model_name: str) -> str:
+    """Get base model name from feature model name."""
+    # Remove feature prefixes
+    for prefix in ["假流式/", "流式抗截断/"]:
+        if model_name.startswith(prefix):
+            return model_name[len(prefix):]
+    return model_name
+
+async def get_anti_truncation_max_attempts() -> int:
+    """
+    Get maximum attempts for anti-truncation continuation.
+    
+    Environment variable: ANTI_TRUNCATION_MAX_ATTEMPTS
+    TOML config key: anti_truncation_max_attempts
+    Default: 3
+    """
+    env_value = os.getenv("ANTI_TRUNCATION_MAX_ATTEMPTS")
+    if env_value:
+        try:
+            return int(env_value)
+        except ValueError:
+            pass
+    
+    return int(await get_config_value("anti_truncation_max_attempts", 3))
+
+# Server Configuration
+async def get_server_host() -> str:
+    """
+    Get server host setting.
+    
+    Environment variable: HOST
+    TOML config key: host
+    Default: 0.0.0.0
+    """
+    return str(await get_config_value("host", "0.0.0.0", "HOST"))
+
+async def get_server_port() -> int:
+    """
+    Get server port setting.
+    
+    Environment variable: PORT
+    TOML config key: port
+    Default: 7861
+    """
+    env_value = os.getenv("PORT")
+    if env_value:
+        try:
+            return int(env_value)
+        except ValueError:
+            pass
+    
+    return int(await get_config_value("port", 7861))
+
+async def get_api_password() -> str:
+    """
+    Get API password setting for chat endpoints.
+    
+    Environment variable: API_PASSWORD
+    TOML config key: api_password
+    Default: Uses PASSWORD env var for compatibility, otherwise 'pwd'
+    """
+    # 优先使用 API_PASSWORD，如果没有则使用通用 PASSWORD 保证兼容性
+    api_password = await get_config_value("api_password", None, "API_PASSWORD")
+    if api_password is not None:
+        return str(api_password)
+    
+    # 兼容性：使用通用密码
+    return str(await get_config_value("password", "pwd", "PASSWORD"))
+
+async def get_panel_password() -> str:
+    """
+    Get panel password setting for web interface.
+    
+    Environment variable: PANEL_PASSWORD
+    TOML config key: panel_password
+    Default: Uses PASSWORD env var for compatibility, otherwise 'pwd'
+    """
+    # 优先使用 PANEL_PASSWORD，如果没有则使用通用 PASSWORD 保证兼容性
+    panel_password = await get_config_value("panel_password", None, "PANEL_PASSWORD")
+    if panel_password is not None:
+        return str(panel_password)
+    
+    # 兼容性：使用通用密码
+    return str(await get_config_value("password", "pwd", "PASSWORD"))
+
+async def get_server_password() -> str:
+    """
+    Get server password setting (deprecated, use get_api_password or get_panel_password).
+    
+    Environment variable: PASSWORD
+    TOML config key: password
+    Default: pwd
+    """
+    return str(await get_config_value("password", "pwd", "PASSWORD"))
+
+async def get_credentials_dir() -> str:
+    """
+    Get credentials directory setting.
+    
+    Environment variable: CREDENTIALS_DIR
+    TOML config key: credentials_dir
+    Default: ./creds
+    """
+    return str(await get_config_value("credentials_dir", "./creds", "CREDENTIALS_DIR"))
+
+async def get_code_assist_endpoint() -> str:
+    """
+    Get Code Assist endpoint setting.
+    
+    Environment variable: CODE_ASSIST_ENDPOINT
+    TOML config key: code_assist_endpoint
+    Default: https://cloudcode-pa.googleapis.com
+    """
+    return str(await get_config_value("code_assist_endpoint", "https://cloudcode-pa.googleapis.com", "CODE_ASSIST_ENDPOINT"))
+
+async def get_auto_load_env_creds() -> bool:
+    """
+    Get auto load environment credentials setting.
+    
+    Environment variable: AUTO_LOAD_ENV_CREDS
+    TOML config key: auto_load_env_creds
+    Default: False
+    """
+    env_value = os.getenv("AUTO_LOAD_ENV_CREDS")
+    if env_value:
+        return env_value.lower() in ("true", "1", "yes", "on")
+    
+    return bool(await get_config_value("auto_load_env_creds", False))
+
+async def get_compatibility_mode_enabled() -> bool:
+    """
+    Get compatibility mode setting.
+    
+    兼容性模式：启用后所有system消息全部转换成user，停用system_instructions。
+    该选项可能会降低模型理解能力，但是能避免流式空回的情况。
+    
+    Environment variable: COMPATIBILITY_MODE
+    TOML config key: compatibility_mode_enabled
+    Default: True
+    """
+    env_value = os.getenv("COMPATIBILITY_MODE")
+    if env_value:
+        return env_value.lower() in ("true", "1", "yes", "on")
+    
+    return bool(await get_config_value("compatibility_mode_enabled", True))
+
+async def get_oauth_proxy_url() -> str:
+    """
+    Get OAuth proxy URL setting.
+    
+    用于Google OAuth2认证的代理URL。
+    
+    Environment variable: OAUTH_PROXY_URL
+    TOML config key: oauth_proxy_url
+    Default: https://oauth2.googleapis.com
+    """
+    return str(await get_config_value("oauth_proxy_url", "https://oauth2.googleapis.com", "OAUTH_PROXY_URL"))
+
+async def get_googleapis_proxy_url() -> str:
+    """
+    Get Google APIs proxy URL setting.
+    
+    用于Google APIs调用的代理URL。
+    
+    Environment variable: GOOGLEAPIS_PROXY_URL
+    TOML config key: googleapis_proxy_url
+    Default: https://www.googleapis.com
+    """
+    return str(await get_config_value("googleapis_proxy_url", "https://www.googleapis.com", "GOOGLEAPIS_PROXY_URL"))
+
+
+async def get_resource_manager_api_url() -> str:
+    """
+    Get Google Cloud Resource Manager API URL setting.
+    
+    用于Google Cloud Resource Manager API的URL。
+    
+    Environment variable: RESOURCE_MANAGER_API_URL
+    TOML config key: resource_manager_api_url
+    Default: https://cloudresourcemanager.googleapis.com
+    """
+    return str(await get_config_value("resource_manager_api_url", "https://cloudresourcemanager.googleapis.com", "RESOURCE_MANAGER_API_URL"))
+
+async def get_service_usage_api_url() -> str:
+    """
+    Get Google Cloud Service Usage API URL setting.
+    
+    用于Google Cloud Service Usage API的URL。
+    
+    Environment variable: SERVICE_USAGE_API_URL
+    TOML config key: service_usage_api_url
+    Default: https://serviceusage.googleapis.com
+    """
+    return str(await get_config_value("service_usage_api_url", "https://serviceusage.googleapis.com", "SERVICE_USAGE_API_URL"))
+
+
+# MongoDB Configuration
+async def get_mongodb_uri() -> str:
+    """
+    Get MongoDB connection URI setting.
+    
+    MongoDB连接URI，用于分布式部署时的数据存储。
+    设置此项后将不再使用本地/creds和TOML文件。
+    
+    Environment variable: MONGODB_URI
+    TOML config key: mongodb_uri
+    Default: None (使用本地文件存储)
+    
+    示例格式:
+    - mongodb://username:password@localhost:27017/database
+    - mongodb+srv://username:password@cluster.mongodb.net/database
+    """
+    return str(await get_config_value("mongodb_uri", "", "MONGODB_URI"))
+
+async def get_mongodb_database() -> str:
+    """
+    Get MongoDB database name setting.
+    
+    MongoDB数据库名称。
+    
+    Environment variable: MONGODB_DATABASE
+    TOML config key: mongodb_database
+    Default: gcli2api
+    """
+    return str(await get_config_value("mongodb_database", "gcli2api", "MONGODB_DATABASE"))
+
+async def is_mongodb_mode() -> bool:
+    """
+    Check if MongoDB mode is enabled.
+    
+    检查是否启用了MongoDB模式。
+    如果配置了MongoDB URI，则启用MongoDB模式，不再使用本地文件。
+    
+    Returns:
+        bool: True if MongoDB mode is enabled, False otherwise
+    """
+    mongodb_uri = await get_mongodb_uri()
+    return bool(mongodb_uri and mongodb_uri.strip())

log.py

@@ -0,0 +1,143 @@
+"""
+日志模块 - 使用环境变量配置
+"""
+import os
+import sys
+import threading
+from datetime import datetime
+
+# 日志级别定义
+LOG_LEVELS = {
+    'debug': 0,
+    'info': 1,
+    'warning': 2,
+    'error': 3,
+    'critical': 4
+}
+
+# 线程锁，用于文件写入同步
+_file_lock = threading.Lock()
+
+# 文件写入状态标志
+_file_writing_disabled = False
+_disable_reason = None
+
+def _get_current_log_level():
+    """获取当前日志级别"""
+    level = os.getenv('LOG_LEVEL', 'info').lower()
+    return LOG_LEVELS.get(level, LOG_LEVELS['info'])
+
+def _get_log_file_path():
+    """获取日志文件路径"""
+    return os.getenv('LOG_FILE', 'log.txt')
+
+def _write_to_file(message: str):
+    """线程安全地写入日志文件"""
+    global _file_writing_disabled, _disable_reason
+    
+    # 如果文件写入已被禁用，直接返回
+    if _file_writing_disabled:
+        return
+    
+    try:
+        log_file = _get_log_file_path()
+        with _file_lock:
+            with open(log_file, 'a', encoding='utf-8') as f:
+                f.write(message + '\n')
+                f.flush()  # 强制刷新到磁盘，确保实时写入
+    except (PermissionError, OSError, IOError) as e:
+        # 检测只读文件系统或权限问题，禁用文件写入
+        _file_writing_disabled = True
+        _disable_reason = str(e)
+        print(f"Warning: File system appears to be read-only or permission denied. Disabling log file writing: {e}", file=sys.stderr)
+        print(f"Log messages will continue to display in console only.", file=sys.stderr)
+    except Exception as e:
+        # 其他异常仍然输出警告但不禁用写入（可能是临时问题）
+        print(f"Warning: Failed to write to log file: {e}", file=sys.stderr)
+
+def _log(level: str, message: str):
+    """
+    内部日志函数
+    """
+    level = level.lower()
+    if level not in LOG_LEVELS:
+        print(f"Warning: Unknown log level '{level}'", file=sys.stderr)
+        return
+    
+    # 检查日志级别
+    current_level = _get_current_log_level()
+    if LOG_LEVELS[level] < current_level:
+        return
+    
+    # 格式化日志消息
+    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    entry = f"[{timestamp}] [{level.upper()}] {message}"
+    
+    # 输出到控制台
+    if level in ('error', 'critical'):
+        print(entry, file=sys.stderr)
+    else:
+        print(entry)
+    
+    # 实时写入文件
+    _write_to_file(entry)
+
+def set_log_level(level: str):
+    """设置日志级别提示"""
+    level = level.lower()
+    if level not in LOG_LEVELS:
+        print(f"Warning: Unknown log level '{level}'. Valid levels: {', '.join(LOG_LEVELS.keys())}")
+        return False
+    
+    print(f"Note: To set log level '{level}', please set LOG_LEVEL environment variable")
+    return True
+
+class Logger:
+    """支持 log('info', 'msg') 和 log.info('msg') 两种调用方式"""
+    
+    def __call__(self, level: str, message: str):
+        """支持 log('info', 'message') 调用方式"""
+        _log(level, message)
+
+    def debug(self, message: str):
+        """记录调试信息"""
+        _log('debug', message)
+    
+    def info(self, message: str):
+        """记录一般信息"""
+        _log('info', message)
+    
+    def warning(self, message: str):
+        """记录警告信息"""
+        _log('warning', message)
+    
+    def error(self, message: str):
+        """记录错误信息"""
+        _log('error', message)
+    
+    def critical(self, message: str):
+        """记录严重错误信息"""
+        _log('critical', message)
+    
+    def get_current_level(self) -> str:
+        """获取当前日志级别名称"""
+        current_level = _get_current_log_level()
+        for name, value in LOG_LEVELS.items():
+            if value == current_level:
+                return name
+        return 'info'
+    
+    def get_log_file(self) -> str:
+        """获取当前日志文件路径"""
+        return _get_log_file_path()
+    
+
+# 导出全局日志实例
+log = Logger()
+
+# 导出的公共接口
+__all__ = ['log', 'set_log_level', 'LOG_LEVELS']
+
+# 使用说明:
+# 1. 设置日志级别: export LOG_LEVEL=debug (或在.env文件中设置)
+# 2. 设置日志文件: export LOG_FILE=log.txt (或在.env文件中设置)

multi_user_auth_web.py

@@ -0,0 +1,275 @@
+#!/usr/bin/env python3
+"""
+OAuth Web 服务器 - 独立的OAuth认证服务
+提供简化的OAuth认证界面，只包含验证功能，不包含上传和管理功能
+"""
+
+from log import log
+import asyncio
+from contextlib import asynccontextmanager
+from fastapi import FastAPI, HTTPException, Depends
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from pydantic import BaseModel
+
+from src.auth import (
+    create_auth_url, 
+    verify_password, 
+    generate_auth_token, 
+    verify_auth_token,
+    asyncio_complete_auth_flow,
+    complete_auth_flow_from_callback_url,
+    CALLBACK_HOST,
+)
+
+# 创建FastAPI应用
+app = FastAPI(
+    title="Google OAuth 认证服务",
+    description="独立的OAuth认证服务，用于获取Google Cloud认证文件",
+)
+
+# HTTP Bearer认证
+security = HTTPBearer()
+
+# 请求模型
+class LoginRequest(BaseModel):
+    password: str
+
+class AuthStartRequest(BaseModel):
+    project_id: str = None  # 现在是可选的，支持自动检测
+
+class AuthCallbackRequest(BaseModel):
+    project_id: str = None  # 现在是可选的，支持自动检测
+
+class AuthCallbackUrlRequest(BaseModel):
+    callback_url: str  # OAuth回调完整URL
+    project_id: str = None  # 可选的项目ID
+
+def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
+    """验证认证令牌"""
+    if not verify_auth_token(credentials.credentials):
+        raise HTTPException(status_code=401, detail="无效的认证令牌")
+    return credentials.credentials
+
+
+@app.get("/", response_class=HTMLResponse)
+async def serve_oauth_page():
+    """提供OAuth认证页面"""
+    try:
+        # 读取HTML文件
+        html_file_path = "./front/multi_user_auth_web.html"
+        
+        with open(html_file_path, "r", encoding="utf-8") as f:
+            html_content = f.read()
+        
+        return HTMLResponse(content=html_content)
+    except FileNotFoundError:
+        raise HTTPException(status_code=404, detail="认证页面不存在")
+    except Exception as e:
+        log.error(f"加载认证页面失败: {e}")
+        raise HTTPException(status_code=500, detail="服务器内部错误")
+
+@app.post("/auth/login")
+async def login(request: LoginRequest):
+    """用户登录"""
+    try:
+        if await verify_password(request.password):
+            token = generate_auth_token()
+            return JSONResponse(content={"token": token, "message": "登录成功"})
+        else:
+            raise HTTPException(status_code=401, detail="密码错误")
+    except HTTPException:
+        raise
+    except Exception as e:
+        log.error(f"登录失败: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/auth/start")
+async def start_auth(request: AuthStartRequest, token: str = Depends(verify_token)):
+    """开始认证流程，支持自动检测项目ID"""
+    try:
+        # 如果没有提供项目ID，尝试自动检测
+        project_id = request.project_id
+        if not project_id:
+            log.info("未提供项目ID，后续将尝试自动检测...")
+        
+        # 使用认证令牌作为用户会话标识
+        user_session = token if token else None
+        result = await create_auth_url(project_id, user_session)
+        
+        if result['success']:
+            # 构建动态回调URL
+            callback_port = result.get('callback_port')
+            callback_url = f"http://{CALLBACK_HOST}:{callback_port}" if callback_port else None
+            
+            response_data = {
+                "auth_url": result['auth_url'],
+                "state": result['state'],
+                "auto_project_detection": result.get('auto_project_detection', False),
+                "detected_project_id": result.get('detected_project_id')
+            }
+            
+            # 如果有回调端口信息，添加到响应中
+            if callback_port:
+                response_data["callback_port"] = callback_port
+                response_data["callback_url"] = callback_url
+            
+            return JSONResponse(content=response_data)
+        else:
+            raise HTTPException(status_code=500, detail=result['error'])
+            
+    except HTTPException:
+        raise
+    except Exception as e:
+        log.error(f"开始认证流程失败: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/auth/callback")
+async def auth_callback(request: AuthCallbackRequest, token: str = Depends(verify_token)):
+    """处理认证回调（异步等待），支持自动检测项目ID"""
+    try:
+        # 项目ID现在是可选的，在回调处理中进行自动检测
+        project_id = request.project_id
+        
+        # 使用认证令牌作为用户会话标识
+        user_session = token if token else None
+        # 异步等待OAuth回调完成
+        result = await asyncio_complete_auth_flow(project_id, user_session)
+        
+        if result['success']:
+            return JSONResponse(content={
+                "credentials": result['credentials'],
+                "file_path": result['file_path'],
+                "message": "认证成功，凭证已保存",
+                "auto_detected_project": result.get('auto_detected_project', False)
+            })
+        else:
+            # 如果需要手动项目ID或项目选择，在响应中标明
+            if result.get('requires_manual_project_id'):
+                # 使用JSON响应
+                return JSONResponse(
+                    status_code=400,
+                    content={
+                        "error": result['error'],
+                        "requires_manual_project_id": True
+                    }
+                )
+            elif result.get('requires_project_selection'):
+                # 返回项目列表供用户选择
+                return JSONResponse(
+                    status_code=400,
+                    content={
+                        "error": result['error'],
+                        "requires_project_selection": True,
+                        "available_projects": result['available_projects']
+                    }
+                )
+            else:
+                raise HTTPException(status_code=400, detail=result['error'])
+            
+    except HTTPException:
+        raise
+    except Exception as e:
+        log.error(f"处理认证回调失败: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/auth/callback-url")
+async def auth_callback_url(request: AuthCallbackUrlRequest, token: str = Depends(verify_token)):
+    """从回调URL直接完成认证，无需启动本地服务器"""
+    try:
+        # 验证URL格式
+        if not request.callback_url or not request.callback_url.startswith(('http://', 'https://')):
+            raise HTTPException(status_code=400, detail="请提供有效的回调URL")
+        
+        # 从回调URL完成认证
+        result = await complete_auth_flow_from_callback_url(request.callback_url, request.project_id)
+        
+        if result['success']:
+            return JSONResponse(content={
+                "credentials": result['credentials'],
+                "file_path": result['file_path'],
+                "message": "从回调URL认证成功，凭证已保存",
+                "auto_detected_project": result.get('auto_detected_project', False)
+            })
+        else:
+            # 处理各种错误情况
+            if result.get('requires_manual_project_id'):
+                return JSONResponse(
+                    status_code=400,
+                    content={
+                        "error": result['error'],
+                        "requires_manual_project_id": True
+                    }
+                )
+            elif result.get('requires_project_selection'):
+                return JSONResponse(
+                    status_code=400,
+                    content={
+                        "error": result['error'],
+                        "requires_project_selection": True,
+                        "available_projects": result['available_projects']
+                    }
+                )
+            else:
+                raise HTTPException(status_code=400, detail=result['error'])
+            
+    except HTTPException:
+        raise
+    except Exception as e:
+        log.error(f"从回调URL处理认证失败: {e}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    log.info("OAuth认证服务启动中...")
+
+    # OAuth回调服务器现在动态按需启动，每个认证流程使用独立端口
+    log.info("OAuth回调服务器将为每个认证流程动态分配端口")
+
+    # 从配置获取密码和端口
+    from config import get_panel_password, get_server_port
+    password = await get_panel_password()
+    port = await get_server_port()
+
+    log.info("Web服务已由 ASGI 服务器启动")
+    
+    print("\n" + "="*60)
+    print("🚀 Google OAuth 认证服务已启动")
+    print("="*60)
+    print(f"📱 Web界面: http://localhost:{port}")
+    print(f"🔐 默认密码: {'已设置' if password else 'pwd (请设置PASSWORD环境变量)'}")
+    print(f"🔄 多用户并发: 支持多用户同时认证（动态端口分配）")
+    print("="*60 + "\n")
+
+    try:
+        yield
+    finally:
+        log.info("OAuth认证服务关闭中...")
+        # OAuth服务器由认证流程自动管理，无需手动清理
+        log.info("OAuth认证服务已关闭")
+
+# 注册 lifespan 处理器
+app.router.lifespan_context = lifespan
+
+if __name__ == "__main__":
+    from hypercorn.asyncio import serve
+    from hypercorn.config import Config
+
+    async def main():
+        # 从配置获取端口
+        from config import get_server_port
+        PORT = await get_server_port()
+        
+        config = Config()
+        config.bind = [f"0.0.0.0:{PORT}"]
+        config.accesslog = "-"
+        config.errorlog = "-"
+        config.loglevel = "INFO"
+        
+        await serve(app, config)
+    
+    asyncio.run(main())