"""请求和响应转换工具模块

负责：
1. 将 Gemini 格式的 Payload 转换为 Vcore AI 内部格式
2. 将流式响应聚合成完整的非流式响应
"""

import json
import re
import time
from typing import Any, cast

from src.core.errors import (
    VcoreError,
    InternalError,
    parse_error_response,
)
from src.api.model_config import ModelConfigBuilder
from src.utils.logger import get_logger
from src.utils.string_utils import snake_to_camel, camel_to_snake

logger = get_logger(__name__)

_GEMINI_FUNCTION_NAME_RE = re.compile(r"[^A-Za-z0-9_.-]+")

class RequestTransformer:
    """请求参数转换器"""
    
    def __init__(self, model_builder: ModelConfigBuilder):
        self.model_builder = model_builder

    def build_vcore_payload(
        self,
        model: str,
        gemini_payload: dict[str, Any],
        original_body: dict[str, Any],
        kwargs: dict[str, Any]
    ) -> dict[str, Any]:
        """
        构建 Vcore AI 请求 Payload
        
        Returns:
            new_body
        """
        original_vars: Any = original_body.get('variables', {})
        new_variables: dict[str, Any]
        if hasattr(original_vars, 'model_dump'):
             new_variables = cast(dict[str, Any], original_vars.model_dump())
        elif isinstance(original_vars, dict):
            new_variables = {str(k): v for k, v in cast(dict[Any, Any], original_vars).items()}
        else:
             new_variables = {}

        gemini_payload = self._normalize_gemini_payload(gemini_payload)
        gemini_payload = self._normalize_thought_signature_aliases(gemini_payload)

        target_model = self.model_builder.parse_model_name(model)
        new_variables['model'] = target_model

        # 支持的字段列表（统一使用 camelCase 格式）。尽量覆盖 Gemini generateContent
        # 可透传到 Vcore AI Studio 匿名 GraphQL variables 的字段。
        supported_fields = self._supported_variable_fields()
        canonical_payload = self._canonicalize_supported_fields(gemini_payload, supported_fields)
        
        try:
            from src.core.types import GeminiPayload
            gemini_payload_obj = GeminiPayload.model_validate(canonical_payload)
            dumped_payload = gemini_payload_obj.model_dump(by_alias=True, exclude_none=True)
            
            for field in supported_fields:
                if field in dumped_payload:
                     new_variables[field] = dumped_payload[field]
        except Exception as e:
            logger.debug(f"Pydantic 验证失败，使用基础转换: {e}")
            # 尝试直接从 gemini_payload 透传字段，支持 snake_case 和 camelCase
            for field in supported_fields:
                # 优先使用 camelCase 版本
                if field in canonical_payload:
                    new_variables[field] = canonical_payload[field]
                else:
                    # 尝试 snake_case 版本
                    snake_field = camel_to_snake(field)
                    if snake_field in canonical_payload:
                        new_variables[field] = canonical_payload[snake_field]

        # tools/toolConfig 内部存在大量 snake_case、实验字段和空对象默认值，
        # 使用原始规范化 payload 再走后续专用转换，避免 Pydantic dump 过早丢失未知/None 字段。
        if 'tools' in canonical_payload:
            new_variables['tools'] = canonical_payload['tools']
        if 'toolConfig' in canonical_payload:
            new_variables['toolConfig'] = canonical_payload['toolConfig']

        # 处理 systemInstruction：如果没有 user content，则转换为 user message
        self._handle_system_instruction(new_variables)

        # 特殊处理：contents 格式转换
        if 'contents' in new_variables:
            converted_contents = self._normalize_contents(new_variables['contents'])
            converted_contents = self._handle_inline_data_case(converted_contents)
            converted_contents = self._normalize_contents(converted_contents)
            converted_contents = self._handle_base64_in_contents(converted_contents)
            # 过滤掉空的 parts（Vcore AI 要求每个 content 至少有一个 part）
            converted_contents = self._filter_empty_contents(converted_contents)
            converted_contents = self._ensure_function_call_thought_signatures(converted_contents)
            # 处理 thoughtSignature 字段的 base64 编码
            converted_contents = self._handle_thought_signature(converted_contents)
            new_variables['contents'] = converted_contents
        
        # 特殊处理：tools 格式转换
        if 'tools' in new_variables:
            normalized_tools = self._normalize_tools_format(new_variables['tools'])
            if normalized_tools:
                new_variables['tools'] = normalized_tools
            else:
                # 如果转换结果为空列表，确保移除 tools 字段，同时移除 toolConfig 避免 API 报错
                del new_variables['tools']
                if 'toolConfig' in new_variables:
                    del new_variables['toolConfig']
        
        # 特殊处理：toolConfig 格式转换
        if 'toolConfig' in new_variables:
            normalized_tool_config = self._normalize_tool_config(new_variables['toolConfig'])
            if normalized_tool_config:
                new_variables['toolConfig'] = normalized_tool_config
            else:
                del new_variables['toolConfig']

        # 特殊处理 generationConfig (使用 ModelConfigBuilder 进行格式转换)
        gen_config = self.model_builder.build_generation_config(
            gen_config={},
            gemini_payload=gemini_payload,
            **kwargs
        )
        if gen_config:
            new_variables['generationConfig'] = gen_config
            
        # 特殊处理 safetySettings (如果未提供，则使用默认的宽松设置)
        if 'safetySettings' not in new_variables and 'safety_settings' not in gemini_payload:
            new_variables['safetySettings'] = self.model_builder.build_safety_settings()

        new_body: dict[str, Any] = {
            "querySignature": original_body.get('querySignature'),
            "operationName": original_body.get('operationName'),
            "variables": new_variables
        }
        self._sanitize_vcore_variables_in_place(new_variables)
        
        return new_body

    def _supported_variable_fields(self) -> list[str]:
        """Gemini 下游请求可透传到上游 variables 的字段。"""
        return [
            'contents', 'tools', 'toolConfig', 'systemInstruction',
            'safetySettings', 'generationConfig', 'cachedContent', 'labels',
            # Gemini/Vcore 常见高级字段；上游不支持时会由上游返回明确错误，
            # 这里不主动丢弃，以最大化暴露上游能力。
            'modelArmorConfig', 'cachedContentName', 'requestOptions',
            'session', 'context', 'examples', 'instances', 'parameters',
        ]

    def _canonicalize_supported_fields(self, payload: dict[str, Any], supported_fields: list[str]) -> dict[str, Any]:
        """把下游 Gemini REST/SDK 常见 snake_case 顶层字段并入 camelCase 标准字段。"""
        canonical = payload.copy()
        for field in supported_fields:
            snake_field = camel_to_snake(field)
            if field not in canonical and snake_field in canonical:
                canonical[field] = canonical[snake_field]
        return canonical

    def _normalize_gemini_payload(self, payload: dict[str, Any]) -> dict[str, Any]:
        """兼容 REST、SDK 和部分 OpenAI-like 客户端传来的 Gemini 请求形态。"""
        normalized = payload.copy()
        if 'contents' in normalized:
            normalized['contents'] = self._normalize_contents(normalized['contents'])
        elif 'prompt' in normalized:
            normalized['contents'] = [{"role": "user", "parts": [{"text": str(normalized['prompt'])}]}]
        return normalized

    def _normalize_thought_signature_aliases(self, data: Any) -> Any:
        """递归兼容 thought_signature / thoughtSignature 两种字段名。"""
        if isinstance(data, list):
            return [self._normalize_thought_signature_aliases(item) for item in cast(list[Any], data)]
        if isinstance(data, dict):
            data_dict = cast(dict[str, Any], data)
            normalized: dict[str, Any] = {}
            for key, value in data_dict.items():
                normalized_value = self._normalize_thought_signature_aliases(value) if isinstance(value, (dict, list)) else value
                if key == 'thought_signature':
                    normalized['thoughtSignature'] = normalized_value
                elif key == 'thoughtSignature':
                    normalized['thoughtSignature'] = normalized_value
                else:
                    normalized[key] = normalized_value
            return normalized
        return data

    def _normalize_contents(self, contents: Any) -> Any:
        if contents is None:
            return []
        if isinstance(contents, str):
            return [{"role": "user", "parts": [{"text": contents}]}]
        if isinstance(contents, dict):
            return [self._normalize_content(contents)]
        if isinstance(contents, list):
            normalized: list[Any] = []
            pending_text_parts: list[dict[str, Any]] = []
            for item in cast(list[Any], contents):
                if isinstance(item, str):
                    pending_text_parts.append({"text": item})
                elif isinstance(item, dict):
                    if pending_text_parts:
                        normalized.append({"role": "user", "parts": pending_text_parts})
                        pending_text_parts = []
                    normalized.append(self._normalize_content(cast(dict[str, Any], item)))
            if pending_text_parts:
                normalized.append({"role": "user", "parts": pending_text_parts})
            return normalized
        return contents

    def _normalize_content(self, content: dict[str, Any]) -> dict[str, Any]:
        normalized = content.copy()
        source_role = normalized.get('role')
        if source_role in {'tool', 'function'} and 'parts' not in normalized and ('content' in normalized or 'response' in normalized):
            raw_response = normalized.get('response', normalized.get('content', {}))
            normalized['parts'] = [{
                "functionResponse": {
                    "name": str(normalized.get('name') or normalized.get('function_name') or ""),
                    "response": self._coerce_function_response(raw_response),
                }
            }]
            normalized.pop('content', None)
            normalized.pop('response', None)
        elif 'content' in normalized and 'parts' not in normalized:
            normalized['parts'] = self._normalize_parts(normalized.get('content'))
            normalized.pop('content', None)
        elif 'parts' in normalized:
            normalized['parts'] = self._normalize_parts(normalized.get('parts'))
        elif 'text' in normalized:
            normalized['parts'] = [{"text": str(normalized.pop('text'))}]
        else:
            normalized.setdefault('parts', [])

        if source_role in {'assistant', 'model'}:
            extra_parts = self._openai_tool_history_parts(normalized)
            if extra_parts:
                normalized['parts'] = list(cast(list[Any], normalized.get('parts') or [])) + extra_parts
                normalized.pop('tool_calls', None)
                normalized.pop('function_call', None)
                normalized.pop('functionCall', None)

        role = normalized.get('role')
        if role == 'assistant':
            normalized['role'] = 'model'
        elif role == 'tool':
            normalized['role'] = 'function'
        elif not role:
            normalized['role'] = 'user'
        return normalized

    def _normalize_parts(self, parts: Any) -> list[dict[str, Any]]:
        if parts is None:
            return []
        if isinstance(parts, str):
            return [{"text": parts}]
        if isinstance(parts, dict):
            return [self._normalize_part(parts)]
        if isinstance(parts, list):
            normalized: list[dict[str, Any]] = []
            for part in cast(list[Any], parts):
                if isinstance(part, str):
                    normalized.append({"text": part})
                elif isinstance(part, dict):
                    normalized_part = self._normalize_part(cast(dict[str, Any], part))
                    if normalized_part:
                        normalized.append(normalized_part)
            return normalized
        return [{"text": str(parts)}]

    def _openai_tool_history_parts(self, message: dict[str, Any]) -> list[dict[str, Any]]:
        """兼容下游把 OpenAI tool_calls/function_call 混入 Gemini endpoint 的历史。"""
        parts: list[dict[str, Any]] = []
        tool_calls = message.get('tool_calls') or message.get('toolCalls')
        if isinstance(tool_calls, list):
            for tool_call in cast(list[Any], tool_calls):
                part = self._openai_tool_call_to_part(tool_call)
                if part:
                    parts.append(part)

        function_call = message.get('function_call') or message.get('functionCall')
        if isinstance(function_call, dict):
            part = self._openai_tool_call_to_part({"function": function_call})
            if part:
                parts.append(part)
        return parts

    def _openai_tool_call_to_part(self, tool_call: Any) -> dict[str, Any] | None:
        if not isinstance(tool_call, dict):
            return None
        tool_call_dict = cast(dict[str, Any], tool_call)
        func = tool_call_dict.get('function')
        if isinstance(func, dict):
            func_dict = cast(dict[str, Any], func)
            name = func_dict.get('name') or tool_call_dict.get('name') or tool_call_dict.get('functionName')
            args = func_dict.get('arguments', tool_call_dict.get('arguments', tool_call_dict.get('args', {})))
            thought_signature = (
                tool_call_dict.get('thoughtSignature')
                or tool_call_dict.get('thought_signature')
                or func_dict.get('thoughtSignature')
                or func_dict.get('thought_signature')
            )
        else:
            name = tool_call_dict.get('name') or tool_call_dict.get('function_name') or tool_call_dict.get('functionName')
            args = tool_call_dict.get('arguments', tool_call_dict.get('args', {}))
            thought_signature = tool_call_dict.get('thoughtSignature') or tool_call_dict.get('thought_signature')

        if not name:
            return None
        part: dict[str, Any] = {
            "functionCall": {
                "name": self._sanitize_function_name(str(name)),
                "args": self._coerce_function_args(args),
            }
        }
        if thought_signature:
            part["thoughtSignature"] = thought_signature
        return part

    def _coerce_function_args(self, args: Any) -> dict[str, Any]:
        if isinstance(args, dict):
            return cast(dict[str, Any], args)
        if isinstance(args, str):
            try:
                parsed = json.loads(args)
                return parsed if isinstance(parsed, dict) else {"value": parsed}
            except json.JSONDecodeError:
                return {"raw": args}
        if args is None:
            return {}
        return {"value": args}

    def _coerce_function_response(self, response: Any) -> dict[str, Any]:
        if isinstance(response, dict):
            return cast(dict[str, Any], response)
        if isinstance(response, str):
            try:
                parsed = json.loads(response)
                return parsed if isinstance(parsed, dict) else {"result": parsed}
            except json.JSONDecodeError:
                return {"result": response}
        if response is None:
            return {}
        return {"result": response}

    def _normalize_part(self, part: dict[str, Any]) -> dict[str, Any]:
        part_type = part.get('type')
        if part_type in {'text', 'input_text'}:
            return {"text": str(part.get('text', ''))}
        if part_type in {'image_url', 'input_image'}:
            url_obj = part.get('image_url') or part.get('input_image') or {}
            url = url_obj.get('url') if isinstance(url_obj, dict) else url_obj
            if isinstance(url, str) and url.startswith('data:'):
                mime, data = self._parse_data_uri(url)
                if mime and data:
                    return {"inlineData": {"mimeType": mime, "data": data}}
            if isinstance(url, str) and url.startswith(('http://', 'https://', 'gs://')):
                return {"fileData": {"mimeType": self._guess_mime_from_uri(url), "fileUri": url}}
        if part_type in {'media', 'file', 'file_data'}:
            file_uri = part.get('fileUri') or part.get('file_uri') or part.get('uri') or part.get('url')
            mime_type = part.get('mimeType') or part.get('mime_type') or self._guess_mime_from_uri(str(file_uri or ''))
            if file_uri:
                return {"fileData": {"mimeType": str(mime_type), "fileUri": str(file_uri)}}
        if part_type in {'inline_data', 'inlineData'}:
            inline = part.get('inlineData') or part.get('inline_data') or part
            if isinstance(inline, dict):
                data = inline.get('data')
                mime_type = inline.get('mimeType') or inline.get('mime_type') or part.get('mimeType') or part.get('mime_type')
                if data and mime_type:
                    return {"inlineData": {"mimeType": str(mime_type), "data": str(data)}}

        normalized: dict[str, Any] = {}
        for k, v in part.items():
            if k == 'type':
                continue
            normalized[snake_to_camel(k)] = v
        return normalized

    def _parse_data_uri(self, uri: str) -> tuple[str, str]:
        try:
            header, data = uri.split(',', 1)
            mime = header.split(':', 1)[1].split(';', 1)[0]
            return mime, data
        except (ValueError, IndexError):
            return "", ""

    def _guess_mime_from_uri(self, uri: str) -> str:
        lower = uri.lower().split('?', 1)[0].split('#', 1)[0]
        if lower.endswith(('.jpg', '.jpeg')):
            return 'image/jpeg'
        if lower.endswith('.webp'):
            return 'image/webp'
        if lower.endswith('.gif'):
            return 'image/gif'
        if lower.endswith('.png'):
            return 'image/png'
        if lower.endswith('.mp4'):
            return 'video/mp4'
        if lower.endswith('.mov'):
            return 'video/quicktime'
        if lower.endswith('.webm'):
            return 'video/webm'
        if lower.endswith('.mp3'):
            return 'audio/mpeg'
        if lower.endswith('.wav'):
            return 'audio/wav'
        if lower.endswith('.ogg'):
            return 'audio/ogg'
        if lower.endswith('.pdf'):
            return 'application/pdf'
        if lower.endswith('.txt'):
            return 'text/plain'
        return 'image/png'

    def _convert_tools_format(self, data: Any) -> Any:
        """专门处理工具格式转换，统一转换为 camelCase"""
        if isinstance(data, dict):
            new_dict: dict[str, Any] = {}
            data_dict: dict[str, Any] = cast(dict[str, Any], data)
            for k, v in data_dict.items():
                camel_k = snake_to_camel(k) if '_' in k else k
                # 转换 function_declarations 为 functionDeclarations
                if k in ['function_declarations', 'functionDeclarations']:
                    new_dict['functionDeclarations'] = self._convert_tools_format(v)
                elif k in ['google_search', 'googleSearch']:
                    new_dict['googleSearch'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
                elif k in ['google_search_retrieval', 'googleSearchRetrieval']:
                    new_dict['googleSearchRetrieval'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
                elif k in ['code_execution', 'codeExecution']:
                    new_dict['codeExecution'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
                elif k in ['url_context', 'urlContext']:
                    new_dict['urlContext'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
                elif k in ['function_calling_config', 'functionCallingConfig']:
                    new_dict['functionCallingConfig'] = self._convert_tools_format(v)
                elif k in ['allowed_function_names', 'allowedFunctionNames']:
                    new_dict['allowedFunctionNames'] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
                elif camel_k == "parametersJsonSchema" and isinstance(v, dict):
                    # parametersJsonSchema 需要特殊处理，确保 properties 和 required 字段一致
                    new_dict['parametersJsonSchema'] = self._convert_parameters_schema(cast(dict[str, Any], v))
                elif k == "parameters" and isinstance(v, dict):
                    # Schema 对象需要特殊处理
                    converted_v = v.copy() if isinstance(v, dict) else v
                    new_dict[k] = self._ensure_function_parameters_schema(self._to_native_schema(converted_v))
                elif k == "input_schema" and isinstance(v, dict):
                    new_dict['parameters'] = self._ensure_function_parameters_schema(self._to_native_schema(cast(dict[str, Any], v)))
                elif k == "inputSchema" and isinstance(v, dict):
                    new_dict['parameters'] = self._ensure_function_parameters_schema(self._to_native_schema(cast(dict[str, Any], v)))
                elif k == "name" and not v:  # Vcore AI Function name cannot be empty
                    continue
                elif k == "name" and v:
                    new_dict[k] = self._sanitize_function_name(str(v))
                else:
                    # 对于其他字段，转换为 camelCase（除了特殊字段）
                    new_dict[camel_k] = self._convert_tools_format(v) if isinstance(v, (dict, list)) else v
            return new_dict
        elif isinstance(data, list):
            return [self._convert_tools_format(item) for item in cast(list[Any], data)]
        else:
            return data

    def _convert_parameters_schema(self, schema: dict[str, Any]) -> dict[str, Any]:
        """
        转换 parametersJsonSchema，确保 properties 和 required 字段中的参数名一致
        统一使用 snake_case 格式，避免 camelCase 和 snake_case 混用导致的不匹配
        """
        new_schema: dict[str, Any] = schema.copy()
        unsupported_keys = {
            '$schema', '$id', '$defs', 'definitions', 'additionalProperties',
            'patternProperties', 'unevaluatedProperties', 'dependentSchemas',
            'if', 'then', 'else', 'allOf', 'anyOf', 'oneOf', 'not',
            'examples', 'default', 'nullable'
        }
        for key in unsupported_keys:
            new_schema.pop(key, None)

        if isinstance(new_schema.get('type'), list):
            non_null_types = [item for item in cast(list[Any], new_schema['type']) if item != 'null']
            new_schema['type'] = non_null_types[0] if non_null_types else 'string'
        
        # 处理 properties 字段：将 camelCase 转换为 snake_case
        if 'properties' in new_schema and isinstance(new_schema['properties'], dict):
            old_properties: dict[str, Any] = cast(dict[str, Any], new_schema['properties'])
            new_properties: dict[str, Any] = {}
            
            for prop_name, prop_def in old_properties.items():
                # 将 camelCase 转换为 snake_case
                snake_name = camel_to_snake(str(prop_name))
                new_properties[snake_name] = prop_def
                
                # 递归处理嵌套的 schema
                if isinstance(prop_def, dict):
                    new_properties[snake_name] = self._convert_parameters_schema(cast(dict[str, Any], prop_def))
            
            new_schema['properties'] = new_properties
        
        # 处理 required 字段：确保使用 snake_case（通常已经是正确的）
        if 'required' in new_schema and isinstance(new_schema['required'], list):
            # required 字段通常已经是 snake_case，但为了保险起见，也进行转换
            new_required: list[Any] = []
            for req_name in cast(list[Any], new_schema['required']):
                if isinstance(req_name, str):
                    snake_name = camel_to_snake(req_name)
                    new_required.append(snake_name)
                else:
                    new_required.append(req_name)
            new_schema['required'] = new_required
        
        # 处理其他可能包含 schema 的字段
        for key, value in list(new_schema.items()):
            if key not in ['properties', 'required'] and isinstance(value, dict):
                new_schema[key] = self._convert_parameters_schema(cast(dict[str, Any], value))
        
        return new_schema

    def _sanitize_function_name(self, name: str) -> str:
        cleaned = _GEMINI_FUNCTION_NAME_RE.sub("_", name.strip())[:64].strip("._-")
        if cleaned and not (cleaned[0].isalpha() or cleaned[0] == "_"):
            cleaned = f"tool_{cleaned}"[:64]
        return cleaned or "tool"

    def _ensure_function_parameters_schema(self, schema: Any) -> dict[str, Any]:
        if not isinstance(schema, dict):
            return {"type": "OBJECT", "properties": []}
        ensured = cast(dict[str, Any], schema).copy()
        schema_type = ensured.get('type')
        if not schema_type:
            ensured['type'] = 'OBJECT'
        elif isinstance(schema_type, str):
            ensured['type'] = schema_type.upper()
        if ensured.get('type') == 'OBJECT' and 'properties' not in ensured:
            ensured['properties'] = []
        return ensured

    def _to_native_schema(self, standard_schema: dict[str, Any]) -> dict[str, Any]:
        """
        将标准 JSON Schema 转换为 Vcore AI 原生 Map-style Schema
        
        Args:
            standard_schema: 标准 JSON Schema 对象
            
        Returns:
            Vcore AI 原生 Schema
        """
        
        native_schema = standard_schema.copy()
        unsupported_keys = {
            '$schema', '$id', '$defs', 'definitions', 'additionalProperties',
            'patternProperties', 'unevaluatedProperties', 'dependentSchemas',
            'if', 'then', 'else', 'allOf', 'anyOf', 'oneOf', 'not',
            'examples', 'default', 'nullable'
        }
        for key in unsupported_keys:
            native_schema.pop(key, None)
        
        # Vcore AI 要求类型必须是大写 (例如: STRING, OBJECT, INTEGER)
        if 'type' in native_schema and isinstance(native_schema['type'], list):
            non_null_types = [item for item in cast(list[Any], native_schema['type']) if item != 'null']
            native_schema['type'] = non_null_types[0] if non_null_types else 'string'

        if 'type' in native_schema and isinstance(native_schema['type'], str):
            native_schema['type'] = native_schema['type'].upper()
            
        if 'properties' in native_schema and isinstance(native_schema['properties'], dict):
            native_props: list[dict[str, str | dict[str, Any]]] = []
            props_dict = cast(dict[str, Any], native_schema['properties'])
            for key, value in props_dict.items():
                # 递归处理嵌套对象
                if isinstance(value, dict):
                    converted_value = self._to_native_schema(cast(dict[str, Any], value))
                else:
                    converted_value = {}
                
                native_props.append({
                    "key": str(key),
                    "value": converted_value
                })
            native_schema['properties'] = native_props
        
        # 处理数组项
        if 'items' in native_schema and isinstance(native_schema['items'], dict):
            items_dict = cast(dict[str, Any], native_schema['items'])
            native_schema['items'] = self._to_native_schema(items_dict)
            
        return native_schema
    
    def _handle_system_instruction(self, new_variables: dict[str, Any]) -> None:
        """处理 systemInstruction：如果没有 user content，则转换为 user message"""
        system_instruction_content = new_variables.get('systemInstruction')
        if not system_instruction_content:
            return
            
        contents = new_variables.get('contents', [])
        
        # 检查是否已有 user 角色
        contents_list: list[Any] = cast(list[Any], contents) if isinstance(contents, list) else []
        has_user_role = any(
            isinstance(content, dict) and cast(dict[str, Any], content).get('role') == 'user'
            for content in contents_list
        )
        
        if has_user_role:
            return
            
        # 提取文本内容
        text_from_system = self._extract_text_from_instruction(system_instruction_content)
        if not text_from_system:
            return
            
        # 转换为 user message
        user_message = {
            'role': 'user',
            'parts': [{'text': text_from_system}]
        }
        
        # 显式转换 contents 为 list[Any] 以修复 pylance 报错
        new_contents: list[Any] = list(contents_list)
        new_contents.insert(0, user_message)
        new_variables['contents'] = new_contents
        del new_variables['systemInstruction']
    
    def _extract_text_from_instruction(self, instruction: Any) -> str:
        """从 system instruction 中提取文本内容"""
        if isinstance(instruction, str):
            return instruction
        elif isinstance(instruction, dict):
            instruction_dict = cast(dict[str, Any], instruction)
            parts = instruction_dict.get('parts', [])
            if isinstance(parts, list):
                text_parts = []
                for part in parts:
                    if isinstance(part, dict) and 'text' in part:
                        text_parts.append(str(part['text']))
                return "".join(text_parts)
        return ""
    
    def _normalize_tools_format(self, tools: Any) -> list[dict[str, Any]]:
        """标准化 tools 格式为 Vcore AI 期望的格式 (List[Tool])"""
        converted_tools: Any = self._convert_tools_format(tools)
        tool_keys = {
            'functionDeclarations', 'googleSearch', 'googleSearchRetrieval',
            'codeExecution', 'retrieval', 'urlContext'
        }
        
        if isinstance(converted_tools, dict):
            # 如果是字典，且包含 functionDeclarations，将其包裹在列表中
            if any(key in converted_tools for key in tool_keys):
                normalized_tool = self._normalize_single_tool(cast(dict[str, Any], converted_tools), tool_keys)
                return [normalized_tool] if normalized_tool else []
            # 如果是单个 FunctionDeclaration，包裹成 Tool 再包裹在列表中
            if 'name' in converted_tools:
                decl = self._normalize_function_declaration(cast(dict[str, Any], converted_tools))
                return [{"functionDeclarations": [decl]}] if decl else []
            return []
            
        if not isinstance(converted_tools, list) or len(cast(list[Any], converted_tools)) == 0:
            return []
            
        converted_tools_list: list[Any] = cast(list[Any], converted_tools)
        normalized_tools: list[dict[str, Any]] = []
        function_decls: list[dict[str, Any]] = []
        for item in converted_tools_list:
            if not isinstance(item, dict):
                continue
            item_dict = cast(dict[str, Any], item)
            if any(key in item_dict for key in tool_keys):
                normalized_tool = self._normalize_single_tool(item_dict, tool_keys)
                if normalized_tool:
                    normalized_tools.append(normalized_tool)
            elif item_dict.get('name'):
                decl = self._normalize_function_declaration(item_dict)
                if decl:
                    function_decls.append(decl)

        if function_decls:
            normalized_tools.insert(0, {"functionDeclarations": function_decls})

        return normalized_tools

    def _normalize_single_tool(self, tool: dict[str, Any], tool_keys: set[str]) -> dict[str, Any] | None:
        normalized = {k: v for k, v in tool.items() if k in tool_keys or k.startswith('x')}

        func_decls = tool.get('functionDeclarations')
        if isinstance(func_decls, list):
            declarations: list[dict[str, Any]] = []
            for decl in cast(list[Any], func_decls):
                if isinstance(decl, dict):
                    normalized_decl = self._normalize_function_declaration(cast(dict[str, Any], decl))
                    if normalized_decl:
                        declarations.append(normalized_decl)
            if declarations:
                normalized['functionDeclarations'] = declarations
            else:
                normalized.pop('functionDeclarations', None)

        for native_key in ('googleSearch', 'googleSearchRetrieval', 'codeExecution', 'retrieval', 'urlContext'):
            if native_key in tool:
                native_value = tool.get(native_key)
                normalized[native_key] = native_value if isinstance(native_value, dict) else {}

        return normalized if any(key in normalized for key in tool_keys) else None

    def _normalize_function_declaration(self, declaration: dict[str, Any]) -> dict[str, Any] | None:
        raw_name = declaration.get('name') or declaration.get('function') or declaration.get('functionName')
        if not raw_name:
            return None
        normalized = declaration.copy()
        normalized['name'] = self._sanitize_function_name(str(raw_name))

        if 'parameters' in normalized:
            normalized['parameters'] = self._ensure_function_parameters_schema(normalized['parameters'])
        elif 'parametersJsonSchema' in normalized and isinstance(normalized['parametersJsonSchema'], dict):
            normalized['parametersJsonSchema'] = self._convert_parameters_schema(cast(dict[str, Any], normalized['parametersJsonSchema']))
        else:
            normalized['parameters'] = {"type": "OBJECT", "properties": []}

        return normalized

    def _normalize_tool_config(self, tool_config: Any) -> Any:
        converted = self._convert_tools_format(tool_config)
        if not isinstance(converted, dict):
            return converted
        config = cast(dict[str, Any], converted)
        fcc = config.get('functionCallingConfig')
        if isinstance(fcc, dict):
            fcc_dict = cast(dict[str, Any], fcc).copy()
            mode = fcc_dict.get('mode')
            if isinstance(mode, str):
                mode_upper = mode.upper()
                if mode_upper in {'AUTO', 'ANY', 'NONE', 'MODE_UNSPECIFIED'}:
                    fcc_dict['mode'] = mode_upper
            allowed = fcc_dict.get('allowedFunctionNames')
            if isinstance(allowed, list):
                fcc_dict['allowedFunctionNames'] = [
                    self._sanitize_function_name(str(name))
                    for name in cast(list[Any], allowed)
                    if name is not None and str(name).strip()
                ]
            if not fcc_dict.get('allowedFunctionNames'):
                fcc_dict.pop('allowedFunctionNames', None)
            config['functionCallingConfig'] = fcc_dict
        if not config.get('functionCallingConfig'):
            config.pop('functionCallingConfig', None)
        return config

    def _handle_inline_data_case(self, contents: Any) -> Any:
        """
        递归处理 contents，确保 inlineData/mimeType 字段名正确 (适配各种客户端传参)
        """
        if isinstance(contents, list):
            return [self._handle_inline_data_case(item) for item in cast(list[Any], contents)]
        if isinstance(contents, dict):
            new_dict: dict[str, Any] = {}
            for k, v in cast(dict[str, Any], contents).items():
                camel_k = snake_to_camel(k)
                if k == 'thought_signature':
                    new_dict['thoughtSignature'] = self._handle_inline_data_case(v)
                    continue
                if camel_k == 'inlineData' and isinstance(v, dict):
                    v_dict = cast(dict[str, Any], v)
                    new_inline_data = {}
                    for ik, iv in v_dict.items():
                        camel_ik = snake_to_camel(ik)
                        new_inline_data[camel_ik] = iv
                    new_dict['inlineData'] = new_inline_data
                elif camel_k == 'fileData' and isinstance(v, dict):
                    v_dict = cast(dict[str, Any], v)
                    new_file_data = {}
                    for ik, iv in v_dict.items():
                        camel_ik = snake_to_camel(ik)
                        new_file_data[camel_ik] = iv
                    new_dict['fileData'] = new_file_data
                elif camel_k == 'functionCall' and isinstance(v, dict):
                    func_call = self._convert_tools_format(v)
                    if isinstance(func_call, dict):
                        new_dict['functionCall'] = func_call
                elif camel_k == 'functionResponse' and isinstance(v, dict):
                    func_response = self._convert_tools_format(v)
                    if isinstance(func_response, dict):
                        new_dict['functionResponse'] = func_response
                else:
                    new_dict[camel_k] = self._handle_inline_data_case(v)
            return new_dict
        return contents

    def _handle_base64_in_contents(self, contents: Any) -> Any:
        """
        递归处理 contents 中的 base64 数据。
        将 URL-safe Base64 转换为标准 Base64 并补全 padding。
        """
        try:
            if isinstance(contents, list):
                res_list: list[Any] = [self._handle_base64_in_contents(item) for item in cast(list[Any], contents)]
                return cast(Any, res_list)
            if isinstance(contents, dict):
                new_dict: dict[str, Any] = {}
                for k, v in cast(dict[str, Any], contents).items():
                    if k == 'inlineData' and isinstance(v, dict):
                        v_dict = cast(dict[str, Any], v)
                        if 'data' in v_dict and isinstance(v_dict['data'], str):
                            try:
                                b64_data: str = v_dict['data']
                                b64_data = b64_data.replace('-', '+').replace('_', '/')
                                padding = len(b64_data) % 4
                                if padding:
                                    b64_data += '=' * (4 - padding)
                                
                                new_inline_data = v_dict.copy()
                                new_inline_data['data'] = b64_data
                                new_dict[k] = new_inline_data
                                continue
                            except Exception:
                                pass
                    new_dict[k] = self._handle_base64_in_contents(v)
                return cast(Any, new_dict)
            return contents
        except Exception as e:
            logger.warning(f"Base64 内容处理失败: {e}")
            return cast(Any, contents)

    def _filter_empty_contents(self, contents: Any) -> Any:
        """
        过滤掉空的 contents（parts 为空数组的消息）
        Vcore AI 要求每个 content 至少包含一个 part
        """
        if not isinstance(contents, list):
            return contents
        
        filtered_contents: list[Any] = []
        contents_list: list[Any] = cast(list[Any], contents)
        
        # 收集所有 functionCall 的名称，用于修复 functionResponse
        function_call_names: list[str] = []
        for content in contents_list:
            if isinstance(content, dict):
                content_dict = cast(dict[str, Any], content)
                parts = content_dict.get('parts', [])
                if isinstance(parts, list):
                    for part in cast(list[Any], parts):
                        if isinstance(part, dict):
                            part_dict = cast(dict[str, Any], part)
                            func_call = part_dict.get('functionCall')
                            if not isinstance(func_call, dict):
                                func_call = part_dict.get('function_call')
                            if isinstance(func_call, dict):
                                func_call_dict = cast(dict[str, Any], func_call)
                                name = func_call_dict.get('name')
                                if name and isinstance(name, str):
                                    function_call_names.append(self._sanitize_function_name(name))
        
        for content in contents_list:
            if isinstance(content, dict):
                content_dict: dict[str, Any] = cast(dict[str, Any], content)
                parts = content_dict.get('parts', [])
                # 只保留有 parts 且 parts 不为空的 content
                if isinstance(parts, list) and len(cast(list[Any], parts)) > 0:
                    parts_list: list[Any] = cast(list[Any], parts)
                    # 过滤并验证 parts 中的有效内容
                    valid_parts: list[Any] = []
                    for part in parts_list:
                        if isinstance(part, dict):
                            part_dict = cast(dict[str, Any], part)
                            
                            # 清理并修复 part
                            cleaned_part = self._clean_part_metadata(part_dict, function_call_names)
                            if cleaned_part:
                                valid_parts.append(cleaned_part)
                    
                    if valid_parts:
                        # 更新 content 的 parts
                        filtered_content = content_dict.copy()
                        filtered_content['parts'] = valid_parts
                        filtered_contents.append(filtered_content)
                    else:
                        logger.debug(f"过滤掉空的 content: role={content_dict.get('role', 'unknown')}")
                else:
                    logger.debug(f"过滤掉空的 content: role={content_dict.get('role', 'unknown')}")
            else:
                # 非 Dict 类型的 content，保留
                filtered_contents.append(content)
        
        return filtered_contents

    def _clean_part_metadata(self, part_dict: dict[str, Any], function_call_names: list[str]) -> dict[str, Any] | None:
        """
        清理 part 中的空元数据字段，修复无效的 functionResponse
        
        Args:
            part_dict: 原始 part 字典
            function_call_names: 可用的函数调用名称列表
            
        Returns:
            清理后的 part 字典，如果 part 无效则返回 None
        """
        cleaned_part: dict[str, Any] = {}
        has_valid_content = False
        
        # 处理文本内容
        if 'text' in part_dict:
            text_value = part_dict['text']
            if text_value is not None and str(text_value) != "":
                cleaned_part['text'] = text_value
                has_valid_content = True
        
        # 处理思考标记
        if 'thought' in part_dict:
            cleaned_part['thought'] = part_dict['thought']
        
        # 处理思考签名 (thoughtSignature)
        if 'thoughtSignature' in part_dict:
            cleaned_part['thoughtSignature'] = part_dict['thoughtSignature']
        elif 'thought_signature' in part_dict:
            cleaned_part['thoughtSignature'] = part_dict['thought_signature']
        
        # 处理函数调用
        if 'functionCall' in part_dict or 'function_call' in part_dict:
            func_call = part_dict.get('functionCall') or part_dict.get('function_call')
            if isinstance(func_call, dict):
                func_call_dict = cast(dict[str, Any], func_call)
                if func_call_dict.get('name'):  # 只保留有名称的函数调用
                    fixed_func_call = func_call_dict.copy()
                    fixed_func_call['name'] = self._sanitize_function_name(str(fixed_func_call['name']))
                    if 'args' not in fixed_func_call or fixed_func_call.get('args') is None:
                        fixed_func_call['args'] = {}
                    if isinstance(fixed_func_call.get('args'), str):
                        try:
                            fixed_func_call['args'] = json.loads(cast(str, fixed_func_call['args']))
                        except json.JSONDecodeError:
                            fixed_func_call['args'] = {"raw": fixed_func_call['args']}
                    if not isinstance(fixed_func_call.get('args'), dict):
                        fixed_func_call['args'] = {"value": fixed_func_call.get('args')}
                    cleaned_part['functionCall'] = fixed_func_call
                    has_valid_content = True
        
        # 处理函数响应
        if 'functionResponse' in part_dict or 'function_response' in part_dict:
            func_response = part_dict.get('functionResponse') or part_dict.get('function_response')
            if isinstance(func_response, dict):
                func_response_dict = cast(dict[str, Any], func_response)
                current_name = (
                    func_response_dict.get('name')
                    or func_response_dict.get('functionName')
                    or func_response_dict.get('function_name')
                )
                
                # 如果 name 为空，尝试修复
                if not current_name and function_call_names:
                    inferred_name = function_call_names[-1]  # 使用最后一个 functionCall 的名称
                    logger.warning(f"修复空的 functionResponse.name，推断为: {inferred_name}")
                    
                    fixed_func_response = func_response_dict.copy()
                    fixed_func_response['name'] = self._sanitize_function_name(inferred_name)
                    fixed_func_response['response'] = self._coerce_function_response(fixed_func_response.get('response', {}))
                    cleaned_part['functionResponse'] = fixed_func_response
                    has_valid_content = True
                elif current_name:
                    # name 不为空，直接保留
                    fixed_func_response = func_response_dict.copy()
                    fixed_func_response['name'] = self._sanitize_function_name(str(current_name))
                    fixed_func_response['response'] = self._coerce_function_response(fixed_func_response.get('response', {}))
                    cleaned_part['functionResponse'] = fixed_func_response
                    has_valid_content = True
                # 如果 name 为空且无法推断，则丢弃这个 functionResponse
        
        # 处理内联数据
        if 'inlineData' in part_dict:
            inline_data = part_dict['inlineData']
            if isinstance(inline_data, dict):
                inline_data_dict = cast(dict[str, Any], inline_data)
                # 只保留有实际数据的 inlineData
                if (inline_data_dict.get('data') and
                    str(inline_data_dict['data']).strip() and
                    inline_data_dict.get('mimeType') and
                    str(inline_data_dict['mimeType']).strip()):
                    cleaned_part['inlineData'] = inline_data
                    has_valid_content = True
        
        # 处理文件数据
        if 'fileData' in part_dict:
            file_data = part_dict['fileData']
            if isinstance(file_data, dict):
                file_data_dict = cast(dict[str, Any], file_data)
                # 只保留有实际数据的 fileData
                if (file_data_dict.get('fileUri') and
                    str(file_data_dict['fileUri']).strip() and
                    file_data_dict.get('mimeType') and
                    str(file_data_dict['mimeType']).strip()):
                    cleaned_part['fileData'] = file_data
                    has_valid_content = True

        # 处理代码执行相关 part
        for code_key in ('executableCode', 'codeExecutionResult'):
            if code_key in part_dict and part_dict[code_key]:
                cleaned_part[code_key] = part_dict[code_key]
                has_valid_content = True

        # 透传 Gemini 新增/实验 part；只在字段非空时保留。
        # 核心 part 字段已在上方严格清洗，不能兜底透传；否则会把
        # functionResponse.name=null、空 inlineData/fileData/functionCall 等非法字段重新加入上游 payload。
        passthrough_part_keys = (
            'videoMetadata', 'mediaResolution', 'thought', 'thoughtSignature', 'thought_signature',
        )
        for key in passthrough_part_keys:
            if key in part_dict and key not in cleaned_part and part_dict[key]:
                cleaned_key = 'thoughtSignature' if key == 'thought_signature' else key
                cleaned_part[cleaned_key] = part_dict[key]
                if cleaned_key not in {'thought', 'thoughtSignature', 'videoMetadata', 'mediaResolution'}:
                    has_valid_content = True

        # 保留与有效媒体 part 绑定的元数据
        for metadata_key in ('videoMetadata', 'mediaResolution'):
            if metadata_key in part_dict and part_dict[metadata_key]:
                cleaned_part[metadata_key] = part_dict[metadata_key]
        
        # 只返回有有效内容的 part
        if has_valid_content:
            return cleaned_part
        else:
            logger.debug("过滤掉没有有效内容的 part")
            return None

    def _sanitize_vcore_variables_in_place(self, variables: dict[str, Any]) -> None:
        """最终出口兜底，确保发往匿名 Vcore 的 contents 不含非法核心 part。"""
        contents = variables.get('contents')
        if not isinstance(contents, list):
            return

        function_call_names: list[str] = []
        for content in cast(list[Any], contents):
            if not isinstance(content, dict):
                continue
            parts = cast(dict[str, Any], content).get('parts')
            if not isinstance(parts, list):
                continue
            for part in cast(list[Any], parts):
                if not isinstance(part, dict):
                    continue
                func_call = cast(dict[str, Any], part).get('functionCall') or cast(dict[str, Any], part).get('function_call')
                if isinstance(func_call, dict):
                    name = cast(dict[str, Any], func_call).get('name')
                    if name and str(name).strip():
                        function_call_names.append(self._sanitize_function_name(str(name)))

        sanitized_contents: list[dict[str, Any]] = []
        dropped_function_responses = 0
        for content in cast(list[Any], contents):
            if not isinstance(content, dict):
                continue
            content_dict = cast(dict[str, Any], content).copy()
            parts = content_dict.get('parts')
            if not isinstance(parts, list):
                continue

            sanitized_parts: list[dict[str, Any]] = []
            for part in cast(list[Any], parts):
                if not isinstance(part, dict):
                    continue
                part_dict = cast(dict[str, Any], part)
                cleaned = self._sanitize_vcore_part_final(part_dict, function_call_names)
                if cleaned:
                    if ('functionResponse' in part_dict or 'function_response' in part_dict) and 'functionResponse' not in cleaned:
                        dropped_function_responses += 1
                    sanitized_parts.append(cleaned)

            if sanitized_parts:
                content_dict['parts'] = sanitized_parts
                sanitized_contents.append(content_dict)

        variables['contents'] = sanitized_contents
        if dropped_function_responses:
            logger.warning(f"最终出口丢弃非法 functionResponse part: count={dropped_function_responses}")

    def _sanitize_vcore_part_final(self, part: dict[str, Any], function_call_names: list[str]) -> dict[str, Any] | None:
        cleaned: dict[str, Any] = {}
        has_content = False

        text = part.get('text')
        if text is not None and str(text) != "":
            cleaned['text'] = text
            has_content = True

        if 'thought' in part:
            cleaned['thought'] = part['thought']
        signature = part.get('thoughtSignature') or part.get('thought_signature')
        if signature:
            cleaned['thoughtSignature'] = signature

        func_call = part.get('functionCall') or part.get('function_call')
        if isinstance(func_call, dict):
            call = cast(dict[str, Any], func_call).copy()
            name = call.get('name')
            if name and str(name).strip():
                call['name'] = self._sanitize_function_name(str(name))
                call['args'] = self._coerce_function_args(call.get('args', {}))
                cleaned['functionCall'] = call
                has_content = True

        func_response = part.get('functionResponse') or part.get('function_response')
        if isinstance(func_response, dict):
            response = cast(dict[str, Any], func_response).copy()
            name = response.get('name') or response.get('functionName') or response.get('function_name')
            if not name and function_call_names:
                name = function_call_names[-1]
            if name and str(name).strip():
                response['name'] = self._sanitize_function_name(str(name))
                response['response'] = self._coerce_function_response(response.get('response', {}))
                cleaned['functionResponse'] = response
                has_content = True

        inline_data = part.get('inlineData') or part.get('inline_data')
        if isinstance(inline_data, dict):
            inline_dict = cast(dict[str, Any], inline_data)
            data = inline_dict.get('data')
            mime_type = inline_dict.get('mimeType') or inline_dict.get('mime_type')
            if data and str(data).strip() and mime_type and str(mime_type).strip():
                cleaned['inlineData'] = {**inline_dict, 'mimeType': str(mime_type)}
                cleaned['inlineData'].pop('mime_type', None)
                has_content = True

        file_data = part.get('fileData') or part.get('file_data')
        if isinstance(file_data, dict):
            file_dict = cast(dict[str, Any], file_data)
            file_uri = file_dict.get('fileUri') or file_dict.get('file_uri')
            mime_type = file_dict.get('mimeType') or file_dict.get('mime_type')
            if file_uri and str(file_uri).strip() and mime_type and str(mime_type).strip():
                cleaned['fileData'] = {**file_dict, 'fileUri': str(file_uri), 'mimeType': str(mime_type)}
                cleaned['fileData'].pop('file_uri', None)
                cleaned['fileData'].pop('mime_type', None)
                has_content = True

        for code_key in ('executableCode', 'codeExecutionResult'):
            if part.get(code_key):
                cleaned[code_key] = part[code_key]
                has_content = True

        for metadata_key in ('videoMetadata', 'mediaResolution'):
            if part.get(metadata_key):
                cleaned[metadata_key] = part[metadata_key]

        return cleaned if has_content else None

    def _ensure_function_call_thought_signatures(self, contents: Any) -> Any:
        """
        Gemini 3 工具调用要求历史中的 functionCall part 带 thoughtSignature。
        如果客户端没有回传签名，使用官方保底哨兵值跳过校验，避免 400 错误。
        """
        if not isinstance(contents, list):
            return contents

        patched_count = 0
        patched_names: set[str] = set()
        patched_contents: list[Any] = []
        for content in cast(list[Any], contents):
            if not isinstance(content, dict):
                patched_contents.append(content)
                continue

            content_dict = cast(dict[str, Any], content)
            parts = content_dict.get('parts')
            if not isinstance(parts, list):
                patched_contents.append(content)
                continue

            patched_parts: list[Any] = []
            for part in cast(list[Any], parts):
                if isinstance(part, dict):
                    part_dict = cast(dict[str, Any], part)
                    function_call = part_dict.get('functionCall')
                    if not isinstance(function_call, dict):
                        function_call = part_dict.get('function_call')
                    if isinstance(function_call, dict):
                        patched_part = part_dict.copy()
                        if 'function_call' in patched_part and 'functionCall' not in patched_part:
                            patched_part['functionCall'] = patched_part.pop('function_call')
                        if 'thought_signature' in patched_part and 'thoughtSignature' not in patched_part:
                            patched_part['thoughtSignature'] = patched_part.pop('thought_signature')
                        if not patched_part.get('thoughtSignature'):
                            func_call = cast(dict[str, Any], patched_part.get('functionCall') or {})
                            patched_count += 1
                            patched_names.add(str(func_call.get('name') or 'unknown'))
                            patched_part['thoughtSignature'] = 'skip_thought_signature_validator'
                        patched_parts.append(patched_part)
                    else:
                        patched_parts.append(part)
                else:
                    patched_parts.append(part)

            patched_content = content_dict.copy()
            patched_content['parts'] = patched_parts
            patched_contents.append(patched_content)

        if patched_count:
            logger.debug(
                "已为 %d 个缺少 thoughtSignature 的 functionCall 使用兼容兜底，函数: %s",
                patched_count,
                ", ".join(sorted(patched_names))
            )

        return patched_contents

    def _handle_thought_signature(self, contents: Any) -> Any:
        """
        处理 thoughtSignature 字段的 base64 编码
        确保 thoughtSignature 字段正确编码为 base64 字符串
        """
        import base64
        
        if isinstance(contents, list):
            return [self._handle_thought_signature(item) for item in cast(list[Any], contents)]
        if isinstance(contents, dict):
            new_dict: dict[str, Any] = {}
            contents_dict: dict[str, Any] = cast(dict[str, Any], contents)
            for k, v in contents_dict.items():
                if k == 'parts' and isinstance(v, list):
                    v_list: list[Any] = cast(list[Any], v)
                    # 处理 parts 数组中的每个 part
                    new_parts: list[Any] = []
                    for part in v_list:
                        if isinstance(part, dict):
                            new_part: dict[str, Any] = cast(dict[str, Any], part).copy()
                            # 检查是否有 thoughtSignature 字段
                            if 'thoughtSignature' in new_part:
                                signature_value = new_part['thoughtSignature']
                                if isinstance(signature_value, str):
                                    # 如果是特定的字符串，进行 base64 编码
                                    if signature_value == "skip_thought_signature_validator":
                                        encoded_bytes = base64.b64encode(signature_value.encode('utf-8'))
                                        new_part['thoughtSignature'] = encoded_bytes.decode('utf-8')
                                    # 如果已经是 base64 编码的字符串，保持不变
                                    # 其他情况也保持不变
                            new_parts.append(new_part)
                        else:
                            new_parts.append(part)
                    new_dict[k] = new_parts
                else:
                    new_dict[k] = self._handle_thought_signature(v) if isinstance(v, (dict, list)) else v
            return new_dict
        return contents

    @staticmethod
    def prepare_headers(creds: dict[str, Any]) -> dict[str, str]:
        """准备请求头"""
        # 提取原始头信息
        headers = RequestTransformer._extract_headers_from_creds(creds)
        
        # 设置必要的头信息
        headers['content-type'] = 'application/json'
        
        # 移除可能导致问题的头
        problematic_headers = [
            'content-length', 'Content-Length', 'host', 'Host',
            'connection', 'Connection', 'accept-encoding'
        ]
        for header in problematic_headers:
            headers.pop(header, None)
            
        return headers
    
    @staticmethod
    def _extract_headers_from_creds(creds: dict[str, Any]) -> dict[str, str]:
        """从凭证中提取头信息"""
        if hasattr(creds, 'model_dump') and hasattr(creds, 'headers'):
            headers_attr = getattr(creds, 'headers')
            if isinstance(headers_attr, dict):
                return cast(dict[str, str], headers_attr).copy()
        
        raw_headers = creds.get('headers')
        if isinstance(raw_headers, dict):
            return cast(dict[str, str], raw_headers).copy()
            
        return {}


class ResponseAggregator:
    """响应聚合器"""

    @staticmethod
    def log_stream_forward_complete(prefix: str, chunks: int, bytes_total: int, elapsed: float) -> None:
        logger.info(
            f"{prefix} 流式转发完成: chunks={chunks}, "
            f"bytes={bytes_total}, 耗时={elapsed:.1f}s"
        )

    @staticmethod
    async def aggregate_stream(
        stream_generator: Any,
        _raw_image_response: bool = False,
        progress_context: dict[str, str] | None = None,
    ) -> dict[str, Any]:
        """
        聚合流式响应为非流式对象
        """
        all_parts: list[dict[str, Any]] = []
        candidate_parts_by_index: dict[int, list[dict[str, Any]]] = {}
        finish_reason: str | None = None
        finish_message: str | None = None
        safety_ratings: list[dict[str, Any]] = []
        citation_metadata: dict[str, Any] = {}
        grounding_metadata: dict[str, Any] = {}
        token_count: int | None = None
        avg_logprobs: float | None = None
        logprobs_result: dict[str, Any] | None = None
        candidate_index = 0
        usage_metadata: dict[str, Any] = {}
        
        create_time: str | None = None
        model_version: str | None = None
        prompt_feedback: dict[str, Any] = {}
        response_id: str | None = None
        model_status: dict[str, Any] | None = None
        buffered_chunk_count = 0
        buffered_bytes_total = 0
        aggregate_started_at = time.monotonic()
        aggregate_error: Exception | None = None
        
        try:
            async for stream_item in stream_generator:
                try:
                    if isinstance(stream_item, dict):
                        chunk = stream_item
                        chunk_bytes = 0
                    else:
                        actual_json_str = str(stream_item).strip()
                        if actual_json_str.startswith("data: "):
                            actual_json_str = actual_json_str[6:]

                        if not actual_json_str:
                            continue
                        chunk = json.loads(actual_json_str)
                        chunk_bytes = len(actual_json_str.encode('utf-8'))

                    buffered_chunk_count += 1
                    buffered_bytes_total += chunk_bytes

                    # 检查 chunk 是否包含错误 (使用统一解析逻辑)
                    parsed_error = parse_error_response(chunk)
                    if parsed_error:
                        raise parsed_error

                    # 提取顶层元数据
                    create_time = chunk.get('createTime') or create_time
                    model_version = chunk.get('modelVersion') or model_version
                    prompt_feedback = chunk.get('promptFeedback', {}) or prompt_feedback
                    response_id = chunk.get('responseId') or response_id
                    usage_metadata = chunk.get('usageMetadata', {}) or usage_metadata
                    model_status = chunk.get('modelStatus') or model_status
                    
                    candidates = chunk.get('candidates', [])
                    if candidates:
                        candidates_to_read = candidates if _raw_image_response else candidates[:1]
                        for candidate_offset, candidate in enumerate(candidates_to_read):
                            if not isinstance(candidate, dict):
                                continue
                            candidate_result_index = candidate_offset
                            if candidate.get('index') is not None:
                                try:
                                    candidate_result_index = int(candidate['index'])
                                except (TypeError, ValueError):
                                    candidate_result_index = candidate_offset

                            # 提取 parts。raw 图片响应会读取所有 candidates，支持一次上游响应返回多张图；
                            # 普通 Gemini 非流式响应仍只聚合第一个 candidate，避免改变文本接口语义。
                            content_obj = candidate.get('content', {})
                            raw_parts = content_obj.get('parts', []) if isinstance(content_obj, dict) else []
                            parts = ResponseAggregator._normalize_response_parts(raw_parts)
                            if parts:
                                all_parts.extend(parts)
                                candidate_parts_by_index.setdefault(candidate_result_index, []).extend(
                                    [cast(dict[str, Any], part) for part in cast(list[Any], parts) if isinstance(part, dict)]
                                )

                            # 提取 candidate 元数据；多图 raw 响应不依赖这些字段，只保留首个 candidate 的元数据。
                            if candidate_offset == 0:
                                finish_reason = candidate.get('finishReason') or finish_reason
                                finish_message = candidate.get('finishMessage') or finish_message
                                safety_ratings = candidate.get('safetyRatings') or safety_ratings
                                citation_metadata = candidate.get('citationMetadata') or citation_metadata
                                grounding_metadata = candidate.get('groundingMetadata') or grounding_metadata
                                if candidate.get('tokenCount') is not None:
                                    token_count = candidate['tokenCount']
                                if candidate.get('avgLogprobs') is not None:
                                    avg_logprobs = candidate['avgLogprobs']
                                if candidate.get('logprobsResult') is not None:
                                    logprobs_result = candidate['logprobsResult']
                                if candidate.get('index') is not None:
                                    candidate_index = candidate['index']
                            
                except json.JSONDecodeError as e:
                    logger.debug(f"JSON 解析失败，跳过此块: {e}")
                    continue
                    
        except VcoreError as e:
            aggregate_error = e
            raise
        except Exception as e:
            aggregate_error = e
            raise InternalError(message=f"Non-streaming request error: {e}")
        finally:
            if buffered_chunk_count > 0:
                progress_prefix = progress_context.get("prefix") if isinstance(progress_context, dict) else ""
                prefix = f"{progress_prefix} " if progress_prefix else ""
                state = "异常" if aggregate_error is not None else "完成"
                log_fields = [
                    f"chunks={buffered_chunk_count}",
                    f"bytes={buffered_bytes_total}",
                    f"状态={state}",
                ]
                log_fields.append(f"耗时={time.monotonic() - aggregate_started_at:.1f}s")
                logger.info(
                    f"{prefix}非流式接收缓存结束: {', '.join(log_fields)}"
                )

        # 流式上游通常把一段连续文本拆成多个 text part。非流式响应若原样返回
        # 多个相邻 text part，部分 SDK/客户端会在 part 边界自行插入换行或段落，
        # 导致最终文本出现类似“当\n\n掉”“</\n\nrelationships>”的伪换行。
        # 因此在构造非流式结果前合并相邻、同类型的文本 part，保持文本内容本身不变。
        all_parts = ResponseAggregator._merge_adjacent_text_parts(all_parts)
        if candidate_parts_by_index:
            candidate_parts_by_index = {
                idx: ResponseAggregator._merge_adjacent_text_parts(parts)
                for idx, parts in candidate_parts_by_index.items()
            }
        
        # 处理图片响应特例
        image_outputs = ResponseAggregator._extract_image_outputs(all_parts)
        is_image_response = bool(image_outputs)
        if _raw_image_response and image_outputs:
            return ResponseAggregator._raw_image_response_from_images(image_outputs)

        if is_image_response and candidate_parts_by_index:
            result_candidates: list[dict[str, Any]] = []
            for idx, parts in sorted(candidate_parts_by_index.items(), key=lambda item: item[0]):
                item: dict[str, Any] = {
                    "index": idx,
                    "content": {
                        "parts": parts,
                        "role": "model",
                    },
                }
                if finish_reason:
                    item["finishReason"] = finish_reason.upper()
                result_candidates.append(item)

            result: dict[str, Any] = {"candidates": result_candidates}
            optional_result_fields: dict[str, Any] = {
                "createTime": create_time,
                "modelVersion": model_version,
                "promptFeedback": prompt_feedback,
                "responseId": response_id,
                "usageMetadata": usage_metadata,
                "modelStatus": model_status,
            }
            for key, value in optional_result_fields.items():
                if value is not None and value != {} and value != "":
                    result[key] = value
            return result

        # 处理图片响应特例
        image_outputs = ResponseAggregator._extract_image_outputs(all_parts)
        if _raw_image_response and image_outputs:
            return ResponseAggregator._raw_image_response_from_images(image_outputs)
        
        # 构建最终响应
        if not all_parts:
            all_parts = [{"text": " "}]
        
        result_candidate: dict[str, Any] = {
            "index": candidate_index
        }
        if finish_reason:
            result_candidate["finishReason"] = finish_reason.upper()
            
        result_candidate["content"] = {
            "parts": all_parts,
            "role": "model"
        }
        
        # 构建候选结果，只添加非空字段
        optional_candidate_fields: dict[str, Any] = {
            "finishMessage": finish_message,
            "safetyRatings": safety_ratings,
            "citationMetadata": citation_metadata,
            "groundingMetadata": grounding_metadata,
            "tokenCount": token_count,
            "avgLogprobs": avg_logprobs,
            "logprobsResult": logprobs_result
        }
        
        for key, value in optional_candidate_fields.items():
            if value is not None and value != [] and value != {}:
                result_candidate[key] = value
        
        # 构建最终结果，只添加非空字段
        result: dict[str, Any] = {"candidates": [result_candidate]}
        
        optional_result_fields: dict[str, Any] = {
            "createTime": create_time,
            "modelVersion": model_version,
            "promptFeedback": prompt_feedback,
            "responseId": response_id,
            "usageMetadata": usage_metadata,
            "modelStatus": model_status
        }
        
        for key, value in optional_result_fields.items():
            if value is not None and value != {} and value != "":
                result[key] = value
        
        return result

    @staticmethod
    def _extract_image_outputs(parts: list[dict[str, Any]]) -> list[dict[str, str]]:
        images: list[dict[str, str]] = []
        for part in parts:
            inline_data = part.get('inlineData') or part.get('inline_data')
            if isinstance(inline_data, dict):
                mime_type = inline_data.get('mimeType') or inline_data.get('mime_type') or 'image/png'
                data = inline_data.get('data')
                if isinstance(data, str) and data.strip():
                    images.append({"b64_json": data, "mime_type": str(mime_type)})

            file_data = part.get('fileData') or part.get('file_data')
            if isinstance(file_data, dict):
                mime_type = file_data.get('mimeType') or file_data.get('mime_type') or 'image/png'
                file_uri = file_data.get('fileUri') or file_data.get('file_uri') or file_data.get('uri') or file_data.get('url')
                if isinstance(file_uri, str) and file_uri.strip():
                    images.append({"url": file_uri.strip(), "mime_type": str(mime_type)})
        return images

    @staticmethod
    def _merge_adjacent_text_parts(parts: list[dict[str, Any]]) -> list[dict[str, Any]]:
        """合并相邻文本 part，避免非流式客户端在 part 边界额外插入换行。"""
        merged: list[dict[str, Any]] = []

        def mergeable_text_signature(part: dict[str, Any]) -> tuple[Any, Any] | None:
            if "text" not in part:
                return None
            # 只要是纯文本 part 就允许合并。Vcore/Gemini 可能在 text part 上附带
            # thought/thoughtSignature 以外的无害元数据；这些元数据如果阻止合并，
            # Gemini 非流式客户端仍会在碎片边界插入额外换行。
            blocking_keys = {
                "inlineData", "inline_data", "fileData", "file_data",
                "functionCall", "function_call", "functionResponse", "function_response",
                "executableCode", "executable_code", "codeExecutionResult", "code_execution_result",
            }
            if any(key in part and ResponseAggregator._has_meaningful_part_value(part.get(key)) for key in blocking_keys):
                return None
            # thoughtSignature 标记了独立上下文，保守起见不跨 signature 合并。
            signature = part.get("thoughtSignature") or part.get("thought_signature")
            return (bool(part.get("thought")), signature)

        for raw_part in parts:
            part = dict(raw_part)
            sig = mergeable_text_signature(part)
            if sig is None or not merged:
                merged.append(part)
                continue

            previous = merged[-1]
            previous_sig = mergeable_text_signature(previous)
            if previous_sig == sig:
                previous["text"] = str(previous.get("text", "")) + str(part.get("text", ""))
            else:
                merged.append(part)

        return merged

    @staticmethod
    def _normalize_response_parts(raw_parts: Any) -> list[dict[str, Any]]:
        """清洗上游 part 中的空占位字段，避免 Gemini 非流式返回碎片化 parts。"""
        if not isinstance(raw_parts, list):
            return []

        normalized: list[dict[str, Any]] = []
        placeholder_keys = {
            "inlineData", "inline_data", "fileData", "file_data",
            "functionCall", "function_call", "functionResponse", "function_response",
            "executableCode", "executable_code", "codeExecutionResult", "code_execution_result",
        }

        for item in raw_parts:
            if not isinstance(item, dict):
                continue
            part = dict(cast(dict[str, Any], item))
            if 'inline_data' in part and 'inlineData' not in part:
                inline_data = part['inline_data']
                if isinstance(inline_data, dict):
                    part['inlineData'] = {snake_to_camel(str(k)): v for k, v in cast(dict[Any, Any], inline_data).items()}
                else:
                    part['inlineData'] = inline_data
            if 'file_data' in part and 'fileData' not in part:
                file_data = part['file_data']
                if isinstance(file_data, dict):
                    part['fileData'] = {snake_to_camel(str(k)): v for k, v in cast(dict[Any, Any], file_data).items()}
                else:
                    part['fileData'] = file_data
            for key in list(part.keys()):
                if key in placeholder_keys and not ResponseAggregator._has_meaningful_part_value(part.get(key)):
                    part.pop(key, None)
            # 部分上游/客户端会把 part 类型放在 data/type 上；Gemini 响应无需保留
            # data='text' 这类非标准占位，否则下游可能把 text part 当成多模态 part。
            if part.get("data") == "text":
                part.pop("data", None)
            if part.get("type") == "text":
                part.pop("type", None)
            normalized.append(part)

        return normalized

    @staticmethod
    def _has_meaningful_part_value(value: Any) -> bool:
        """判断 part 子字段是否真实有内容；空壳对象不应阻止文本合并。"""
        if value is None or value is False:
            return False
        if isinstance(value, str):
            return value != ""
        if isinstance(value, dict):
            return any(ResponseAggregator._has_meaningful_part_value(v) for v in value.values())
        if isinstance(value, (list, tuple, set)):
            return any(ResponseAggregator._has_meaningful_part_value(v) for v in value)
        return True

    @staticmethod
    def _raw_image_response_from_images(images: list[dict[str, str]]) -> dict[str, Any]:
        return {
            "created": int(time.time()),
            "data": [
                dict(image)
                for image in images
            ],
        }