api_utils/request_processor.py

"""
请求处理器模块
包含核心的请求处理逻辑
"""

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)