Delete api_utils

api_utils/__init__.py

@@ -1,78 +0,0 @@
-"""
-API工具模块
-提供FastAPI应用初始化、路由处理和工具函数
-"""
-
-# 应用初始化
-from .app import (
-    create_app
-)
-
-# 路由处理器
-from .routes import (
-    read_index,
-    get_css,
-    get_js,
-    get_api_info,
-    health_check,
-    list_models,
-    chat_completions,
-    cancel_request,
-    get_queue_status,
-    websocket_log_endpoint
-)
-
-# 工具函数
-from .utils import (
-    generate_sse_chunk,
-    generate_sse_stop_chunk,
-    generate_sse_error_chunk,
-    use_stream_response,
-    clear_stream_queue,
-    use_helper_get_response,
-    validate_chat_request,
-    prepare_combined_prompt,
-    estimate_tokens,
-    calculate_usage_stats
-)
-
-# 请求处理器
-from .request_processor import (
-    _process_request_refactored
-)
-
-# 队列工作器
-from .queue_worker import (
-    queue_worker
-)
-
-__all__ = [
-    # 应用初始化
-    'create_app',
-    # 路由处理器
-    'read_index',
-    'get_css',
-    'get_js',
-    'get_api_info',
-    'health_check',
-    'list_models',
-    'chat_completions',
-    'cancel_request',
-    'get_queue_status',
-    'websocket_log_endpoint',
-    # 工具函数
-    'generate_sse_chunk',
-    'generate_sse_stop_chunk',
-    'generate_sse_error_chunk',
-    'use_stream_response',
-    'clear_stream_queue',
-    'use_helper_get_response',
-    'validate_chat_request',
-    'prepare_combined_prompt',
-    'estimate_tokens',
-    'calculate_usage_stats',
-    # 请求处理器
-    '_process_request_refactored',
-    # 队列工作器
-    'queue_worker'
-] 

api_utils/app.py

@@ -1,312 +0,0 @@
-"""
-FastAPI应用初始化和生命周期管理
-"""
-
-import asyncio
-import multiprocessing
-import os
-import sys
-from contextlib import asynccontextmanager
-from typing import Optional
-
-from fastapi import FastAPI, Request
-from fastapi.responses import JSONResponse
-from starlette.middleware.base import BaseHTTPMiddleware
-from starlette.types import ASGIApp
-from typing import Callable, Awaitable
-from playwright.async_api import Browser as AsyncBrowser, Playwright as AsyncPlaywright
-
-# --- 配置模块导入 ---
-from config import *
-
-# --- models模块导入 ---
-from models import WebSocketConnectionManager
-
-# --- logging_utils模块导入 ---
-from logging_utils import setup_server_logging, restore_original_streams
-
-# --- browser_utils模块导入 ---
-from browser_utils import (
-    _initialize_page_logic,
-    _close_page_logic,
-    load_excluded_models,
-    _handle_initial_model_state_and_storage
-)
-
-import stream
-from asyncio import Queue, Lock
-from . import auth_utils
-
-# 全局状态变量（这些将在server.py中被引用）
-playwright_manager: Optional[AsyncPlaywright] = None
-browser_instance: Optional[AsyncBrowser] = None
-page_instance = None
-is_playwright_ready = False
-is_browser_connected = False
-is_page_ready = False
-is_initializing = False
-
-global_model_list_raw_json = None
-parsed_model_list = []
-model_list_fetch_event = None
-
-current_ai_studio_model_id = None
-model_switching_lock = None
-
-excluded_model_ids = set()
-
-request_queue = None
-processing_lock = None
-worker_task = None
-
-page_params_cache = {}
-params_cache_lock = None
-
-log_ws_manager = None
-
-STREAM_QUEUE = None
-STREAM_PROCESS = None
-
-# --- Lifespan Context Manager ---
-def _setup_logging():
-    import server
-    log_level_env = os.environ.get('SERVER_LOG_LEVEL', 'INFO')
-    redirect_print_env = os.environ.get('SERVER_REDIRECT_PRINT', 'false')
-    server.log_ws_manager = WebSocketConnectionManager()
-    return setup_server_logging(
-        logger_instance=server.logger,
-        log_ws_manager=server.log_ws_manager,
-        log_level_name=log_level_env,
-        redirect_print_str=redirect_print_env
-    )
-
-def _initialize_globals():
-    import server
-    server.request_queue = Queue()
-    server.processing_lock = Lock()
-    server.model_switching_lock = Lock()
-    server.params_cache_lock = Lock()
-    auth_utils.initialize_keys()
-    server.logger.info("API keys and global locks initialized.")
-
-def _initialize_proxy_settings():
-    import server
-    STREAM_PORT = os.environ.get('STREAM_PORT')
-    if STREAM_PORT == '0':
-        PROXY_SERVER_ENV = os.environ.get('HTTPS_PROXY') or os.environ.get('HTTP_PROXY')
-    else:
-        PROXY_SERVER_ENV = f"http://127.0.0.1:{STREAM_PORT or 3120}/"
-    
-    if PROXY_SERVER_ENV:
-        server.PLAYWRIGHT_PROXY_SETTINGS = {'server': PROXY_SERVER_ENV}
-        if NO_PROXY_ENV:
-            server.PLAYWRIGHT_PROXY_SETTINGS['bypass'] = NO_PROXY_ENV.replace(',', ';')
-        server.logger.info(f"Playwright proxy settings configured: {server.PLAYWRIGHT_PROXY_SETTINGS}")
-    else:
-        server.logger.info("No proxy configured for Playwright.")
-
-async def _start_stream_proxy():
-    import server
-    STREAM_PORT = os.environ.get('STREAM_PORT')
-    if STREAM_PORT != '0':
-        port = int(STREAM_PORT or 3120)
-        STREAM_PROXY_SERVER_ENV = os.environ.get('UNIFIED_PROXY_CONFIG') or os.environ.get('HTTPS_PROXY') or os.environ.get('HTTP_PROXY')
-        server.logger.info(f"Starting STREAM proxy on port {port} with upstream proxy: {STREAM_PROXY_SERVER_ENV}")
-        server.STREAM_QUEUE = multiprocessing.Queue()
-        server.STREAM_PROCESS = multiprocessing.Process(target=stream.start, args=(server.STREAM_QUEUE, port, STREAM_PROXY_SERVER_ENV))
-        server.STREAM_PROCESS.start()
-        server.logger.info("STREAM proxy process started.")
-
-async def _initialize_browser_and_page():
-    import server
-    from playwright.async_api import async_playwright
-    
-    server.logger.info("Starting Playwright...")
-    server.playwright_manager = await async_playwright().start()
-    server.is_playwright_ready = True
-    server.logger.info("Playwright started.")
-
-    ws_endpoint = os.environ.get('CAMOUFOX_WS_ENDPOINT')
-    launch_mode = os.environ.get('LAUNCH_MODE', 'unknown')
-
-    if not ws_endpoint and launch_mode != "direct_debug_no_browser":
-        raise ValueError("CAMOUFOX_WS_ENDPOINT environment variable is missing.")
-
-    if ws_endpoint:
-        server.logger.info(f"Connecting to browser at: {ws_endpoint}")
-        server.browser_instance = await server.playwright_manager.firefox.connect(ws_endpoint, timeout=30000)
-        server.is_browser_connected = True
-        server.logger.info(f"Connected to browser: {server.browser_instance.version}")
-        
-        server.page_instance, server.is_page_ready = await _initialize_page_logic(server.browser_instance)
-        if server.is_page_ready:
-            await _handle_initial_model_state_and_storage(server.page_instance)
-            server.logger.info("Page initialized successfully.")
-        else:
-            server.logger.error("Page initialization failed.")
-    
-    if not server.model_list_fetch_event.is_set():
-        server.model_list_fetch_event.set()
-
-async def _shutdown_resources():
-    import server
-    logger = server.logger
-    logger.info("Shutting down resources...")
-    
-    if server.STREAM_PROCESS:
-        server.STREAM_PROCESS.terminate()
-        logger.info("STREAM proxy terminated.")
-
-    if server.worker_task and not server.worker_task.done():
-        server.worker_task.cancel()
-        try:
-            await asyncio.wait_for(server.worker_task, timeout=5.0)
-        except (asyncio.TimeoutError, asyncio.CancelledError):
-            pass
-        logger.info("Worker task stopped.")
-
-    if server.page_instance:
-        await _close_page_logic()
-    
-    if server.browser_instance and server.browser_instance.is_connected():
-        await server.browser_instance.close()
-        logger.info("Browser connection closed.")
-    
-    if server.playwright_manager:
-        await server.playwright_manager.stop()
-        logger.info("Playwright stopped.")
-
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    """FastAPI application life cycle management"""
-    import server
-    from server import queue_worker
-
-    original_streams = sys.stdout, sys.stderr
-    initial_stdout, initial_stderr = _setup_logging()
-    logger = server.logger
-
-    _initialize_globals()
-    _initialize_proxy_settings()
-    load_excluded_models(EXCLUDED_MODELS_FILENAME)
-    
-    server.is_initializing = True
-    logger.info("Starting AI Studio Proxy Server...")
-
-    try:
-        await _start_stream_proxy()
-        await _initialize_browser_and_page()
-        
-        launch_mode = os.environ.get('LAUNCH_MODE', 'unknown')
-        if server.is_page_ready or launch_mode == "direct_debug_no_browser":
-            server.worker_task = asyncio.create_task(queue_worker())
-            logger.info("Request processing worker started.")
-        else:
-            raise RuntimeError("Failed to initialize browser/page, worker not started.")
-
-        logger.info("Server startup complete.")
-        server.is_initializing = False
-        yield
-    except Exception as e:
-        logger.critical(f"Application startup failed: {e}", exc_info=True)
-        await _shutdown_resources()
-        raise RuntimeError(f"Application startup failed: {e}") from e
-    finally:
-        logger.info("Shutting down server...")
-        await _shutdown_resources()
-        restore_original_streams(initial_stdout, initial_stderr)
-        restore_original_streams(*original_streams)
-        logger.info("Server shutdown complete.")
-
-
-class APIKeyAuthMiddleware(BaseHTTPMiddleware):
-    def __init__(self, app: ASGIApp):
-        super().__init__(app)
-        self.excluded_paths = [
-            "/v1/models",
-            "/health",
-            "/docs",
-            "/openapi.json",
-            # FastAPI 自动生成的其他文档路径
-            "/redoc",
-            "/favicon.ico"
-        ]
-
-    async def dispatch(self, request: Request, call_next: Callable[[Request], Awaitable]):
-        if not auth_utils.API_KEYS:  # 如果 API_KEYS 为空，则不进行验证
-            return await call_next(request)
-
-        # 检查是否是需要保护的路径
-        if not request.url.path.startswith("/v1/"):
-            return await call_next(request)
-
-        # 检查是否是排除的路径
-        for excluded_path in self.excluded_paths:
-            if request.url.path == excluded_path or request.url.path.startswith(excluded_path + "/"):
-                return await call_next(request)
-
-        # 支持多种认证头格式以兼容OpenAI标准
-        api_key = None
-
-        # 1. 优先检查标准的 Authorization: Bearer <token> 头
-        auth_header = request.headers.get("Authorization")
-        if auth_header and auth_header.startswith("Bearer "):
-            api_key = auth_header[7:]  # 移除 "Bearer " 前缀
-
-        # 2. 回退到自定义的 X-API-Key 头（向后兼容）
-        if not api_key:
-            api_key = request.headers.get("X-API-Key")
-
-        if not api_key or not auth_utils.verify_api_key(api_key):
-            return JSONResponse(
-                status_code=401,
-                content={
-                    "error": {
-                        "message": "Invalid or missing API key. Please provide a valid API key using 'Authorization: Bearer <your_key>' or 'X-API-Key: <your_key>' header.",
-                        "type": "invalid_request_error",
-                        "param": None,
-                        "code": "invalid_api_key"
-                    }
-                }
-            )
-        return await call_next(request)
-
-def create_app() -> FastAPI:
-    """创建FastAPI应用实例"""
-    app = FastAPI(
-        title="AI Studio Proxy Server (集成模式)",
-        description="通过 Playwright与 AI Studio 交互的代理服务器。",
-        version="0.6.0-integrated",
-        lifespan=lifespan
-    )
-    
-    # 添加中间件
-    app.add_middleware(APIKeyAuthMiddleware)
-
-    # 注册路由
-    from .routes import (
-        read_index, get_css, get_js, get_api_info,
-        health_check, list_models, chat_completions,
-        cancel_request, get_queue_status, websocket_log_endpoint,
-        get_api_keys, add_api_key, test_api_key, delete_api_key
-    )
-    from fastapi.responses import FileResponse
-    
-    app.get("/", response_class=FileResponse)(read_index)
-    app.get("/webui.css")(get_css)
-    app.get("/webui.js")(get_js)
-    app.get("/api/info")(get_api_info)
-    app.get("/health")(health_check)
-    app.get("/v1/models")(list_models)
-    app.post("/v1/chat/completions")(chat_completions)
-    app.post("/v1/cancel/{req_id}")(cancel_request)
-    app.get("/v1/queue")(get_queue_status)
-    app.websocket("/ws/logs")(websocket_log_endpoint)
-
-    # API密钥管理端点
-    app.get("/api/keys")(get_api_keys)
-    app.post("/api/keys")(add_api_key)
-    app.post("/api/keys/test")(test_api_key)
-    app.delete("/api/keys")(delete_api_key)
-
-    return app

api_utils/auth_utils.py

@@ -1,32 +0,0 @@
-import os
-from typing import Set
-
-API_KEYS: Set[str] = set()
-KEY_FILE_PATH = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "key.txt")
-
-def load_api_keys():
-    """Loads API keys from the key file into the API_KEYS set."""
-    global API_KEYS
-    API_KEYS.clear()
-    if os.path.exists(KEY_FILE_PATH):
-        with open(KEY_FILE_PATH, "r") as f:
-            for line in f:
-                key = line.strip()
-                if key:
-                    API_KEYS.add(key)
-
-def initialize_keys():
-    """Initializes API keys. Ensures key.txt exists and loads keys."""
-    if not os.path.exists(KEY_FILE_PATH):
-        with open(KEY_FILE_PATH, "w") as f:
-            pass  # Create an empty file
-    load_api_keys()
-
-def verify_api_key(api_key_from_header: str) -> bool:
-    """
-    Verifies the API key.
-    Returns True if API_KEYS is empty (no validation) or if the key is valid.
-    """
-    if not API_KEYS:
-        return True
-    return api_key_from_header in API_KEYS

api_utils/dependencies.py

@@ -1,57 +0,0 @@
-"""
-FastAPI 依赖项模块
-"""
-import logging
-from asyncio import Queue, Lock, Event
-from typing import Dict, Any, List, Set
-
-from fastapi import Request
-
-def get_logger() -> logging.Logger:
-    from server import logger
-    return logger
-
-def get_log_ws_manager():
-    from server import log_ws_manager
-    return log_ws_manager
-
-def get_request_queue() -> Queue:
-    from server import request_queue
-    return request_queue
-
-def get_processing_lock() -> Lock:
-    from server import processing_lock
-    return processing_lock
-
-def get_worker_task():
-    from server import worker_task
-    return worker_task
-
-def get_server_state() -> Dict[str, Any]:
-    from server import is_initializing, is_playwright_ready, is_browser_connected, is_page_ready
-    return {
-        "is_initializing": is_initializing,
-        "is_playwright_ready": is_playwright_ready,
-        "is_browser_connected": is_browser_connected,
-        "is_page_ready": is_page_ready,
-    }
-
-def get_page_instance():
-    from server import page_instance
-    return page_instance
-
-def get_model_list_fetch_event() -> Event:
-    from server import model_list_fetch_event
-    return model_list_fetch_event
-
-def get_parsed_model_list() -> List[Dict[str, Any]]:
-    from server import parsed_model_list
-    return parsed_model_list
-
-def get_excluded_model_ids() -> Set[str]:
-    from server import excluded_model_ids
-    return excluded_model_ids
-
-def get_current_ai_studio_model_id() -> str:
-    from server import current_ai_studio_model_id
-    return current_ai_studio_model_id

api_utils/queue_worker.py

@@ -1,266 +0,0 @@
-"""
-队列工作器模块
-处理请求队列中的任务
-"""
-
-import asyncio
-import time
-from fastapi import HTTPException
-
-
-
-async def queue_worker():
-    """队列工作器，处理请求队列中的任务"""
-    # 导入全局变量
-    from server import (
-        logger, request_queue, processing_lock, model_switching_lock, 
-        params_cache_lock
-    )
-    
-    logger.info("--- 队列 Worker 已启动 ---")
-    
-    # 检查并初始化全局变量
-    if request_queue is None:
-        logger.info("初始化 request_queue...")
-        from asyncio import Queue
-        request_queue = Queue()
-    
-    if processing_lock is None:
-        logger.info("初始化 processing_lock...")
-        from asyncio import Lock
-        processing_lock = Lock()
-    
-    if model_switching_lock is None:
-        logger.info("初始化 model_switching_lock...")
-        from asyncio import Lock
-        model_switching_lock = Lock()
-    
-    if params_cache_lock is None:
-        logger.info("初始化 params_cache_lock...")
-        from asyncio import Lock
-        params_cache_lock = Lock()
-    
-    was_last_request_streaming = False
-    last_request_completion_time = 0
-    
-    while True:
-        request_item = None
-        result_future = None
-        req_id = "UNKNOWN"
-        completion_event = None
-        
-        try:
-            # 检查队列中的项目，清理已断开连接的请求
-            queue_size = request_queue.qsize()
-            if queue_size > 0:
-                checked_count = 0
-                items_to_requeue = []
-                processed_ids = set()
-                
-                while checked_count < queue_size and checked_count < 10:
-                    try:
-                        item = request_queue.get_nowait()
-                        item_req_id = item.get("req_id", "unknown")
-                        
-                        if item_req_id in processed_ids:
-                            items_to_requeue.append(item)
-                            continue
-                            
-                        processed_ids.add(item_req_id)
-                        
-                        if not item.get("cancelled", False):
-                            item_http_request = item.get("http_request")
-                            if item_http_request:
-                                try:
-                                    if await item_http_request.is_disconnected():
-                                        logger.info(f"[{item_req_id}] (Worker Queue Check) 检测到客户端已断开，标记为取消。")
-                                        item["cancelled"] = True
-                                        item_future = item.get("result_future")
-                                        if item_future and not item_future.done():
-                                            item_future.set_exception(HTTPException(status_code=499, detail=f"[{item_req_id}] Client disconnected while queued."))
-                                except Exception as check_err:
-                                    logger.error(f"[{item_req_id}] (Worker Queue Check) Error checking disconnect: {check_err}")
-                        
-                        items_to_requeue.append(item)
-                        checked_count += 1
-                    except asyncio.QueueEmpty:
-                        break
-                
-                for item in items_to_requeue:
-                    await request_queue.put(item)
-            
-            # 获取下一个请求
-            try:
-                request_item = await asyncio.wait_for(request_queue.get(), timeout=5.0)
-            except asyncio.TimeoutError:
-                # 如果5秒内没有新请求，继续循环检查
-                continue
-            
-            req_id = request_item["req_id"]
-            request_data = request_item["request_data"]
-            http_request = request_item["http_request"]
-            result_future = request_item["result_future"]
-            
-            if request_item.get("cancelled", False):
-                logger.info(f"[{req_id}] (Worker) 请求已取消，跳过。")
-                if not result_future.done():
-                    result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 请求已被用户取消"))
-                request_queue.task_done()
-                continue
-            
-            is_streaming_request = request_data.stream
-            logger.info(f"[{req_id}] (Worker) 取出请求。模式: {'流式' if is_streaming_request else '非流式'}")
-            
-            # 流式请求间隔控制
-            current_time = time.time()
-            if was_last_request_streaming and is_streaming_request and (current_time - last_request_completion_time < 1.0):
-                delay_time = max(0.5, 1.0 - (current_time - last_request_completion_time))
-                logger.info(f"[{req_id}] (Worker) 连续流式请求，添加 {delay_time:.2f}s 延迟...")
-                await asyncio.sleep(delay_time)
-            
-            if await http_request.is_disconnected():
-                logger.info(f"[{req_id}] (Worker) 客户端在等待锁时断开。取消。")
-                if not result_future.done():
-                    result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端关闭了请求"))
-                request_queue.task_done()
-                continue
-            
-            logger.info(f"[{req_id}] (Worker) 等待处理锁...")
-            async with processing_lock:
-                logger.info(f"[{req_id}] (Worker) 已获取处理锁。开始核心处理...")
-                
-                if await http_request.is_disconnected():
-                    logger.info(f"[{req_id}] (Worker) 客户端在获取锁后断开。取消。")
-                    if not result_future.done():
-                        result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端关闭了请求"))
-                elif result_future.done():
-                    logger.info(f"[{req_id}] (Worker) Future 在处理前已完成/取消。跳过。")
-                else:
-                    # 调用实际的请求处理函数
-                    try:
-                        from api_utils import _process_request_refactored
-                        returned_value = await _process_request_refactored(
-                            req_id, request_data, http_request, result_future
-                        )
-                        
-                        completion_event, submit_btn_loc, client_disco_checker = None, None, None
-                        current_request_was_streaming = False
-
-                        if isinstance(returned_value, tuple) and len(returned_value) == 3:
-                            completion_event, submit_btn_loc, client_disco_checker = returned_value
-                            if completion_event is not None:
-                                current_request_was_streaming = True
-                                logger.info(f"[{req_id}] (Worker) _process_request_refactored returned stream info (event, locator, checker).")
-                            else:
-                                current_request_was_streaming = False
-                                logger.info(f"[{req_id}] (Worker) _process_request_refactored returned a tuple, but completion_event is None (likely non-stream or early exit).")
-                        elif returned_value is None:
-                            current_request_was_streaming = False
-                            logger.info(f"[{req_id}] (Worker) _process_request_refactored returned non-stream completion (None).")
-                        else:
-                            current_request_was_streaming = False
-                            logger.warning(f"[{req_id}] (Worker) _process_request_refactored returned unexpected type: {type(returned_value)}")
-
-                        # 关键修复：在锁内等待流式完成（与原始参考文件一致）
-                        if completion_event:
-                            logger.info(f"[{req_id}] (Worker) 等待流式生成器完成信号...")
-                            try:
-                                from server import RESPONSE_COMPLETION_TIMEOUT
-                                await asyncio.wait_for(completion_event.wait(), timeout=RESPONSE_COMPLETION_TIMEOUT/1000 + 60)
-                                logger.info(f"[{req_id}] (Worker) ✅ 流式生成器完成信号收到。")
-
-                                # 等待发送按钮禁用确认流式响应完全结束
-                                if submit_btn_loc and client_disco_checker:
-                                    logger.info(f"[{req_id}] (Worker) 流式响应完成，检查并处理发送按钮状态...")
-                                    wait_timeout_ms = 30000  # 30 seconds
-                                    try:
-                                        from playwright.async_api import expect as expect_async
-                                        from api_utils.request_processor import ClientDisconnectedError
-
-                                        # 检查客户端连接状态
-                                        client_disco_checker("流式响应后按钮状态检查 - 前置检查: ")
-                                        await asyncio.sleep(0.5)  # 给UI一点时间更新
-
-                                        # 检查按钮是否仍然启用，如果启用则直接点击停止
-                                        logger.info(f"[{req_id}] (Worker) 检查发送按钮状态...")
-                                        try:
-                                            is_button_enabled = await submit_btn_loc.is_enabled(timeout=2000)
-                                            logger.info(f"[{req_id}] (Worker) 发��按钮启用状态: {is_button_enabled}")
-
-                                            if is_button_enabled:
-                                                # 流式响应完成后按钮仍启用，直接点击停止
-                                                logger.info(f"[{req_id}] (Worker) 流式响应完成但按钮仍启用，主动点击按钮停止生成...")
-                                                await submit_btn_loc.click(timeout=5000, force=True)
-                                                logger.info(f"[{req_id}] (Worker) ✅ 发送按钮点击完成。")
-                                            else:
-                                                logger.info(f"[{req_id}] (Worker) 发送按钮已禁用，无需点击。")
-                                        except Exception as button_check_err:
-                                            logger.warning(f"[{req_id}] (Worker) 检查按钮状态失败: {button_check_err}")
-
-                                        # 等待按钮最终禁用
-                                        logger.info(f"[{req_id}] (Worker) 等待发送按钮最终禁用...")
-                                        await expect_async(submit_btn_loc).to_be_disabled(timeout=wait_timeout_ms)
-                                        logger.info(f"[{req_id}] ✅ 发送按钮已禁用。")
-
-                                    except Exception as e_pw_disabled:
-                                        logger.warning(f"[{req_id}] ⚠️ 流式响应后按钮状态处理超时或错误: {e_pw_disabled}")
-                                        from api_utils.request_processor import save_error_snapshot
-                                        await save_error_snapshot(f"stream_post_submit_button_handling_timeout_{req_id}")
-                                    except ClientDisconnectedError:
-                                        logger.info(f"[{req_id}] 客户端在流式响应后按钮状态处理时断开连接。")
-                                elif current_request_was_streaming:
-                                    logger.warning(f"[{req_id}] (Worker) 流式请求但 submit_btn_loc 或 client_disco_checker 未提供。跳过按钮禁用等待。")
-
-                            except asyncio.TimeoutError:
-                                logger.warning(f"[{req_id}] (Worker) ⚠️ 等待流式生成器完成信号超时。")
-                                if not result_future.done():
-                                    result_future.set_exception(HTTPException(status_code=504, detail=f"[{req_id}] Stream generation timed out waiting for completion signal."))
-                            except Exception as ev_wait_err:
-                                logger.error(f"[{req_id}] (Worker) ❌ 等待流式完成事件时出错: {ev_wait_err}")
-                                if not result_future.done():
-                                    result_future.set_exception(HTTPException(status_code=500, detail=f"[{req_id}] Error waiting for stream completion: {ev_wait_err}"))
-
-                    except Exception as process_err:
-                        logger.error(f"[{req_id}] (Worker) _process_request_refactored execution error: {process_err}")
-                        if not result_future.done():
-                            result_future.set_exception(HTTPException(status_code=500, detail=f"[{req_id}] Request processing error: {process_err}"))
-            
-            logger.info(f"[{req_id}] (Worker) 释放处理锁。")
-
-            # 在释放处理锁后立即执行清空操作
-            try:
-                # 清空流式队列缓存
-                from api_utils import clear_stream_queue
-                await clear_stream_queue()
-
-                # 清空聊天历史（对于所有模式：流式和非流式）
-                if submit_btn_loc and client_disco_checker:
-                    from server import page_instance, is_page_ready
-                    if page_instance and is_page_ready:
-                        from browser_utils.page_controller import PageController
-                        page_controller = PageController(page_instance, logger, req_id)
-                        logger.info(f"[{req_id}] (Worker) 执行聊天历史清空（{'流式' if completion_event else '非流式'}模式）...")
-                        await page_controller.clear_chat_history(client_disco_checker)
-                        logger.info(f"[{req_id}] (Worker) ✅ 聊天历史清空完成。")
-                else:
-                    logger.info(f"[{req_id}] (Worker) 跳过聊天历史清空：缺少必要参数（submit_btn_loc: {bool(submit_btn_loc)}, client_disco_checker: {bool(client_disco_checker)}）")
-            except Exception as clear_err:
-                logger.error(f"[{req_id}] (Worker) 清空操作时发生错误: {clear_err}", exc_info=True)
-
-            was_last_request_streaming = is_streaming_request
-            last_request_completion_time = time.time()
-            
-        except asyncio.CancelledError:
-            logger.info("--- 队列 Worker 被取消 ---")
-            if result_future and not result_future.done():
-                result_future.cancel("Worker cancelled")
-            break
-        except Exception as e:
-            logger.error(f"[{req_id}] (Worker) ❌ 处理请求时发生意外错误: {e}", exc_info=True)
-            if result_future and not result_future.done():
-                result_future.set_exception(HTTPException(status_code=500, detail=f"[{req_id}] 服务器内部错误: {e}"))
-        finally:
-            if request_item:
-                request_queue.task_done()
-    
-    logger.info("--- 队列 Worker 已停止 ---") 

api_utils/request_processor.py

@@ -1,795 +0,0 @@
-"""
-请求处理器模块
-包含核心的请求处理逻辑
-"""
-
-import asyncio
-import json
-import os
-import random
-import time
-from typing import Optional, Tuple, Callable, AsyncGenerator
-from asyncio import Event, Future
-
-from fastapi import HTTPException, Request
-from fastapi.responses import JSONResponse, StreamingResponse
-from playwright.async_api import Page as AsyncPage, Locator, Error as PlaywrightAsyncError, expect as expect_async
-
-# --- 配置模块导入 ---
-from config import *
-
-# --- models模块导入 ---
-from models import ChatCompletionRequest, ClientDisconnectedError
-
-# --- browser_utils模块导入 ---
-from browser_utils import (
-    switch_ai_studio_model,
-    save_error_snapshot
-)
-
-# --- api_utils模块导入 ---
-from .utils import (
-    validate_chat_request,
-    prepare_combined_prompt,
-    generate_sse_chunk,
-    generate_sse_stop_chunk,
-    use_stream_response,
-    calculate_usage_stats
-)
-from browser_utils.page_controller import PageController
-
-
-async def _initialize_request_context(req_id: str, request: ChatCompletionRequest) -> dict:
-    """初始化请求上下文"""
-    from server import (
-        logger, page_instance, is_page_ready, parsed_model_list,
-        current_ai_studio_model_id, model_switching_lock, page_params_cache,
-        params_cache_lock
-    )
-    
-    logger.info(f"[{req_id}] 开始处理请求...")
-    logger.info(f"[{req_id}]   请求参数 - Model: {request.model}, Stream: {request.stream}")
-    
-    context = {
-        'logger': logger,
-        'page': page_instance,
-        'is_page_ready': is_page_ready,
-        'parsed_model_list': parsed_model_list,
-        'current_ai_studio_model_id': current_ai_studio_model_id,
-        'model_switching_lock': model_switching_lock,
-        'page_params_cache': page_params_cache,
-        'params_cache_lock': params_cache_lock,
-        'is_streaming': request.stream,
-        'model_actually_switched': False,
-        'requested_model': request.model,
-        'model_id_to_use': None,
-        'needs_model_switching': False
-    }
-    
-    return context
-
-
-async def _analyze_model_requirements(req_id: str, context: dict, request: ChatCompletionRequest) -> dict:
-    """分析模型需求并确定是否需要切换"""
-    logger = context['logger']
-    current_ai_studio_model_id = context['current_ai_studio_model_id']
-    parsed_model_list = context['parsed_model_list']
-    requested_model = request.model
-    
-    if requested_model and requested_model != MODEL_NAME:
-        requested_model_id = requested_model.split('/')[-1]
-        logger.info(f"[{req_id}] 请求使用模型: {requested_model_id}")
-        
-        if parsed_model_list:
-            valid_model_ids = [m.get("id") for m in parsed_model_list]
-            if requested_model_id not in valid_model_ids:
-                raise HTTPException(
-                    status_code=400,
-                    detail=f"[{req_id}] Invalid model '{requested_model_id}'. Available models: {', '.join(valid_model_ids)}"
-                )
-        
-        context['model_id_to_use'] = requested_model_id
-        if current_ai_studio_model_id != requested_model_id:
-            context['needs_model_switching'] = True
-            logger.info(f"[{req_id}] 需要切换模型: 当前={current_ai_studio_model_id} -> 目标={requested_model_id}")
-    
-    return context
-
-
-async def _setup_disconnect_monitoring(req_id: str, http_request: Request, result_future: Future) -> Tuple[Event, asyncio.Task, Callable]:
-    """设置客户端断开连接监控"""
-    from server import logger
-    
-    client_disconnected_event = Event()
-    
-    async def check_disconnect_periodically():
-        while not client_disconnected_event.is_set():
-            try:
-                if await http_request.is_disconnected():
-                    logger.info(f"[{req_id}] 客户端断开，设置事件。")
-                    client_disconnected_event.set()
-                    if not result_future.done():
-                        result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端关闭了请求"))
-                    break
-                await asyncio.sleep(1.0)
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                logger.error(f"[{req_id}] (Disco Check Task) 错误: {e}")
-                client_disconnected_event.set()
-                if not result_future.done():
-                    result_future.set_exception(HTTPException(status_code=500, detail=f"[{req_id}] Internal disconnect checker error: {e}"))
-                break
-    
-    disconnect_check_task = asyncio.create_task(check_disconnect_periodically())
-    
-    def check_client_disconnected(stage: str = ""):
-        if client_disconnected_event.is_set():
-            logger.info(f"[{req_id}] 在 '{stage}' 检测到客户端断开连接。")
-            raise ClientDisconnectedError(f"[{req_id}] Client disconnected at stage: {stage}")
-        return False
-    
-    return client_disconnected_event, disconnect_check_task, check_client_disconnected
-
-
-async def _validate_page_status(req_id: str, context: dict, check_client_disconnected: Callable) -> None:
-    """验证页面状态"""
-    page = context['page']
-    is_page_ready = context['is_page_ready']
-    
-    if not page or page.is_closed() or not is_page_ready:
-        raise HTTPException(status_code=503, detail=f"[{req_id}] AI Studio 页面丢失或未就绪。", headers={"Retry-After": "30"})
-    
-    check_client_disconnected("Initial Page Check")
-
-
-async def _handle_model_switching(req_id: str, context: dict, check_client_disconnected: Callable) -> dict:
-    """处理模型切换逻辑"""
-    if not context['needs_model_switching']:
-        return context
-    
-    logger = context['logger']
-    page = context['page']
-    model_switching_lock = context['model_switching_lock']
-    model_id_to_use = context['model_id_to_use']
-    
-    import server
-    
-    async with model_switching_lock:
-        if server.current_ai_studio_model_id != model_id_to_use:
-            logger.info(f"[{req_id}] 准备切换模型: {server.current_ai_studio_model_id} -> {model_id_to_use}")
-            switch_success = await switch_ai_studio_model(page, model_id_to_use, req_id)
-            if switch_success:
-                server.current_ai_studio_model_id = model_id_to_use
-                context['model_actually_switched'] = True
-                context['current_ai_studio_model_id'] = model_id_to_use
-                logger.info(f"[{req_id}] ✅ 模型切换成功: {server.current_ai_studio_model_id}")
-            else:
-                await _handle_model_switch_failure(req_id, page, model_id_to_use, server.current_ai_studio_model_id, logger)
-    
-    return context
-
-
-async def _handle_model_switch_failure(req_id: str, page: AsyncPage, model_id_to_use: str, model_before_switch: str, logger) -> None:
-    """处理模型切换失败的情况"""
-    import server
-    
-    logger.warning(f"[{req_id}] ❌ 模型切换至 {model_id_to_use} 失败。")
-    # 尝试恢复全局状态
-    server.current_ai_studio_model_id = model_before_switch
-    
-    raise HTTPException(
-        status_code=422,
-        detail=f"[{req_id}] 未能切换到模型 '{model_id_to_use}'。请确保模型可用。"
-    )
-
-
-async def _handle_parameter_cache(req_id: str, context: dict) -> None:
-    """处理参数缓存"""
-    logger = context['logger']
-    params_cache_lock = context['params_cache_lock']
-    page_params_cache = context['page_params_cache']
-    current_ai_studio_model_id = context['current_ai_studio_model_id']
-    model_actually_switched = context['model_actually_switched']
-    
-    async with params_cache_lock:
-        cached_model_for_params = page_params_cache.get("last_known_model_id_for_params")
-        
-        if model_actually_switched or (current_ai_studio_model_id != cached_model_for_params):
-            logger.info(f"[{req_id}] 模型已更改，参数缓存失效。")
-            page_params_cache.clear()
-            page_params_cache["last_known_model_id_for_params"] = current_ai_studio_model_id
-
-
-async def _prepare_and_validate_request(req_id: str, request: ChatCompletionRequest, check_client_disconnected: Callable) -> str:
-    """准备和验证请求"""
-    try:
-        validate_chat_request(request.messages, req_id)
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=f"[{req_id}] 无效请求: {e}")
-    
-    prepared_prompt = prepare_combined_prompt(request.messages, req_id)
-    check_client_disconnected("After Prompt Prep")
-    
-    return prepared_prompt
-
-async def _handle_response_processing(req_id: str, request: ChatCompletionRequest, page: AsyncPage,
-                                    context: dict, result_future: Future,
-                                    submit_button_locator: Locator, check_client_disconnected: Callable) -> Optional[Tuple[Event, Locator, Callable]]:
-    """处理响应生成"""
-    from server import logger
-    
-    is_streaming = request.stream
-    current_ai_studio_model_id = context.get('current_ai_studio_model_id')
-    
-    # 检查是否使用辅助流
-    stream_port = os.environ.get('STREAM_PORT')
-    use_stream = stream_port != '0'
-    
-    if use_stream:
-        return await _handle_auxiliary_stream_response(req_id, request, context, result_future, submit_button_locator, check_client_disconnected)
-    else:
-        return await _handle_playwright_response(req_id, request, page, context, result_future, submit_button_locator, check_client_disconnected)
-
-
-async def _handle_auxiliary_stream_response(req_id: str, request: ChatCompletionRequest, context: dict, 
-                                          result_future: Future, submit_button_locator: Locator, 
-                                          check_client_disconnected: Callable) -> Optional[Tuple[Event, Locator, Callable]]:
-    """使用辅助流处理响应"""
-    from server import logger
-    
-    is_streaming = request.stream
-    current_ai_studio_model_id = context.get('current_ai_studio_model_id')
-    
-    def generate_random_string(length):
-        charset = "abcdefghijklmnopqrstuvwxyz0123456789"
-        return ''.join(random.choice(charset) for _ in range(length))
-
-    if is_streaming:
-        try:
-            completion_event = Event()
-            
-            async def create_stream_generator_from_helper(event_to_set: Event) -> AsyncGenerator[str, None]:
-                last_reason_pos = 0
-                last_body_pos = 0
-                model_name_for_stream = current_ai_studio_model_id or MODEL_NAME
-                chat_completion_id = f"{CHAT_COMPLETION_ID_PREFIX}{req_id}-{int(time.time())}-{random.randint(100, 999)}"
-                created_timestamp = int(time.time())
-                
-                # 用于收集完整内容以计算usage
-                full_reasoning_content = ""
-                full_body_content = ""
-
-                try:
-                    async for raw_data in use_stream_response(req_id):
-                        # 检查客户端是否断开连接
-                        try:
-                            check_client_disconnected(f"流式生成器循环 ({req_id}): ")
-                        except ClientDisconnectedError:
-                            logger.info(f"[{req_id}] 客户端断开连接，终止流式生成")
-                            break
-                        
-                        # 确保 data 是字典类型
-                        if isinstance(raw_data, str):
-                            try:
-                                data = json.loads(raw_data)
-                            except json.JSONDecodeError:
-                                logger.warning(f"[{req_id}] 无法解析流数据JSON: {raw_data}")
-                                continue
-                        elif isinstance(raw_data, dict):
-                            data = raw_data
-                        else:
-                            logger.warning(f"[{req_id}] 未知的流数据类型: {type(raw_data)}")
-                            continue
-                        
-                        # 确保必要的键存在
-                        if not isinstance(data, dict):
-                            logger.warning(f"[{req_id}] 数据不是字典类型: {data}")
-                            continue
-                        
-                        reason = data.get("reason", "")
-                        body = data.get("body", "")
-                        done = data.get("done", False)
-                        function = data.get("function", [])
-                        
-                        # 更新完整内容记录
-                        if reason:
-                            full_reasoning_content = reason
-                        if body:
-                            full_body_content = body
-                        
-                        # 处理推理内容
-                        if len(reason) > last_reason_pos:
-                            output = {
-                                "id": chat_completion_id,
-                                "object": "chat.completion.chunk",
-                                "model": model_name_for_stream,
-                                "created": created_timestamp,
-                                "choices":[{
-                                    "index": 0,
-                                    "delta":{
-                                        "role": "assistant",
-                                        "content": None,
-                                        "reasoning_content": reason[last_reason_pos:],
-                                    },
-                                    "finish_reason": None,
-                                    "native_finish_reason": None,
-                                }]
-                            }
-                            last_reason_pos = len(reason)
-                            yield f"data: {json.dumps(output, ensure_ascii=False, separators=(',', ':'))}\n\n"
-                        
-                        # 处理主体内容
-                        if len(body) > last_body_pos:
-                            finish_reason_val = None
-                            if done:
-                                finish_reason_val = "stop"
-                            
-                            delta_content = {"role": "assistant", "content": body[last_body_pos:]}
-                            choice_item = {
-                                "index": 0,
-                                "delta": delta_content,
-                                "finish_reason": finish_reason_val,
-                                "native_finish_reason": finish_reason_val,
-                            }
-
-                            if done and function and len(function) > 0:
-                                tool_calls_list = []
-                                for func_idx, function_call_data in enumerate(function):
-                                    tool_calls_list.append({
-                                        "id": f"call_{generate_random_string(24)}",
-                                        "index": func_idx,
-                                        "type": "function",
-                                        "function": {
-                                            "name": function_call_data["name"],
-                                            "arguments": json.dumps(function_call_data["params"]),
-                                        },
-                                    })
-                                delta_content["tool_calls"] = tool_calls_list
-                                choice_item["finish_reason"] = "tool_calls"
-                                choice_item["native_finish_reason"] = "tool_calls"
-                                delta_content["content"] = None
-
-                            output = {
-                                "id": chat_completion_id,
-                                "object": "chat.completion.chunk",
-                                "model": model_name_for_stream,
-                                "created": created_timestamp,
-                                "choices": [choice_item]
-                            }
-                            last_body_pos = len(body)
-                            yield f"data: {json.dumps(output, ensure_ascii=False, separators=(',', ':'))}\n\n"
-                        
-                        # 处理只有done=True但没有新内容的情况（仅有函数调用或纯结束）
-                        elif done:
-                            # 如果有函数调用但没有新的body内容
-                            if function and len(function) > 0:
-                                delta_content = {"role": "assistant", "content": None}
-                                tool_calls_list = []
-                                for func_idx, function_call_data in enumerate(function):
-                                    tool_calls_list.append({
-                                        "id": f"call_{generate_random_string(24)}",
-                                        "index": func_idx,
-                                        "type": "function",
-                                        "function": {
-                                            "name": function_call_data["name"],
-                                            "arguments": json.dumps(function_call_data["params"]),
-                                        },
-                                    })
-                                delta_content["tool_calls"] = tool_calls_list
-                                choice_item = {
-                                    "index": 0,
-                                    "delta": delta_content,
-                                    "finish_reason": "tool_calls",
-                                    "native_finish_reason": "tool_calls",
-                                }
-                            else:
-                                # 纯结束，没有新内容和函数调用
-                                choice_item = {
-                                    "index": 0,
-                                    "delta": {"role": "assistant"},
-                                    "finish_reason": "stop",
-                                    "native_finish_reason": "stop",
-                                }
-
-                            output = {
-                                "id": chat_completion_id,
-                                "object": "chat.completion.chunk",
-                                "model": model_name_for_stream,
-                                "created": created_timestamp,
-                                "choices": [choice_item]
-                            }
-                            yield f"data: {json.dumps(output, ensure_ascii=False, separators=(',', ':'))}\n\n"
-                
-                except ClientDisconnectedError:
-                    logger.info(f"[{req_id}] 流式生成器中检测到客户端断开连接")
-                except Exception as e:
-                    logger.error(f"[{req_id}] 流式生成器处理过程中发生错误: {e}", exc_info=True)
-                    # 发送错误信息给客户端
-                    try:
-                        error_chunk = {
-                            "id": chat_completion_id,
-                            "object": "chat.completion.chunk",
-                            "model": model_name_for_stream,
-                            "created": created_timestamp,
-                            "choices": [{
-                                "index": 0,
-                                "delta": {"role": "assistant", "content": f"\n\n[错误: {str(e)}]"},
-                                "finish_reason": "stop",
-                                "native_finish_reason": "stop",
-                            }]
-                        }
-                        yield f"data: {json.dumps(error_chunk, ensure_ascii=False, separators=(',', ':'))}\n\n"
-                    except Exception:
-                        pass  # 如果无法发送错误信息，继续处理结束逻辑
-                finally:
-                    # 计算usage统计
-                    try:
-                        usage_stats = calculate_usage_stats(
-                            [msg.model_dump() for msg in request.messages],
-                            full_body_content,
-                            full_reasoning_content
-                        )
-                        logger.info(f"[{req_id}] 计算的token使用统计: {usage_stats}")
-                        
-                        # 发送带usage的最终chunk
-                        final_chunk = {
-                            "id": chat_completion_id,
-                            "object": "chat.completion.chunk",
-                            "model": model_name_for_stream,
-                            "created": created_timestamp,
-                            "choices": [{
-                                "index": 0,
-                                "delta": {},
-                                "finish_reason": "stop",
-                                "native_finish_reason": "stop"
-                            }],
-                            "usage": usage_stats
-                        }
-                        yield f"data: {json.dumps(final_chunk, ensure_ascii=False, separators=(',', ':'))}\n\n"
-                        logger.info(f"[{req_id}] 已发送带usage统计的最终chunk")
-                        
-                    except Exception as usage_err:
-                        logger.error(f"[{req_id}] 计算或发送usage统计时出错: {usage_err}")
-                    
-                    # 确保总是发送 [DONE] 标记
-                    try:
-                        logger.info(f"[{req_id}] 流式生成器完成，发送 [DONE] 标记")
-                        yield "data: [DONE]\n\n"
-                    except Exception as done_err:
-                        logger.error(f"[{req_id}] 发送 [DONE] 标记时出错: {done_err}")
-                    
-                    # 确保事件被设置
-                    if not event_to_set.is_set():
-                        event_to_set.set()
-                        logger.info(f"[{req_id}] 流式生成器完成事件已设置")
-
-            stream_gen_func = create_stream_generator_from_helper(completion_event)
-            if not result_future.done():
-                result_future.set_result(StreamingResponse(stream_gen_func, media_type="text/event-stream"))
-            else:
-                if not completion_event.is_set():
-                    completion_event.set()
-            
-            return completion_event, submit_button_locator, check_client_disconnected
-
-        except Exception as e:
-            logger.error(f"[{req_id}] 从队列获取流式数据时出错: {e}", exc_info=True)
-            if completion_event and not completion_event.is_set():
-                completion_event.set()
-            raise
-
-    else:  # 非流式
-        content = None
-        reasoning_content = None
-        functions = None
-        final_data_from_aux_stream = None
-
-        async for raw_data in use_stream_response(req_id):
-            check_client_disconnected(f"非流式辅助流 - 循环中 ({req_id}): ")
-            
-            # 确保 data 是字典类型
-            if isinstance(raw_data, str):
-                try:
-                    data = json.loads(raw_data)
-                except json.JSONDecodeError:
-                    logger.warning(f"[{req_id}] 无法解析非流式数据JSON: {raw_data}")
-                    continue
-            elif isinstance(raw_data, dict):
-                data = raw_data
-            else:
-                logger.warning(f"[{req_id}] 非流式未知数据类型: {type(raw_data)}")
-                continue
-            
-            # 确保数据是字典类型
-            if not isinstance(data, dict):
-                logger.warning(f"[{req_id}] 非流式数据不是字典类型: {data}")
-                continue
-                
-            final_data_from_aux_stream = data
-            if data.get("done"):
-                content = data.get("body")
-                reasoning_content = data.get("reason")
-                functions = data.get("function")
-                break
-        
-        if final_data_from_aux_stream and final_data_from_aux_stream.get("reason") == "internal_timeout":
-            logger.error(f"[{req_id}] 非流式请求通过辅助流失败: 内部超时")
-            raise HTTPException(status_code=502, detail=f"[{req_id}] 辅助流处理错误 (内部超时)")
-
-        if final_data_from_aux_stream and final_data_from_aux_stream.get("done") is True and content is None:
-             logger.error(f"[{req_id}] 非流式请求通过辅助流完成但未提供内容")
-             raise HTTPException(status_code=502, detail=f"[{req_id}] 辅助流完成但未提供内容")
-
-        model_name_for_json = current_ai_studio_model_id or MODEL_NAME
-        message_payload = {"role": "assistant", "content": content}
-        finish_reason_val = "stop"
-
-        if functions and len(functions) > 0:
-            tool_calls_list = []
-            for func_idx, function_call_data in enumerate(functions):
-                tool_calls_list.append({
-                    "id": f"call_{generate_random_string(24)}",
-                    "index": func_idx,
-                    "type": "function",
-                    "function": {
-                        "name": function_call_data["name"],
-                        "arguments": json.dumps(function_call_data["params"]),
-                    },
-                })
-            message_payload["tool_calls"] = tool_calls_list
-            finish_reason_val = "tool_calls"
-            message_payload["content"] = None
-        
-        if reasoning_content:
-            message_payload["reasoning_content"] = reasoning_content
-
-        # 计算token使用统计
-        usage_stats = calculate_usage_stats(
-            [msg.model_dump() for msg in request.messages],
-            content or "",
-            reasoning_content
-        )
-
-        response_payload = {
-            "id": f"{CHAT_COMPLETION_ID_PREFIX}{req_id}-{int(time.time())}",
-            "object": "chat.completion",
-            "created": int(time.time()),
-            "model": model_name_for_json,
-            "choices": [{
-                "index": 0,
-                "message": message_payload,
-                "finish_reason": finish_reason_val,
-                "native_finish_reason": finish_reason_val,
-            }],
-            "usage": usage_stats
-        }
-
-        if not result_future.done():
-            result_future.set_result(JSONResponse(content=response_payload))
-        return None
-
-
-async def _handle_playwright_response(req_id: str, request: ChatCompletionRequest, page: AsyncPage, 
-                                    context: dict, result_future: Future, submit_button_locator: Locator, 
-                                    check_client_disconnected: Callable) -> Optional[Tuple[Event, Locator, Callable]]:
-    """使用Playwright处理响应"""
-    from server import logger
-    
-    is_streaming = request.stream
-    current_ai_studio_model_id = context.get('current_ai_studio_model_id')
-    
-    logger.info(f"[{req_id}] 定位响应元素...")
-    response_container = page.locator(RESPONSE_CONTAINER_SELECTOR).last
-    response_element = response_container.locator(RESPONSE_TEXT_SELECTOR)
-    
-    try:
-        await expect_async(response_container).to_be_attached(timeout=20000)
-        check_client_disconnected("After Response Container Attached: ")
-        await expect_async(response_element).to_be_attached(timeout=90000)
-        logger.info(f"[{req_id}] 响应元素已定位。")
-    except (PlaywrightAsyncError, asyncio.TimeoutError, ClientDisconnectedError) as locate_err:
-        if isinstance(locate_err, ClientDisconnectedError):
-            raise
-        logger.error(f"[{req_id}] ❌ 错误: 定位响应元素失败或超时: {locate_err}")
-        await save_error_snapshot(f"response_locate_error_{req_id}")
-        raise HTTPException(status_code=502, detail=f"[{req_id}] 定位AI Studio响应元素失败: {locate_err}")
-    except Exception as locate_exc:
-        logger.exception(f"[{req_id}] ❌ 错误: 定位响应元素时意外错误")
-        await save_error_snapshot(f"response_locate_unexpected_{req_id}")
-        raise HTTPException(status_code=500, detail=f"[{req_id}] 定位响应元素时意外错误: {locate_exc}")
-
-    check_client_disconnected("After Response Element Located: ")
-
-    if is_streaming:
-        completion_event = Event()
-
-        async def create_response_stream_generator():
-            try:
-                # 使用PageController获取响应
-                page_controller = PageController(page, logger, req_id)
-                final_content = await page_controller.get_response(check_client_disconnected)
-                
-                # 生成流式响应 - 保持Markdown格式
-                # 按行分割以保持换行符和Markdown结构
-                lines = final_content.split('\n')
-                for line_idx, line in enumerate(lines):
-                    # 检查客户端是否断开连接
-                    try:
-                        check_client_disconnected(f"Playwright流式生成器循环 ({req_id}): ")
-                    except ClientDisconnectedError:
-                        logger.info(f"[{req_id}] Playwright流式生成器中检测到客户端断开连接")
-                        break
-
-                    # 输出当前行的内容（包括空行，以保持Markdown格式）
-                    if line:  # 非空行按字符分块输出
-                        chunk_size = 5  # 每次输出5个字符，平衡速度和体验
-                        for i in range(0, len(line), chunk_size):
-                            chunk = line[i:i+chunk_size]
-                            yield generate_sse_chunk(chunk, req_id, current_ai_studio_model_id or MODEL_NAME)
-                            await asyncio.sleep(0.03)  # 适中的输出速度
-
-                    # 添加换行符（除了最后一行）
-                    if line_idx < len(lines) - 1:
-                        yield generate_sse_chunk('\n', req_id, current_ai_studio_model_id or MODEL_NAME)
-                        await asyncio.sleep(0.01)
-                
-                # 计算并发送带usage的完成块
-                usage_stats = calculate_usage_stats(
-                    [msg.model_dump() for msg in request.messages],
-                    final_content,
-                    ""  # Playwright模式没有reasoning content
-                )
-                logger.info(f"[{req_id}] Playwright非流式计算的token使用统计: {usage_stats}")
-                
-                # 发送带usage的完成块
-                yield generate_sse_stop_chunk(req_id, current_ai_studio_model_id or MODEL_NAME, "stop", usage_stats)
-                
-            except ClientDisconnectedError:
-                logger.info(f"[{req_id}] Playwright流式生成器中检测到客户端断开连接")
-            except Exception as e:
-                logger.error(f"[{req_id}] Playwright流式生成器处理过程中发生错误: {e}", exc_info=True)
-                # 发送错误信息给客户端
-                try:
-                    yield generate_sse_chunk(f"\n\n[错误: {str(e)}]", req_id, current_ai_studio_model_id or MODEL_NAME)
-                    yield generate_sse_stop_chunk(req_id, current_ai_studio_model_id or MODEL_NAME)
-                except Exception:
-                    pass  # 如果无法发送错误信息，继续处理结束逻辑
-            finally:
-                # 确保事件被设置
-                if not completion_event.is_set():
-                    completion_event.set()
-                    logger.info(f"[{req_id}] Playwright流式生成器完成事件已设置")
-
-        stream_gen_func = create_response_stream_generator()
-        if not result_future.done():
-            result_future.set_result(StreamingResponse(stream_gen_func, media_type="text/event-stream"))
-        
-        return completion_event, submit_button_locator, check_client_disconnected
-    else:
-        # 使用PageController获取响应
-        page_controller = PageController(page, logger, req_id)
-        final_content = await page_controller.get_response(check_client_disconnected)
-        
-        # 计算token使用统计
-        usage_stats = calculate_usage_stats(
-            [msg.model_dump() for msg in request.messages],
-            final_content,
-            ""  # Playwright模式没有reasoning content
-        )
-        logger.info(f"[{req_id}] Playwright非流式计算的token使用统计: {usage_stats}")
-        
-        response_payload = {
-            "id": f"{CHAT_COMPLETION_ID_PREFIX}{req_id}-{int(time.time())}",
-            "object": "chat.completion",
-            "created": int(time.time()),
-            "model": current_ai_studio_model_id or MODEL_NAME,
-            "choices": [{
-                "index": 0,
-                "message": {"role": "assistant", "content": final_content},
-                "finish_reason": "stop"
-            }],
-            "usage": usage_stats
-        }
-        
-        if not result_future.done():
-            result_future.set_result(JSONResponse(content=response_payload))
-        
-        return None
-
-
-async def _cleanup_request_resources(req_id: str, disconnect_check_task: Optional[asyncio.Task], 
-                                   completion_event: Optional[Event], result_future: Future, 
-                                   is_streaming: bool) -> None:
-    """清理请求资源"""
-    from server import logger
-    
-    if disconnect_check_task and not disconnect_check_task.done():
-        disconnect_check_task.cancel()
-        try: 
-            await disconnect_check_task
-        except asyncio.CancelledError: 
-            pass
-        except Exception as task_clean_err: 
-            logger.error(f"[{req_id}] 清理任务时出错: {task_clean_err}")
-    
-    logger.info(f"[{req_id}] 处理完成。")
-    
-    if is_streaming and completion_event and not completion_event.is_set() and (result_future.done() and result_future.exception() is not None):
-         logger.warning(f"[{req_id}] 流式请求异常，确保完成事件已设置。")
-         completion_event.set()
-
-
-async def _process_request_refactored(
-    req_id: str,
-    request: ChatCompletionRequest,
-    http_request: Request,
-    result_future: Future
-) -> Optional[Tuple[Event, Locator, Callable[[str], bool]]]:
-    """核心请求处理函数 - 重构版本"""
-    
-    context = await _initialize_request_context(req_id, request)
-    context = await _analyze_model_requirements(req_id, context, request)
-    
-    client_disconnected_event, disconnect_check_task, check_client_disconnected = await _setup_disconnect_monitoring(
-        req_id, http_request, result_future
-    )
-    
-    page = context['page']
-    submit_button_locator = page.locator(SUBMIT_BUTTON_SELECTOR) if page else None
-    completion_event = None
-    
-    try:
-        await _validate_page_status(req_id, context, check_client_disconnected)
-        
-        page_controller = PageController(page, context['logger'], req_id)
-
-        await _handle_model_switching(req_id, context, check_client_disconnected)
-        await _handle_parameter_cache(req_id, context)
-        
-        prepared_prompt = await _prepare_and_validate_request(req_id, request, check_client_disconnected)
-
-        # 使用PageController处理页面交互
-        # 注意：聊天历史清空已移至队列处理锁释放后执行
-
-        await page_controller.adjust_parameters(
-            request.model_dump(exclude_none=True), # 使用 exclude_none=True 避免传递None值
-            context['page_params_cache'],
-            context['params_cache_lock'],
-            context['model_id_to_use'],
-            context['parsed_model_list'],
-            check_client_disconnected
-        )
-        
-        await page_controller.submit_prompt(prepared_prompt, check_client_disconnected)
-        
-        # 响应处理仍然需要在这里，因为它决定了是流式还是非流式，并设置future
-        response_result = await _handle_response_processing(
-            req_id, request, page, context, result_future, submit_button_locator, check_client_disconnected
-        )
-        
-        if response_result:
-            completion_event, _, _ = response_result
-        
-        return completion_event, submit_button_locator, check_client_disconnected
-        
-    except ClientDisconnectedError as disco_err:
-        context['logger'].info(f"[{req_id}] 捕获到客户端断开连接信号: {disco_err}")
-        if not result_future.done():
-             result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] Client disconnected during processing."))
-    except HTTPException as http_err:
-        context['logger'].warning(f"[{req_id}] 捕获到 HTTP 异常: {http_err.status_code} - {http_err.detail}")
-        if not result_future.done():
-            result_future.set_exception(http_err)
-    except PlaywrightAsyncError as pw_err:
-        context['logger'].error(f"[{req_id}] 捕获到 Playwright 错误: {pw_err}")
-        await save_error_snapshot(f"process_playwright_error_{req_id}")
-        if not result_future.done():
-            result_future.set_exception(HTTPException(status_code=502, detail=f"[{req_id}] Playwright interaction failed: {pw_err}"))
-    except Exception as e:
-        context['logger'].exception(f"[{req_id}] 捕获到意外错误")
-        await save_error_snapshot(f"process_unexpected_error_{req_id}")
-        if not result_future.done():
-            result_future.set_exception(HTTPException(status_code=500, detail=f"[{req_id}] Unexpected server error: {e}"))
-    finally:
-        await _cleanup_request_resources(req_id, disconnect_check_task, completion_event, result_future, request.stream)

api_utils/request_processor_backup.py

@@ -1,274 +0,0 @@
-"""
-请求处理器模块
-包含核心的请求处理逻辑
-"""
-
-import asyncio
-import json
-import os
-import random
-import time
-from typing import Optional, Tuple, Callable, AsyncGenerator
-from asyncio import Event, Future
-
-from fastapi import HTTPException, Request
-from fastapi.responses import JSONResponse, StreamingResponse
-from playwright.async_api import Page as AsyncPage, Locator, Error as PlaywrightAsyncError, expect as expect_async, TimeoutError
-
-# --- 配置模块导入 ---
-from config import *
-
-# --- models模块导入 ---
-from models import ChatCompletionRequest, ClientDisconnectedError
-
-# --- browser_utils模块导入 ---
-from browser_utils import (
-    switch_ai_studio_model, 
-    save_error_snapshot,
-    _wait_for_response_completion,
-    _get_final_response_content,
-    detect_and_extract_page_error
-)
-
-# --- api_utils模块导入 ---
-from .utils import (
-    validate_chat_request, 
-    prepare_combined_prompt,
-    generate_sse_chunk,
-    generate_sse_stop_chunk,
-    generate_sse_error_chunk,
-    use_helper_get_response,
-    use_stream_response
-)
-
-
-async def _process_request_refactored(
-    req_id: str,
-    request: ChatCompletionRequest,
-    http_request: Request,
-    result_future: Future
-) -> Optional[Tuple[Event, Locator, Callable[[str], bool]]]:
-    """核心请求处理函数 - 完整版本"""
-    global current_ai_studio_model_id
-    
-    # 导入全局变量
-    from server import (
-        logger, page_instance, is_page_ready, parsed_model_list,
-        current_ai_studio_model_id, model_switching_lock, page_params_cache,
-        params_cache_lock
-    )
-    
-    model_actually_switched_in_current_api_call = False
-    logger.info(f"[{req_id}] (Refactored Process) 开始处理请求...")
-    logger.info(f"[{req_id}]   请求参数 - Model: {request.model}, Stream: {request.stream}")
-    logger.info(f"[{req_id}]   请求参数 - Temperature: {request.temperature}")
-    logger.info(f"[{req_id}]   请求参数 - Max Output Tokens: {request.max_output_tokens}")
-    logger.info(f"[{req_id}]   请求参数 - Stop Sequences: {request.stop}")
-    logger.info(f"[{req_id}]   请求参数 - Top P: {request.top_p}")
-    
-    is_streaming = request.stream
-    page: Optional[AsyncPage] = page_instance
-    completion_event: Optional[Event] = None
-    requested_model = request.model
-    model_id_to_use = None
-    needs_model_switching = False
-    
-    if requested_model and requested_model != MODEL_NAME:
-        requested_model_parts = requested_model.split('/')
-        requested_model_id = requested_model_parts[-1] if len(requested_model_parts) > 1 else requested_model
-        logger.info(f"[{req_id}] 请求使用模型: {requested_model_id}")
-        if parsed_model_list:
-            valid_model_ids = [m.get("id") for m in parsed_model_list]
-            if requested_model_id not in valid_model_ids:
-                logger.error(f"[{req_id}] ❌ 无效的模型ID: {requested_model_id}。可用模型: {valid_model_ids}")
-                raise HTTPException(status_code=400, detail=f"[{req_id}] Invalid model '{requested_model_id}'. Available models: {', '.join(valid_model_ids)}")
-        model_id_to_use = requested_model_id
-        if current_ai_studio_model_id != model_id_to_use:
-            needs_model_switching = True
-            logger.info(f"[{req_id}] 需要切换模型: 当前={current_ai_studio_model_id} -> 目标={model_id_to_use}")
-        else:
-            logger.info(f"[{req_id}] 请求模型与当前模型相同 ({model_id_to_use})，无需切换")
-    else:
-        logger.info(f"[{req_id}] 未指定具体模型或使用代理模型名称，将使用当前模型: {current_ai_studio_model_id or '未知'}")
-    
-    client_disconnected_event = Event()
-    disconnect_check_task = None
-    input_field_locator = page.locator(INPUT_SELECTOR) if page else None
-    submit_button_locator = page.locator(SUBMIT_BUTTON_SELECTOR) if page else None
-
-    async def check_disconnect_periodically():
-        while not client_disconnected_event.is_set():
-            try:
-                if await http_request.is_disconnected():
-                    logger.info(f"[{req_id}] (Disco Check Task) 客户端断开。设置事件并尝试停止。")
-                    client_disconnected_event.set()
-                    try:
-                        if submit_button_locator and await submit_button_locator.is_enabled(timeout=1500):
-                             if input_field_locator and await input_field_locator.input_value(timeout=1500) == '':
-                                 logger.info(f"[{req_id}] (Disco Check Task)   点击停止...")
-                                 await submit_button_locator.click(timeout=3000, force=True)
-                    except Exception as click_err: 
-                        logger.warning(f"[{req_id}] (Disco Check Task) 停止按钮点击失败: {click_err}")
-                    if not result_future.done(): 
-                        result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] 客户端在处理期间关闭了请求"))
-                    break
-                await asyncio.sleep(1.0)
-            except asyncio.CancelledError: 
-                break
-            except Exception as e:
-                logger.error(f"[{req_id}] (Disco Check Task) 错误: {e}")
-                client_disconnected_event.set()
-                if not result_future.done(): 
-                    result_future.set_exception(HTTPException(status_code=500, detail=f"[{req_id}] Internal disconnect checker error: {e}"))
-                break
-    
-    disconnect_check_task = asyncio.create_task(check_disconnect_periodically())
-    
-    def check_client_disconnected(*args):
-        msg_to_log = ""
-        if len(args) == 1 and isinstance(args[0], str):
-            msg_to_log = args[0]
-
-        if client_disconnected_event.is_set():
-            logger.info(f"[{req_id}] {msg_to_log}检测到客户端断开连接事件。")
-            raise ClientDisconnectedError(f"[{req_id}] Client disconnected event set.")
-        return False
-    
-    try:
-        if not page or page.is_closed() or not is_page_ready:
-            raise HTTPException(status_code=503, detail=f"[{req_id}] AI Studio 页面丢失或未就绪。", headers={"Retry-After": "30"})
-        
-        check_client_disconnected("Initial Page Check: ")
-        
-        # 模型切换逻辑
-        if needs_model_switching and model_id_to_use:
-            async with model_switching_lock:
-                model_before_switch_attempt = current_ai_studio_model_id
-                if current_ai_studio_model_id != model_id_to_use:
-                    logger.info(f"[{req_id}] 获取锁后准备切换: 当前内存中模型={current_ai_studio_model_id}, 目标={model_id_to_use}")
-                    switch_success = await switch_ai_studio_model(page, model_id_to_use, req_id)
-                    if switch_success:
-                        current_ai_studio_model_id = model_id_to_use
-                        model_actually_switched_in_current_api_call = True
-                        logger.info(f"[{req_id}] ✅ 模型切换成功。全局模型状态已更新为: {current_ai_studio_model_id}")
-                    else:
-                        logger.warning(f"[{req_id}] ❌ 模型切换至 {model_id_to_use} 失败 (AI Studio 未接受或覆盖了更改)。")
-                        active_model_id_after_fail = model_before_switch_attempt
-                        try:
-                            final_prefs_str_after_fail = await page.evaluate("() => localStorage.getItem('aiStudioUserPreference')")
-                            if final_prefs_str_after_fail:
-                                final_prefs_obj_after_fail = json.loads(final_prefs_str_after_fail)
-                                model_path_in_final_prefs = final_prefs_obj_after_fail.get("promptModel")
-                                if model_path_in_final_prefs and isinstance(model_path_in_final_prefs, str):
-                                    active_model_id_after_fail = model_path_in_final_prefs.split('/')[-1]
-                        except Exception as read_final_prefs_err:
-                            logger.error(f"[{req_id}] 切换失败后读取最终 localStorage 出错: {read_final_prefs_err}")
-                        current_ai_studio_model_id = active_model_id_after_fail
-                        logger.info(f"[{req_id}] 全局模型状态在切换失败后设置为 (或保持为): {current_ai_studio_model_id}")
-                        actual_displayed_model_name = "未知 (无法读取)"
-                        try:
-                            model_wrapper_locator = page.locator('#mat-select-value-0 mat-select-trigger').first
-                            actual_displayed_model_name = await model_wrapper_locator.inner_text(timeout=3000)
-                        except Exception:
-                            pass
-                        raise HTTPException(
-                            status_code=422,
-                            detail=f"[{req_id}] AI Studio 未能应用所请求的模型 '{model_id_to_use}' 或该模型不受支持。请选择 AI Studio 网页界面中可用的模型。当前实际生效的模型 ID 为 '{current_ai_studio_model_id}', 页面显示为 '{actual_displayed_model_name}'."
-                        )
-                else:
-                    logger.info(f"[{req_id}] 获取锁后发现模型已是目标模型 {current_ai_studio_model_id}，无需切换")
-        
-        # 参数缓存处理
-        async with params_cache_lock:
-            cached_model_for_params = page_params_cache.get("last_known_model_id_for_params")
-            if model_actually_switched_in_current_api_call or \
-               (current_ai_studio_model_id is not None and current_ai_studio_model_id != cached_model_for_params):
-                action_taken = "Invalidating" if page_params_cache else "Initializing"
-                logger.info(f"[{req_id}] {action_taken} parameter cache. Reason: Model context changed (switched this call: {model_actually_switched_in_current_api_call}, current model: {current_ai_studio_model_id}, cache model: {cached_model_for_params}).")
-                page_params_cache.clear()
-                if current_ai_studio_model_id:
-                    page_params_cache["last_known_model_id_for_params"] = current_ai_studio_model_id
-            else:
-                logger.debug(f"[{req_id}] Parameter cache for model '{cached_model_for_params}' remains valid (current model: '{current_ai_studio_model_id}', switched this call: {model_actually_switched_in_current_api_call}).")
-        
-        # 验证请求
-        try: 
-            validate_chat_request(request.messages, req_id)
-        except ValueError as e: 
-            raise HTTPException(status_code=400, detail=f"[{req_id}] 无效请求: {e}")
-        
-        # 准备提示
-        prepared_prompt = prepare_combined_prompt(request.messages, req_id)
-        check_client_disconnected("After Prompt Prep: ")
-        
-        # 这里需要添加完整的处理逻辑 - 由于函数太长，暂时返回简化响应
-        logger.info(f"[{req_id}] (Refactored Process) 处理完整逻辑 - 需要从备份恢复剩余部分")
-        
-        # 简单响应用于测试
-        if is_streaming:
-            completion_event = Event()
-            
-            async def create_simple_stream_generator():
-                try:
-                    yield generate_sse_chunk("正在处理请求...", req_id, MODEL_NAME)
-                    await asyncio.sleep(1)
-                    yield generate_sse_chunk("处理完成", req_id, MODEL_NAME)
-                    yield generate_sse_stop_chunk(req_id, MODEL_NAME)
-                    yield "data: [DONE]\n\n"
-                finally:
-                    if not completion_event.is_set():
-                        completion_event.set()
-            
-            if not result_future.done():
-                result_future.set_result(StreamingResponse(create_simple_stream_generator(), media_type="text/event-stream"))
-            
-            return completion_event, submit_button_locator, check_client_disconnected
-        else:
-            response_payload = {
-                "id": f"{CHAT_COMPLETION_ID_PREFIX}{req_id}-{int(time.time())}",
-                "object": "chat.completion",
-                "created": int(time.time()),
-                "model": MODEL_NAME,
-                "choices": [{
-                    "index": 0,
-                    "message": {"role": "assistant", "content": "处理完成 - 需要完整逻辑"},
-                    "finish_reason": "stop"
-                }],
-                "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
-            }
-            
-            if not result_future.done():
-                result_future.set_result(JSONResponse(content=response_payload))
-            
-            return None
-        
-    except ClientDisconnectedError as disco_err:
-        logger.info(f"[{req_id}] (Refactored Process) 捕获到客户端断开连接信号: {disco_err}")
-        if not result_future.done():
-             result_future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] Client disconnected during processing."))
-    except HTTPException as http_err:
-        logger.warning(f"[{req_id}] (Refactored Process) 捕获到 HTTP 异常: {http_err.status_code} - {http_err.detail}")
-        if not result_future.done(): 
-            result_future.set_exception(http_err)
-    except Exception as e:
-        logger.exception(f"[{req_id}] (Refactored Process) 捕获到意外错误")
-        await save_error_snapshot(f"process_unexpected_error_{req_id}")
-        if not result_future.done(): 
-            result_future.set_exception(HTTPException(status_code=500, detail=f"[{req_id}] Unexpected server error: {e}"))
-    finally:
-        if disconnect_check_task and not disconnect_check_task.done():
-            disconnect_check_task.cancel()
-            try: 
-                await disconnect_check_task
-            except asyncio.CancelledError: 
-                pass
-            except Exception as task_clean_err: 
-                logger.error(f"[{req_id}] 清理任务时出错: {task_clean_err}")
-        
-        logger.info(f"[{req_id}] (Refactored Process) 处理完成。")
-        
-        if is_streaming and completion_event and not completion_event.is_set() and (result_future.done() and result_future.exception() is not None):
-             logger.warning(f"[{req_id}] (Refactored Process) 流式请求异常，确保完成事件已设置。")
-             completion_event.set()
-        
-        return completion_event, submit_button_locator, check_client_disconnected 

api_utils/routes.py

@@ -1,374 +0,0 @@
-"""
-FastAPI路由处理器模块
-包含所有API端点的处理函数
-"""
-
-import asyncio
-import os
-import random
-import time
-import uuid
-from typing import Dict, List, Any, Set
-from asyncio import Queue, Future, Lock, Event
-import logging
-
-from fastapi import HTTPException, Request, WebSocket, WebSocketDisconnect, Depends
-from fastapi.responses import JSONResponse, FileResponse
-from pydantic import BaseModel
-from playwright.async_api import Page as AsyncPage
-
-# --- 配置模块导入 ---
-from config import *
-
-# --- models模块导入 ---
-from models import ChatCompletionRequest, WebSocketConnectionManager
-
-# --- browser_utils模块导入 ---
-from browser_utils import _handle_model_list_response
-
-# --- 依赖项导入 ---
-from .dependencies import *
-
-
-# --- 静态文件端点 ---
-async def read_index(logger: logging.Logger = Depends(get_logger)):
-    """返回主页面"""
-    index_html_path = os.path.join(os.path.dirname(__file__), "..", "index.html")
-    if not os.path.exists(index_html_path):
-        logger.error(f"index.html not found at {index_html_path}")
-        raise HTTPException(status_code=404, detail="index.html not found")
-    return FileResponse(index_html_path)
-
-
-async def get_css(logger: logging.Logger = Depends(get_logger)):
-    """返回CSS文件"""
-    css_path = os.path.join(os.path.dirname(__file__), "..", "webui.css")
-    if not os.path.exists(css_path):
-        logger.error(f"webui.css not found at {css_path}")
-        raise HTTPException(status_code=404, detail="webui.css not found")
-    return FileResponse(css_path, media_type="text/css")
-
-
-async def get_js(logger: logging.Logger = Depends(get_logger)):
-    """返回JavaScript文件"""
-    js_path = os.path.join(os.path.dirname(__file__), "..", "webui.js")
-    if not os.path.exists(js_path):
-        logger.error(f"webui.js not found at {js_path}")
-        raise HTTPException(status_code=404, detail="webui.js not found")
-    return FileResponse(js_path, media_type="application/javascript")
-
-
-# --- API信息端点 ---
-async def get_api_info(request: Request, current_ai_studio_model_id: str = Depends(get_current_ai_studio_model_id)):
-    """返回API信息"""
-    from api_utils import auth_utils
-
-    server_port = request.url.port or os.environ.get('SERVER_PORT_INFO', '8000')
-    host = request.headers.get('host') or f"127.0.0.1:{server_port}"
-    scheme = request.headers.get('x-forwarded-proto', 'http')
-    base_url = f"{scheme}://{host}"
-    api_base = f"{base_url}/v1"
-    effective_model_name = current_ai_studio_model_id or MODEL_NAME
-
-    api_key_required = bool(auth_utils.API_KEYS)
-    api_key_count = len(auth_utils.API_KEYS)
-
-    if api_key_required:
-        message = f"API Key is required. {api_key_count} valid key(s) configured."
-    else:
-        message = "API Key is not required."
-
-    return JSONResponse(content={
-        "model_name": effective_model_name,
-        "api_base_url": api_base,
-        "server_base_url": base_url,
-        "api_key_required": api_key_required,
-        "api_key_count": api_key_count,
-        "auth_header": "Authorization: Bearer <token> or X-API-Key: <token>" if api_key_required else None,
-        "openai_compatible": True,
-        "supported_auth_methods": ["Authorization: Bearer", "X-API-Key"] if api_key_required else [],
-        "message": message
-    })
-
-
-# --- 健康检查端点 ---
-async def health_check(
-    server_state: Dict[str, Any] = Depends(get_server_state),
-    worker_task = Depends(get_worker_task),
-    request_queue: Queue = Depends(get_request_queue)
-):
-    """健康检查"""
-    is_worker_running = bool(worker_task and not worker_task.done())
-    launch_mode = os.environ.get('LAUNCH_MODE', 'unknown')
-    browser_page_critical = launch_mode != "direct_debug_no_browser"
-    
-    core_ready_conditions = [not server_state["is_initializing"], server_state["is_playwright_ready"]]
-    if browser_page_critical:
-        core_ready_conditions.extend([server_state["is_browser_connected"], server_state["is_page_ready"]])
-    
-    is_core_ready = all(core_ready_conditions)
-    status_val = "OK" if is_core_ready and is_worker_running else "Error"
-    q_size = request_queue.qsize() if request_queue else -1
-    
-    status_message_parts = []
-    if server_state["is_initializing"]: status_message_parts.append("初始化进行中")
-    if not server_state["is_playwright_ready"]: status_message_parts.append("Playwright 未就绪")
-    if browser_page_critical:
-        if not server_state["is_browser_connected"]: status_message_parts.append("浏览器未连接")
-        if not server_state["is_page_ready"]: status_message_parts.append("页面未就绪")
-    if not is_worker_running: status_message_parts.append("Worker 未运行")
-    
-    status = {
-        "status": status_val,
-        "message": "",
-        "details": {**server_state, "workerRunning": is_worker_running, "queueLength": q_size, "launchMode": launch_mode, "browserAndPageCritical": browser_page_critical}
-    }
-    
-    if status_val == "OK":
-        status["message"] = f"服务运行中;队列长度: {q_size}。"
-        return JSONResponse(content=status, status_code=200)
-    else:
-        status["message"] = f"服务不可用;问题: {(', '.join(status_message_parts) or '未知原因')}. 队列长度: {q_size}."
-        return JSONResponse(content=status, status_code=503)
-
-
-# --- 模型列表端点 ---
-async def list_models(
-    logger: logging.Logger = Depends(get_logger),
-    model_list_fetch_event: Event = Depends(get_model_list_fetch_event),
-    page_instance: AsyncPage = Depends(get_page_instance),
-    parsed_model_list: List[Dict[str, Any]] = Depends(get_parsed_model_list),
-    excluded_model_ids: Set[str] = Depends(get_excluded_model_ids)
-):
-    """获取模型列表"""
-    logger.info("[API] 收到 /v1/models 请求。")
-    
-    if not model_list_fetch_event.is_set() and page_instance and not page_instance.is_closed():
-        logger.info("/v1/models: 模型列表事件未设置，尝试刷新页面...")
-        try:
-            await page_instance.reload(wait_until="domcontentloaded", timeout=20000)
-            await asyncio.wait_for(model_list_fetch_event.wait(), timeout=10.0)
-        except Exception as e:
-            logger.error(f"/v1/models: 刷新或等待模型列表时出错: {e}")
-        finally:
-            if not model_list_fetch_event.is_set():
-                model_list_fetch_event.set()
-    
-    if parsed_model_list:
-        final_model_list = [m for m in parsed_model_list if m.get("id") not in excluded_model_ids]
-        return {"object": "list", "data": final_model_list}
-    else:
-        logger.warning("模型列表为空，返回默认后备模型。")
-        return {"object": "list", "data": [{
-            "id": DEFAULT_FALLBACK_MODEL_ID, "object": "model", "created": int(time.time()),
-            "owned_by": "camoufox-proxy-fallback"
-        }]}
-
-
-# --- 聊天完成端点 ---
-async def chat_completions(
-    request: ChatCompletionRequest,
-    http_request: Request,
-    logger: logging.Logger = Depends(get_logger),
-    request_queue: Queue = Depends(get_request_queue),
-    server_state: Dict[str, Any] = Depends(get_server_state),
-    worker_task = Depends(get_worker_task)
-):
-    """处理聊天完成请求"""
-    req_id = ''.join(random.choices('abcdefghijklmnopqrstuvwxyz0123456789', k=7))
-    logger.info(f"[{req_id}] 收到 /v1/chat/completions 请求 (Stream={request.stream})")
-    
-    launch_mode = os.environ.get('LAUNCH_MODE', 'unknown')
-    browser_page_critical = launch_mode != "direct_debug_no_browser"
-    
-    service_unavailable = server_state["is_initializing"] or \
-                          not server_state["is_playwright_ready"] or \
-                          (browser_page_critical and (not server_state["is_page_ready"] or not server_state["is_browser_connected"])) or \
-                          not worker_task or worker_task.done()
-    
-    if service_unavailable:
-        raise HTTPException(status_code=503, detail=f"[{req_id}] 服务当前不可用。请稍后重试。", headers={"Retry-After": "30"})
-    
-    result_future = Future()
-    await request_queue.put({
-        "req_id": req_id, "request_data": request, "http_request": http_request,
-        "result_future": result_future, "enqueue_time": time.time(), "cancelled": False
-    })
-    
-    try:
-        timeout_seconds = RESPONSE_COMPLETION_TIMEOUT / 1000 + 120
-        return await asyncio.wait_for(result_future, timeout=timeout_seconds)
-    except asyncio.TimeoutError:
-        raise HTTPException(status_code=504, detail=f"[{req_id}] 请求处理超时。")
-    except asyncio.CancelledError:
-        raise HTTPException(status_code=499, detail=f"[{req_id}] 请求被客户端取消。")
-    except Exception as e:
-        logger.exception(f"[{req_id}] 等待Worker响应时出错")
-        raise HTTPException(status_code=500, detail=f"[{req_id}] 服务器内部错误: {e}")
-
-
-# --- 取消请求相关 ---
-async def cancel_queued_request(req_id: str, request_queue: Queue, logger: logging.Logger) -> bool:
-    """取消队列中的请求"""
-    items_to_requeue = []
-    found = False
-    try:
-        while not request_queue.empty():
-            item = request_queue.get_nowait()
-            if item.get("req_id") == req_id:
-                logger.info(f"[{req_id}] 在队列中找到请求，标记为已取消。")
-                item["cancelled"] = True
-                if (future := item.get("result_future")) and not future.done():
-                    future.set_exception(HTTPException(status_code=499, detail=f"[{req_id}] Request cancelled."))
-                found = True
-            items_to_requeue.append(item)
-    finally:
-        for item in items_to_requeue:
-            await request_queue.put(item)
-    return found
-
-
-async def cancel_request(
-    req_id: str,
-    logger: logging.Logger = Depends(get_logger),
-    request_queue: Queue = Depends(get_request_queue)
-):
-    """取消请求端点"""
-    logger.info(f"[{req_id}] 收到取消请求。")
-    if await cancel_queued_request(req_id, request_queue, logger):
-        return JSONResponse(content={"success": True, "message": f"Request {req_id} marked as cancelled."})
-    else:
-        return JSONResponse(status_code=404, content={"success": False, "message": f"Request {req_id} not found in queue."})
-
-
-# --- 队列状态端点 ---
-async def get_queue_status(
-    request_queue: Queue = Depends(get_request_queue),
-    processing_lock: Lock = Depends(get_processing_lock)
-):
-    """获取队列状态"""
-    queue_items = list(request_queue._queue)
-    return JSONResponse(content={
-        "queue_length": len(queue_items),
-        "is_processing_locked": processing_lock.locked(),
-        "items": sorted([
-            {
-                "req_id": item.get("req_id", "unknown"),
-                "enqueue_time": item.get("enqueue_time", 0),
-                "wait_time_seconds": round(time.time() - item.get("enqueue_time", 0), 2),
-                "is_streaming": item.get("request_data").stream,
-                "cancelled": item.get("cancelled", False)
-            } for item in queue_items
-        ], key=lambda x: x.get("enqueue_time", 0))
-    })
-
-
-# --- WebSocket日志端点 ---
-async def websocket_log_endpoint(
-    websocket: WebSocket,
-    logger: logging.Logger = Depends(get_logger),
-    log_ws_manager: WebSocketConnectionManager = Depends(get_log_ws_manager)
-):
-    """WebSocket日志端点"""
-    if not log_ws_manager:
-        await websocket.close(code=1011)
-        return
-    
-    client_id = str(uuid.uuid4())
-    try:
-        await log_ws_manager.connect(client_id, websocket)
-        while True:
-            await websocket.receive_text() # Keep connection alive
-    except WebSocketDisconnect:
-        pass
-    except Exception as e:
-        logger.error(f"日志 WebSocket (客户端 {client_id}) 发生异常: {e}", exc_info=True)
-    finally:
-        log_ws_manager.disconnect(client_id)
-
-
-# --- API密钥管理数据模型 ---
-class ApiKeyRequest(BaseModel):
-    key: str
-
-class ApiKeyTestRequest(BaseModel):
-    key: str
-
-
-# --- API密钥管理端点 ---
-async def get_api_keys(logger: logging.Logger = Depends(get_logger)):
-    """获取API密钥列表"""
-    from api_utils import auth_utils
-    try:
-        auth_utils.initialize_keys()
-        keys_info = [{"value": key, "status": "有效"} for key in auth_utils.API_KEYS]
-        return JSONResponse(content={"success": True, "keys": keys_info, "total_count": len(keys_info)})
-    except Exception as e:
-        logger.error(f"获取API密钥列表失败: {e}")
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-async def add_api_key(request: ApiKeyRequest, logger: logging.Logger = Depends(get_logger)):
-    """添加API密钥"""
-    from api_utils import auth_utils
-    key_value = request.key.strip()
-    if not key_value or len(key_value) < 8:
-        raise HTTPException(status_code=400, detail="无效的API密钥格式。")
-    
-    auth_utils.initialize_keys()
-    if key_value in auth_utils.API_KEYS:
-        raise HTTPException(status_code=400, detail="该API密钥已存在。")
-
-    try:
-        key_file_path = os.path.join(os.path.dirname(__file__), "..", "key.txt")
-        with open(key_file_path, 'a+', encoding='utf-8') as f:
-            f.seek(0)
-            if f.read(): f.write("\n")
-            f.write(key_value)
-        
-        auth_utils.initialize_keys()
-        logger.info(f"API密钥已添加: {key_value[:4]}...{key_value[-4:]}")
-        return JSONResponse(content={"success": True, "message": "API密钥添加成功", "key_count": len(auth_utils.API_KEYS)})
-    except Exception as e:
-        logger.error(f"添加API密钥失败: {e}")
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-async def test_api_key(request: ApiKeyTestRequest, logger: logging.Logger = Depends(get_logger)):
-    """测试API密钥"""
-    from api_utils import auth_utils
-    key_value = request.key.strip()
-    if not key_value:
-        raise HTTPException(status_code=400, detail="API密钥不能为空。")
-    
-    auth_utils.initialize_keys()
-    is_valid = auth_utils.verify_api_key(key_value)
-    logger.info(f"API密钥测试: {key_value[:4]}...{key_value[-4:]} - {'有效' if is_valid else '无效'}")
-    return JSONResponse(content={"success": True, "valid": is_valid, "message": "密钥有效" if is_valid else "密钥无效或不存在"})
-
-
-async def delete_api_key(request: ApiKeyRequest, logger: logging.Logger = Depends(get_logger)):
-    """删除API密钥"""
-    from api_utils import auth_utils
-    key_value = request.key.strip()
-    if not key_value:
-        raise HTTPException(status_code=400, detail="API密钥不能为空。")
-
-    auth_utils.initialize_keys()
-    if key_value not in auth_utils.API_KEYS:
-        raise HTTPException(status_code=404, detail="API密钥不存在。")
-
-    try:
-        key_file_path = os.path.join(os.path.dirname(__file__), "..", "key.txt")
-        with open(key_file_path, 'r', encoding='utf-8') as f:
-            lines = f.readlines()
-        
-        with open(key_file_path, 'w', encoding='utf-8') as f:
-            f.writelines(line for line in lines if line.strip() != key_value)
-            
-        auth_utils.initialize_keys()
-        logger.info(f"API密钥已删除: {key_value[:4]}...{key_value[-4:]}")
-        return JSONResponse(content={"success": True, "message": "API密钥删除成功", "key_count": len(auth_utils.API_KEYS)})
-    except Exception as e:
-        logger.error(f"删除API密钥失败: {e}")
-        raise HTTPException(status_code=500, detail=str(e))

api_utils/utils.py

@@ -1,372 +0,0 @@
-"""
-API工具函数模块
-包含SSE生成、流处理、token统计和请求验证等工具函数
-"""
-
-import asyncio
-import json
-import time
-import datetime
-from typing import Any, Dict, List, Optional, AsyncGenerator
-from asyncio import Queue
-from models import Message
-
-
-
-# --- SSE生成函数 ---
-def generate_sse_chunk(delta: str, req_id: str, model: str) -> str:
-    """生成SSE数据块"""
-    chunk_data = {
-        "id": f"chatcmpl-{req_id}",
-        "object": "chat.completion.chunk",
-        "created": int(time.time()),
-        "model": model,
-        "choices": [{"index": 0, "delta": {"content": delta}, "finish_reason": None}]
-    }
-    return f"data: {json.dumps(chunk_data)}\n\n"
-
-
-def generate_sse_stop_chunk(req_id: str, model: str, reason: str = "stop", usage: dict = None) -> str:
-    """生成SSE停止块"""
-    stop_chunk_data = {
-        "id": f"chatcmpl-{req_id}",
-        "object": "chat.completion.chunk",
-        "created": int(time.time()),
-        "model": model,
-        "choices": [{"index": 0, "delta": {}, "finish_reason": reason}]
-    }
-    
-    # 添加usage信息（如果提供）
-    if usage:
-        stop_chunk_data["usage"] = usage
-    
-    return f"data: {json.dumps(stop_chunk_data)}\n\ndata: [DONE]\n\n"
-
-
-def generate_sse_error_chunk(message: str, req_id: str, error_type: str = "server_error") -> str:
-    """生成SSE错误块"""
-    error_chunk = {"error": {"message": message, "type": error_type, "param": None, "code": req_id}}
-    return f"data: {json.dumps(error_chunk)}\n\n"
-
-
-# --- 流处理工具函数 ---
-async def use_stream_response(req_id: str) -> AsyncGenerator[Any, None]:
-    """使用流响应（从服务器的全局队列获取数据）"""
-    from server import STREAM_QUEUE, logger
-    import queue
-    
-    if STREAM_QUEUE is None:
-        logger.warning(f"[{req_id}] STREAM_QUEUE is None, 无法使用流响应")
-        return
-    
-    logger.info(f"[{req_id}] 开始使用流响应")
-    
-    empty_count = 0
-    max_empty_retries = 300  # 30秒超时
-    data_received = False
-    
-    try:
-        while True:
-            try:
-                # 从队列中获取数据
-                data = STREAM_QUEUE.get_nowait()
-                if data is None:  # 结束标志
-                    logger.info(f"[{req_id}] 接收到流结束标志")
-                    break
-                
-                # 重置空计数器
-                empty_count = 0
-                data_received = True
-                logger.debug(f"[{req_id}] 接收到流数据: {type(data)} - {str(data)[:200]}...")
-                
-                # 检查是否是JSON字符串形式的结束标志
-                if isinstance(data, str):
-                    try:
-                        parsed_data = json.loads(data)
-                        if parsed_data.get("done") is True:
-                            logger.info(f"[{req_id}] 接收到JSON格式的完成标志")
-                            yield parsed_data
-                            break
-                        else:
-                            yield parsed_data
-                    except json.JSONDecodeError:
-                        # 如果不是JSON，直接返回字符串
-                        logger.debug(f"[{req_id}] 返回非JSON字符串数据")
-                        yield data
-                else:
-                    # 直接返回数据
-                    yield data
-                    
-                    # 检查字典类型的结束标志
-                    if isinstance(data, dict) and data.get("done") is True:
-                        logger.info(f"[{req_id}] 接收到字典格式的完成标志")
-                        break
-                
-            except (queue.Empty, asyncio.QueueEmpty):
-                empty_count += 1
-                if empty_count % 50 == 0:  # 每5秒记录一次等待状态
-                    logger.info(f"[{req_id}] 等待流数据... ({empty_count}/{max_empty_retries})")
-                
-                if empty_count >= max_empty_retries:
-                    if not data_received:
-                        logger.error(f"[{req_id}] 流响应队列空读取次数达到上限且未收到任何数据，可能是辅助流未启动或出错")
-                    else:
-                        logger.warning(f"[{req_id}] 流响应队列空读取次数达到上限 ({max_empty_retries})，结束读取")
-                    
-                    # 返回超时完成信号，而不是简单退出
-                    yield {"done": True, "reason": "internal_timeout", "body": "", "function": []}
-                    return
-                    
-                await asyncio.sleep(0.1)  # 100ms等待
-                continue
-                
-    except Exception as e:
-        logger.error(f"[{req_id}] 使用流响应时出错: {e}")
-        raise
-    finally:
-        logger.info(f"[{req_id}] 流响应使用完成，数据接收状态: {data_received}")
-
-
-async def clear_stream_queue():
-    """清空流队列（与原始参考文件保持一致）"""
-    from server import STREAM_QUEUE, logger
-    import queue
-
-    if STREAM_QUEUE is None:
-        logger.info("流队列未初始化或已被禁用，跳过清空操作。")
-        return
-
-    while True:
-        try:
-            data_chunk = await asyncio.to_thread(STREAM_QUEUE.get_nowait)
-            # logger.info(f"清空流式队列缓存，丢弃数据: {data_chunk}")
-        except queue.Empty:
-            logger.info("流式队列已清空 (捕获到 queue.Empty)。")
-            break
-        except Exception as e:
-            logger.error(f"清空流式队列时发生意外错误: {e}", exc_info=True)
-            break
-    logger.info("流式队列缓存清空完毕。")
-
-
-# --- Helper response generator ---
-async def use_helper_get_response(helper_endpoint: str, helper_sapisid: str) -> AsyncGenerator[str, None]:
-    """使用Helper服务获取响应的生成器"""
-    from server import logger
-    import aiohttp
-
-    logger.info(f"正在尝试使用Helper端点: {helper_endpoint}")
-
-    try:
-        async with aiohttp.ClientSession() as session:
-            headers = {
-                'Content-Type': 'application/json',
-                'Cookie': f'SAPISID={helper_sapisid}' if helper_sapisid else ''
-            }
-            
-            async with session.get(helper_endpoint, headers=headers) as response:
-                if response.status == 200:
-                    async for chunk in response.content.iter_chunked(1024):
-                        if chunk:
-                            yield chunk.decode('utf-8', errors='ignore')
-                else:
-                    logger.error(f"Helper端点返回错误状态: {response.status}")
-                    
-    except Exception as e:
-        logger.error(f"使用Helper端点时出错: {e}")
-
-
-# --- 请求验证函数 ---
-def validate_chat_request(messages: List[Message], req_id: str) -> Dict[str, Optional[str]]:
-    """验证聊天请求"""
-    from server import logger
-    
-    if not messages:
-        raise ValueError(f"[{req_id}] 无效请求: 'messages' 数组缺失或为空。")
-    
-    if not any(msg.role != 'system' for msg in messages):
-        raise ValueError(f"[{req_id}] 无效请求: 所有消息都是系统消息。至少需要一条用户或助手消息。")
-    
-    # 返回验证结果
-    return {
-        "error": None,
-        "warning": None
-    }
-
-
-# --- 提示准备函数 ---
-def prepare_combined_prompt(messages: List[Message], req_id: str) -> str:
-    """准备组合提示"""
-    from server import logger
-    
-    logger.info(f"[{req_id}] (准备提示) 正在从 {len(messages)} 条消息准备组合提示 (包括历史)。")
-    
-    combined_parts = []
-    system_prompt_content: Optional[str] = None
-    processed_system_message_indices = set()
-    
-    # 处理系统消息
-    for i, msg in enumerate(messages):
-        if msg.role == 'system':
-            content = msg.content
-            if isinstance(content, str) and content.strip():
-                system_prompt_content = content.strip()
-                processed_system_message_indices.add(i)
-                logger.info(f"[{req_id}] (准备提示) 在索引 {i} 找到并使用系统提示: '{system_prompt_content[:80]}...'")
-                system_instr_prefix = "系统指令:\n"
-                combined_parts.append(f"{system_instr_prefix}{system_prompt_content}")
-            else:
-                logger.info(f"[{req_id}] (准备提示) 在索引 {i} 忽略非字符串或空的系统消息。")
-                processed_system_message_indices.add(i)
-            break
-    
-    role_map_ui = {"user": "用户", "assistant": "助手", "system": "系统", "tool": "工具"}
-    turn_separator = "\n---\n"
-    
-    # 处理其他消息
-    for i, msg in enumerate(messages):
-        if i in processed_system_message_indices:
-            continue
-        
-        if msg.role == 'system':
-            logger.info(f"[{req_id}] (准备提示) 跳过在索引 {i} 的后续系统消息。")
-            continue
-        
-        if combined_parts:
-            combined_parts.append(turn_separator)
-        
-        role = msg.role or 'unknown'
-        role_prefix_ui = f"{role_map_ui.get(role, role.capitalize())}:\n"
-        current_turn_parts = [role_prefix_ui]
-        
-        content = msg.content or ''
-        content_str = ""
-        
-        if isinstance(content, str):
-            content_str = content.strip()
-        elif isinstance(content, list):
-            # 处理多模态内容
-            text_parts = []
-            for item in content:
-                if hasattr(item, 'type') and item.type == 'text':
-                    text_parts.append(item.text or '')
-                elif isinstance(item, dict) and item.get('type') == 'text':
-                    text_parts.append(item.get('text', ''))
-                else:
-                    logger.warning(f"[{req_id}] (准备提示) 警告: 在索引 {i} 的消息中忽略非文本或未知类型的 content item")
-            content_str = "\n".join(text_parts).strip()
-        else:
-            logger.warning(f"[{req_id}] (准备提示) 警告: 角色 {role} 在索引 {i} 的内容类型意外 ({type(content)}) 或为 None。")
-            content_str = str(content or "").strip()
-        
-        if content_str:
-            current_turn_parts.append(content_str)
-        
-        # 处理工具调用
-        tool_calls = msg.tool_calls
-        if role == 'assistant' and tool_calls:
-            if content_str:
-                current_turn_parts.append("\n")
-            
-            tool_call_visualizations = []
-            for tool_call in tool_calls:
-                if hasattr(tool_call, 'type') and tool_call.type == 'function':
-                    function_call = tool_call.function
-                    func_name = function_call.name if function_call else None
-                    func_args_str = function_call.arguments if function_call else None
-                    
-                    try:
-                        parsed_args = json.loads(func_args_str if func_args_str else '{}')
-                        formatted_args = json.dumps(parsed_args, indent=2, ensure_ascii=False)
-                    except (json.JSONDecodeError, TypeError):
-                        formatted_args = func_args_str if func_args_str is not None else "{}"
-                    
-                    tool_call_visualizations.append(
-                        f"请求调用函数: {func_name}\n参数:\n{formatted_args}"
-                    )
-            
-            if tool_call_visualizations:
-                current_turn_parts.append("\n".join(tool_call_visualizations))
-        
-        if len(current_turn_parts) > 1 or (role == 'assistant' and tool_calls):
-            combined_parts.append("".join(current_turn_parts))
-        elif not combined_parts and not current_turn_parts:
-            logger.info(f"[{req_id}] (准备提示) 跳过角色 {role} 在索引 {i} 的空消息 (且无工具调用)。")
-        elif len(current_turn_parts) == 1 and not combined_parts:
-            logger.info(f"[{req_id}] (准备提示) 跳过角色 {role} 在索引 {i} 的空消息 (只有前缀)。")
-    
-    final_prompt = "".join(combined_parts)
-    if final_prompt:
-        final_prompt += "\n"
-    
-    preview_text = final_prompt[:300].replace('\n', '\\n')
-    logger.info(f"[{req_id}] (准备提示) 组合提示长度: {len(final_prompt)}。预览: '{preview_text}...'")
-    
-    return final_prompt 
-
-
-def estimate_tokens(text: str) -> int:
-    """
-    估算文本的token数量
-    使用简单的字符计数方法：
-    - 英文：大约4个字符 = 1个token
-    - 中文：大约1.5个字符 = 1个token  
-    - 混合文本：采用加权平均
-    """
-    if not text:
-        return 0
-    
-    # 统计中文字符数量（包括中文标点）
-    chinese_chars = sum(1 for char in text if '\u4e00' <= char <= '\u9fff' or '\u3000' <= char <= '\u303f' or '\uff00' <= char <= '\uffef')
-    
-    # 统计非中文字符数量
-    non_chinese_chars = len(text) - chinese_chars
-    
-    # 计算token估算
-    chinese_tokens = chinese_chars / 1.5  # 中文大约1.5字符/token
-    english_tokens = non_chinese_chars / 4.0  # 英文大约4字符/token
-    
-    return max(1, int(chinese_tokens + english_tokens))
-
-
-def calculate_usage_stats(messages: List[dict], response_content: str, reasoning_content: str = None) -> dict:
-    """
-    计算token使用统计
-    
-    Args:
-        messages: 请求中的消息列表
-        response_content: 响应内容
-        reasoning_content: 推理内容（可选）
-    
-    Returns:
-        包含token使用统计的字典
-    """
-    # 计算输入token（prompt tokens）
-    prompt_text = ""
-    for message in messages:
-        role = message.get("role", "")
-        content = message.get("content", "")
-        prompt_text += f"{role}: {content}\n"
-    
-    prompt_tokens = estimate_tokens(prompt_text)
-    
-    # 计算输出token（completion tokens）
-    completion_text = response_content or ""
-    if reasoning_content:
-        completion_text += reasoning_content
-    
-    completion_tokens = estimate_tokens(completion_text)
-    
-    # 总token数
-    total_tokens = prompt_tokens + completion_tokens
-    
-    return {
-        "prompt_tokens": prompt_tokens,
-        "completion_tokens": completion_tokens,
-        "total_tokens": total_tokens
-    } 
-
-
-def generate_sse_stop_chunk_with_usage(req_id: str, model: str, usage_stats: dict, reason: str = "stop") -> str:
-    """生成带usage统计的SSE停止块"""
-    return generate_sse_stop_chunk(req_id, model, reason, usage_stats)