src/api/oai_adapter.py

"""OpenAI 兼容适配层

将 OpenAI Chat Completion 格式转换为 Gemini 格式（请求），
将 Gemini SSE 格式转换为 OpenAI 格式（响应）。
"""

import json
import math
import re
import time
import uuid
import copy
from typing import Any

from src.utils.logger import get_logger

logger = get_logger(__name__)

DEFAULT_IMAGE_MODEL = "gemini-2.5-flash-image"
OPENAI_IMAGE_MODEL_ALIASES = {
    "gpt-image-1",
    "dall-e-2",
    "dall-e-3",
}

FINISH_REASON_MAP = {
    "STOP": "stop",
    "MAX_TOKENS": "length",
    "SAFETY": "content_filter",
    "RECITATION": "content_filter",
    "TOOL_CALLS": "tool_calls",
    "MALFORMED_FUNCTION_CALL": "tool_calls",
    "BLOCKLIST": "content_filter",
    "PROHIBITED_CONTENT": "content_filter",
    "SPII": "content_filter",
    "OTHER": "stop",
}

_OPENAI_WEB_SEARCH_TOOL_TYPES = {
    "web_search",
    "web_search_preview",
    "web_search_preview_2025_03_11",
    "google_search",
}
_OPENAI_CODE_TOOL_TYPES = {"code_interpreter", "code_execution"}
_OPENAI_URL_CONTEXT_TOOL_TYPES = {"url_context", "urlcontext"}
_OPENAI_FILE_SEARCH_TOOL_TYPES = {"file_search", "retrieval"}
_OPENAI_FALLBACK_TOOL_TYPES = {
    "computer_use",
    "computer_use_preview",
    "mcp",
    "custom",
}
_GEMINI_FUNCTION_NAME_RE = re.compile(r"[^A-Za-z0-9_.-]+")


class OAIRequestConverter:
    """OpenAI → Gemini 请求转换"""

    @staticmethod
    def convert(body: dict[str, Any]) -> tuple[str, dict[str, Any]]:
        """将 OAI ChatCompletion 请求转为 (model, gemini_payload)"""
        model = body["model"]
        messages = body.get("messages", [])

        contents: list[dict[str, Any]] = []
        system_parts: list[dict[str, str]] = []
        tool_call_name_by_id: dict[str, str] = {}

        for msg in messages:
            role = msg.get("role", "")
            content = msg.get("content")

            if role in {"system", "developer"}:
                system_text = _content_to_text(content)
                if system_text:
                    system_parts.append({"text": system_text})

            elif role == "user":
                parts = _convert_content_to_parts(content)
                if parts:
                    contents.append({"role": "user", "parts": parts})

            elif role == "assistant":
                parts: list[dict[str, Any]] = []
                assistant_text = _content_to_text(content)
                if assistant_text:
                    parts.append({"text": assistant_text})
                tool_calls = msg.get("tool_calls")
                if tool_calls:
                    for tc in tool_calls:
                        parsed = _extract_oai_tool_call(tc)
                        if not parsed:
                            continue
                        tc_id, func_name, args_obj, thought_signature = parsed
                        if tc_id:
                            tool_call_name_by_id[str(tc_id)] = func_name
                        function_call_part: dict[str, Any] = {"functionCall": {"name": func_name, "args": args_obj}}
                        if thought_signature:
                            function_call_part["thoughtSignature"] = thought_signature
                        parts.append(function_call_part)
                if parts:
                    contents.append({"role": "model", "parts": parts})

            elif role == "tool":
                tool_call_id = msg.get("tool_call_id")
                name = msg.get("name") or tool_call_name_by_id.get(str(tool_call_id), "unknown")
                raw = msg.get("content", "")
                resp_obj = _parse_tool_response(raw)
                contents.append({
                    "role": "function",
                    "parts": [{"functionResponse": {"name": name, "response": resp_obj}}]
                })
            
            elif role == "function":
                name = msg.get("name") or "unknown"
                resp_obj = _parse_tool_response(msg.get("content", ""))
                contents.append({
                    "role": "function",
                    "parts": [{"functionResponse": {"name": name, "response": resp_obj}}]
                })

        gemini_payload: dict[str, Any] = {"contents": contents}

        if system_parts:
            gemini_payload["systemInstruction"] = {"parts": system_parts}

        # tools：OpenAI function tools 直接映射为 Gemini functionDeclarations；
        # OpenAI web/code/url 等内建工具尽量映射为 Gemini 原生工具，其他工具降级为函数声明。
        oai_tools = body.get("tools")
        if not oai_tools and body.get("functions"):
            oai_tools = [{"type": "function", "function": f} for f in body.get("functions", [])]
        declared_tool_names: set[str] = set()
        native_tools: list[dict[str, Any]] = []
        if oai_tools:
            func_decls = []
            for t in oai_tools:
                f = _extract_oai_function_tool(t)
                if f:
                    decl = _oai_function_to_gemini_declaration(f)
                    declared_tool_names.add(str(decl["name"]))
                    func_decls.append(decl)
                else:
                    mapped_tools = _map_oai_tool_to_gemini_tools(t)
                    native_tools.extend(mapped_tools)
                    for mapped_tool in mapped_tools:
                        for mapped_decl in mapped_tool.get("functionDeclarations", []) if isinstance(mapped_tool, dict) else []:
                            if isinstance(mapped_decl, dict) and mapped_decl.get("name"):
                                declared_tool_names.add(str(mapped_decl["name"]))
            gemini_tools: list[dict[str, Any]] = []
            if func_decls:
                gemini_tools.append({"functionDeclarations": func_decls})
            gemini_tools.extend(native_tools)
            if gemini_tools:
                gemini_payload["tools"] = gemini_tools

        # tool_choice
        tc = body.get("tool_choice", body.get("function_call"))
        if tc:
            tc_value = tc.lower() if isinstance(tc, str) else tc
            if tc_value == "none":
                gemini_payload.pop("tools", None)
                gemini_payload.pop("toolConfig", None)
            elif tc_value == "auto":
                if declared_tool_names:
                    gemini_payload["toolConfig"] = {"functionCallingConfig": {"mode": "AUTO"}}
            elif isinstance(tc_value, str) and tc_value in {"required", "any"}:
                if declared_tool_names:
                    gemini_payload["toolConfig"] = {"functionCallingConfig": {"mode": "ANY"}}
            elif isinstance(tc, dict) and (tc.get("type") == "function" or "name" in tc):
                fn_name = tc.get("function", {}).get("name") if tc.get("type") == "function" else tc.get("name")
                if fn_name:
                    fn_name = _sanitize_gemini_function_name(str(fn_name))
                    if fn_name not in declared_tool_names:
                        _ensure_synthetic_function_tool(gemini_payload, fn_name)
                        declared_tool_names.add(fn_name)
                    gemini_payload["toolConfig"] = {
                        "functionCallingConfig": {"mode": "ANY", "allowedFunctionNames": [fn_name]}
                    }
            elif isinstance(tc, dict):
                # OpenAI/Responses 的非 function tool_choice（如 web_search/code_interpreter）
                # Gemini 没有等价的强制选择配置；保留 tools 映射，不额外发送 functionCallingConfig。
                pass
            else:
                raise ValueError(f"Unsupported tool_choice/function_call: {tc}")

        # generationConfig
        gen_cfg: dict[str, Any] = {}
        for oai_key, gemini_key in [
            ("temperature", "temperature"),
            ("top_p", "topP"),
            ("top_k", "topK"),
            ("presence_penalty", "presencePenalty"),
            ("frequency_penalty", "frequencyPenalty"),
            ("seed", "seed"),
        ]:
            if oai_key in body and body[oai_key] is not None:
                gen_cfg[gemini_key] = body[oai_key]

        n_val = body.get("n")
        if n_val is not None and isinstance(n_val, int) and n_val > 1:
            gen_cfg["candidateCount"] = n_val

        max_tokens = body.get("max_tokens") or body.get("max_completion_tokens")
        if max_tokens is not None:
            gen_cfg["maxOutputTokens"] = max_tokens

        if body.get("logprobs") is not None:
            gen_cfg["responseLogprobs"] = bool(body.get("logprobs"))
        if body.get("top_logprobs") is not None:
            gen_cfg["logprobs"] = body.get("top_logprobs")

        stop = body.get("stop")
        if stop is not None:
            gen_cfg["stopSequences"] = [stop] if isinstance(stop, str) else stop

        rf = body.get("response_format")
        if isinstance(rf, dict):
            rf_type = rf.get("type")
            if rf_type == "json_object":
                gen_cfg["responseMimeType"] = "application/json"
            elif rf_type == "json_schema":
                gen_cfg["responseMimeType"] = "application/json"
                schema = rf.get("json_schema", {}).get("schema")
                if schema:
                    gen_cfg["responseSchema"] = _sanitize_schema_for_gemini(schema)

        modalities = body.get("modalities") or body.get("response_modalities")
        if isinstance(modalities, list):
            converted_modalities = [_convert_modality(m) for m in modalities if _convert_modality(m)]
            if converted_modalities:
                gen_cfg["responseModalities"] = converted_modalities
        elif _looks_like_image_model(model):
            gen_cfg.setdefault("responseModalities", ["TEXT", "IMAGE"])

        if isinstance(n_val, int) and n_val > 1:
            response_modalities = gen_cfg.get("responseModalities")
            if isinstance(response_modalities, list) and any(str(item).upper() == "IMAGE" for item in response_modalities):
                image_config = gen_cfg.get("imageConfig")
                if not isinstance(image_config, dict):
                    image_config = {}
                image_config["numberOfImages"] = n_val
                gen_cfg["imageConfig"] = image_config
                gen_cfg.pop("candidateCount", None)

        if gen_cfg:
            gemini_payload["generationConfig"] = gen_cfg

        oai_safety = body.get("safety_settings") or body.get("safetySettings")
        if isinstance(oai_safety, list):
            gemini_payload["safetySettings"] = oai_safety

        labels = body.get("labels") or body.get("metadata")
        if isinstance(labels, dict):
            gemini_payload["labels"] = {str(k): str(v)[:63] for k, v in labels.items() if v is not None}

        cached_content = body.get("cached_content") or body.get("cachedContent")
        if isinstance(cached_content, str) and cached_content:
            gemini_payload["cachedContent"] = cached_content

        return model, gemini_payload


class OAIImageRequestConverter:
    """OpenAI Images API → Gemini 图片生成请求转换"""

    @staticmethod
    def resolve_model(model: Any) -> str:
        if not model:
            return DEFAULT_IMAGE_MODEL
        model_str = str(model)
        if model_str in OPENAI_IMAGE_MODEL_ALIASES:
            return DEFAULT_IMAGE_MODEL
        return model_str

    @staticmethod
    def convert_generation(body: dict[str, Any]) -> tuple[str, dict[str, Any], int, str]:
        model = OAIImageRequestConverter.resolve_model(body.get("model"))
        prompt = _append_negative_prompt(str(body.get("prompt") or ""), body.get("negative_prompt"))
        if not prompt.strip():
            raise ValueError("prompt is required")

        n = _coerce_positive_int(body.get("n"), default=1, maximum=8)
        response_format = str(body.get("response_format") or "b64_json")
        payload = OAIImageRequestConverter.build_payload(
            model=model,
            prompt=prompt,
            n=n,
            images=[],
            mask=None,
            size=body.get("size"),
            quality=body.get("quality"),
            style=body.get("style"),
            background=body.get("background"),
            output_format=body.get("output_format"),
        )
        return model, payload, n, response_format

    @staticmethod
    def build_payload(
        model: str,
        prompt: str,
        n: int = 1,
        images: list[dict[str, str]] | None = None,
        mask: dict[str, str] | None = None,
        size: Any = None,
        quality: Any = None,
        style: Any = None,
        background: Any = None,
        output_format: Any = None,
        mode: str = "generation",
    ) -> dict[str, Any]:
        prompt_text = _build_image_prompt(
            prompt=prompt,
            size=size,
            quality=quality,
            style=style,
            background=background,
            mode=mode,
            has_mask=bool(mask),
        )

        parts: list[dict[str, Any]] = [{"text": prompt_text}]
        for image in images or []:
            if image.get("data") and image.get("mimeType"):
                parts.append({"inlineData": {"mimeType": image["mimeType"], "data": _normalize_base64(image["data"])}})
        if mask and mask.get("data") and mask.get("mimeType"):
            parts.append({"text": "Use the following image as the edit mask when applying the requested change."})
            parts.append({"inlineData": {"mimeType": mask["mimeType"], "data": _normalize_base64(mask["data"])}})

        generation_config: dict[str, Any] = {
            "responseModalities": ["TEXT", "IMAGE"],
        }

        image_config: dict[str, Any] = {}
        if n > 1:
            image_config["numberOfImages"] = n
        
        aspect_ratio = _size_to_aspect_ratio(size)
        if aspect_ratio:
            image_config["aspectRatio"] = aspect_ratio
        image_size = _size_to_image_size(size)
        if image_size and "gemini-3" in model:
            image_config["imageSize"] = image_size
        if image_config:
            generation_config["imageConfig"] = image_config

        return {
            "contents": [{"role": "user", "parts": parts}],
            "generationConfig": generation_config,
        }


class OAIResponseConverter:
    """Gemini → OpenAI 响应转换"""

    @staticmethod
    def _extract_images_from_parts(parts: list[dict[str, Any]]) -> list[dict[str, str]]:
        images: list[dict[str, str]] = []
        for part in parts:
            image = _extract_image_from_part(part)
            if image:
                images.append(image)
        return images

    @staticmethod
    def gemini_chunk_to_openai_image_partial(
        chunk: dict[str, Any],
        response_format: str = "b64_json",
    ) -> list[dict[str, Any]]:
        """从 Gemini 流式 chunk 中提取符合 OpenAI Images data item 的图片。"""
        items: list[dict[str, Any]] = []
        for part in _iter_response_parts(chunk):
            image = _extract_image_from_part(part)
            if not image:
                continue
            if image["kind"] == "file":
                items.append({"url": image["url"]})
            else:
                items.append({"b64_json": image["data"]})
        return items

    @staticmethod
    def convert_realtime_chunk(
        chunk: dict[str, Any],
        model: str,
        request_id: str,
        is_first: bool,
        has_prior_tool_calls: bool = False,
    ) -> list[str]:
        """将单个 Gemini 增量 dict 转为 OAI SSE 事件列表（真流式用）"""
        candidates = chunk.get("candidates", [])
        if not candidates:
            candidates = [{}]

        usage_meta = chunk.get("usageMetadata")
        created = int(time.time())
        base = {"id": f"chatcmpl-{request_id}", "object": "chat.completion.chunk", "created": created, "model": model}
        events: list[str] = []

        if is_first:
            first_choices = []
            for i in range(len(candidates)):
                first_choices.append({"index": i, "delta": {"role": "assistant"}, "finish_reason": None})
            if first_choices:
                events.append(_sse_line({**base, "choices": first_choices}))

        for i, candidate in enumerate(candidates):
            parts = (candidate.get("content") or {}).get("parts", [])
            finish = candidate.get("finishReason")
            text_content, tool_calls, reasoning = _extract_parts(parts, for_stream=True)
            image_items = OAIResponseConverter._extract_images_from_parts(parts)

            if reasoning:
                events.append(_sse_line({**base, "choices": [{"index": i, "delta": {"reasoning_content": reasoning}, "finish_reason": None}]}))

            if text_content:
                events.append(_sse_line({**base, "choices": [{"index": i, "delta": {"content": text_content}, "finish_reason": None}]}))

            if image_items:
                events.append(_sse_line({**base, "choices": [{"index": i, "delta": {"images": [
                    _image_to_oai_chat_image(image)
                    for image in image_items
                ]}, "finish_reason": None}]}))

            if tool_calls:
                events.append(_sse_line({**base, "choices": [{"index": i, "delta": {"tool_calls": tool_calls}, "finish_reason": None}]}))

            if finish:
                oai_finish = _map_finish_reason(finish, has_tool_calls=bool(tool_calls) or has_prior_tool_calls)
                finish_evt: dict[str, Any] = {**base, "choices": [{"index": i, "delta": {}, "finish_reason": oai_finish}]}
                if usage_meta and i == len(candidates) - 1:
                    finish_evt["usage"] = _convert_usage(usage_meta)
                if candidate.get("logprobsResult"):
                    finish_evt["choices"][0]["logprobs"] = candidate.get("logprobsResult")
                events.append(_sse_line(finish_evt))

        return events

    @staticmethod
    def convert_realtime_chunk_objects(
        chunk: dict[str, Any],
        model: str,
        request_id: str,
        is_first: bool,
        has_prior_tool_calls: bool = False,
    ) -> list[dict[str, Any]]:
        """将单个 Gemini 增量 dict 转为 OAI SSE payload 对象列表。"""
        candidates = chunk.get("candidates", [])
        if not candidates:
            candidates = [{}]

        usage_meta = chunk.get("usageMetadata")
        created = int(time.time())
        base = {"id": f"chatcmpl-{request_id}", "object": "chat.completion.chunk", "created": created, "model": model}
        payloads: list[dict[str, Any]] = []

        if is_first:
            first_choices = []
            for i in range(len(candidates)):
                first_choices.append({"index": i, "delta": {"role": "assistant"}, "finish_reason": None})
            if first_choices:
                payloads.append({**base, "choices": first_choices})

        for i, candidate in enumerate(candidates):
            parts = (candidate.get("content") or {}).get("parts", [])
            finish = candidate.get("finishReason")
            text_content, tool_calls, reasoning = _extract_parts(parts, for_stream=True)
            image_items = OAIResponseConverter._extract_images_from_parts(parts)

            if reasoning:
                payloads.append({**base, "choices": [{"index": i, "delta": {"reasoning_content": reasoning}, "finish_reason": None}]})

            if text_content:
                # 防止单条 SSE 过大（如内嵌 3MB Base64 图片）撑爆部分客户端的单行读取缓冲区
                # 这里将长文本按 32KB 分块发送
                chunk_size = 32768
                for j in range(0, len(text_content), chunk_size):
                    text_chunk = text_content[j:j+chunk_size]
                    payloads.append({**base, "choices": [{"index": i, "delta": {"content": text_chunk}, "finish_reason": None}]})

            if image_items:
                payloads.append({**base, "choices": [{"index": i, "delta": {"images": [
                    _image_to_oai_chat_image(image)
                    for image in image_items
                ]}, "finish_reason": None}]})

            if tool_calls:
                payloads.append({**base, "choices": [{"index": i, "delta": {"tool_calls": tool_calls}, "finish_reason": None}]})

            if finish:
                oai_finish = _map_finish_reason(finish, has_tool_calls=bool(tool_calls) or has_prior_tool_calls)
                finish_evt: dict[str, Any] = {**base, "choices": [{"index": i, "delta": {}, "finish_reason": oai_finish}]}
                if usage_meta and i == len(candidates) - 1: # Usage is attached only to the last candidate's finish
                    finish_evt["usage"] = _convert_usage(usage_meta)
                if candidate.get("logprobsResult"):
                    finish_evt["choices"][0]["logprobs"] = candidate.get("logprobsResult")
                payloads.append(finish_evt)

        return payloads

    @staticmethod
    def gemini_sse_to_oai_stream(gemini_chunk: str, model: str, request_id: str) -> list[str]:
        """将单条 Gemini SSE 转为多条 OAI SSE 事件（假流式用）"""
        data = _parse_gemini_sse(gemini_chunk)
        if data is None:
            return []
        return OAIResponseConverter.convert_realtime_chunk(data, model, request_id, is_first=True)

    @staticmethod
    def gemini_json_to_oai_json(gemini_response: dict[str, Any], model: str) -> dict[str, Any]:
        """将 Gemini 非流式响应转为 OAI ChatCompletion JSON"""
        request_id = uuid.uuid4().hex[:24]
        candidates = gemini_response.get("candidates", [])
        if not candidates:
            candidates = [{}]

        usage_meta = gemini_response.get("usageMetadata")

        choices = []
        for i, candidate in enumerate(candidates):
            parts = (candidate.get("content") or {}).get("parts", [])
            finish = candidate.get("finishReason")
            text_content, tool_calls, reasoning = _extract_parts(parts, for_stream=False)
            image_items = OAIResponseConverter._extract_images_from_parts(parts)
            oai_finish = _map_finish_reason(finish, has_tool_calls=bool(tool_calls)) if finish else ("tool_calls" if tool_calls else "stop")

            message: dict[str, Any] = {"role": "assistant", "content": text_content or None}
            if tool_calls:
                message["tool_calls"] = tool_calls
            if reasoning:
                message["reasoning_content"] = reasoning
            if image_items:
                message["images"] = [
                    _image_to_oai_chat_image(image)
                    for image in image_items
                ]

            choice = {"index": i, "message": message, "finish_reason": oai_finish}
            if candidate.get("logprobsResult"):
                choice["logprobs"] = candidate.get("logprobsResult")
            choices.append(choice)

        result: dict[str, Any] = {
            "id": f"chatcmpl-{request_id}",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": model,
            "choices": choices,
        }
        if usage_meta:
            result["usage"] = _convert_usage(usage_meta)

        return result

    @staticmethod
    def gemini_json_to_oai_image_data(gemini_response: dict[str, Any], response_format: str = "b64_json") -> list[dict[str, Any]]:
        """从 Gemini 响应中抽取 OpenAI Images API data 数组。"""
        items: list[dict[str, Any]] = []
        raw_data = gemini_response.get("data")
        if isinstance(raw_data, list):
            for item in raw_data:
                if not isinstance(item, dict):
                    continue
                b64_data = item.get("b64_json")
                url = item.get("url")
                if isinstance(b64_data, str) and b64_data.strip():
                    items.append({"b64_json": _normalize_base64(b64_data)})
                elif isinstance(url, str) and url:
                    items.append({"url": url})
            if items:
                return items

        for part in _iter_response_parts(gemini_response):
            image = _extract_image_from_part(part)
            if not image:
                continue
            if image["kind"] == "file":
                items.append({"url": image["url"]})
            else:
                items.append({"b64_json": image["data"]})
        return items


# ==================== 内部工具函数 ====================

def _convert_content_to_parts(content: Any) -> list[dict[str, Any]]:
    """将 OpenAI 多模态 message content 转为 Gemini parts"""
    if content is None:
        return []
    if isinstance(content, str):
        return [{"text": content}]

    parts: list[dict[str, Any]] = []
    if isinstance(content, list):
        for item in content:
            if isinstance(item, str):
                parts.append({"text": item})
            elif isinstance(item, dict):
                t = item.get("type")
                if t in {"text", "input_text"}:
                    text = item.get("text")
                    if text is not None:
                        parts.append({"text": str(text)})
                elif t in {"image_url", "input_image"}:
                    url_obj = item.get("image_url") or item.get("input_image") or {}
                    url = url_obj.get("url") if isinstance(url_obj, dict) else url_obj
                    image_part = _image_url_to_part(str(url or ""))
                    if image_part:
                        parts.append(image_part)
                elif "text" in item and len(item) == 1:
                    parts.append({"text": str(item["text"])})
    return parts


def _content_to_text(content: Any) -> str:
    if content is None:
        return ""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        texts: list[str] = []
        for item in content:
            if isinstance(item, str):
                texts.append(item)
            elif isinstance(item, dict):
                item_type = item.get("type")
                if item_type in {"text", "input_text"} and item.get("text") is not None:
                    texts.append(str(item["text"]))
                elif item.get("text") is not None and len(item) == 1:
                    texts.append(str(item["text"]))
        return "".join(texts)
    return str(content)


def _parse_tool_response(raw: Any) -> dict[str, Any]:
    if isinstance(raw, dict):
        return raw
    if isinstance(raw, list):
        return {"result": raw}
    try:
        parsed = json.loads(raw) if isinstance(raw, str) else raw
        return parsed if isinstance(parsed, dict) else {"result": parsed}
    except json.JSONDecodeError:
        return {"result": raw}


def _extract_oai_tool_call(tool_call: Any) -> tuple[str | None, str, Any, str | None] | None:
    """兼容标准与常见非标准 OpenAI tool_call 形态。"""
    if not isinstance(tool_call, dict):
        return None

    tc_id = tool_call.get("id") or tool_call.get("tool_call_id") or tool_call.get("call_id")
    thought_signature = tool_call.get("thoughtSignature") or tool_call.get("thought_signature")
    func = tool_call.get("function")
    if isinstance(func, dict):
        name = func.get("name") or tool_call.get("name")
        args = func.get("arguments", tool_call.get("arguments", tool_call.get("args", {})))
        thought_signature = thought_signature or func.get("thoughtSignature") or func.get("thought_signature")
    else:
        name = tool_call.get("name") or tool_call.get("function_name")
        args = tool_call.get("arguments", tool_call.get("args", {}))

    if not name:
        return None
    return (
        str(tc_id) if tc_id else None,
        str(name),
        _coerce_function_args(args),
        str(thought_signature) if thought_signature else None,
    )


def _extract_oai_function_tool(tool: Any) -> dict[str, Any] | None:
    if not isinstance(tool, dict):
        return None
    tool_type = str(tool.get("type") or "").lower()
    if tool_type and tool_type != "function":
        return None
    if tool.get("type") == "function" and isinstance(tool.get("function"), dict):
        function_obj = tool["function"]
        return function_obj if function_obj.get("name") else None
    if tool.get("function") and isinstance(tool.get("function"), str):
        copied = tool.copy()
        copied["name"] = copied.pop("function")
        return copied if copied.get("name") else None
    if tool.get("type") == "function" and tool.get("name"):
        return tool
    if tool.get("name") and ("parameters" in tool or "description" in tool):
        return tool
    return None


def _oai_function_to_gemini_declaration(function_obj: dict[str, Any]) -> dict[str, Any]:
    """OpenAI function tool → Gemini FunctionDeclaration，并补齐 Vcore 必填字段。"""
    raw_name = function_obj.get("name") or function_obj.get("function") or "tool"
    name = _sanitize_gemini_function_name(str(raw_name))
    decl: dict[str, Any] = {"name": name}

    description = function_obj.get("description")
    if description:
        decl["description"] = str(description)

    parameters = function_obj.get("parameters") or function_obj.get("input_schema") or function_obj.get("inputSchema")
    if isinstance(parameters, dict):
        decl["parameters"] = _ensure_object_schema(_sanitize_schema_for_gemini(parameters))
    else:
        decl["parameters"] = {"type": "object", "properties": {}}
    return decl


def _map_oai_tool_to_gemini_tools(tool: Any) -> list[dict[str, Any]]:
    """尽量把 OpenAI/Responses 非 function 工具映射到 Gemini 原生工具或安全降级函数。"""
    if not isinstance(tool, dict):
        return []

    tool_type = str(tool.get("type") or "").lower()
    if tool_type in _OPENAI_WEB_SEARCH_TOOL_TYPES:
        return [{"googleSearch": {}}]
    if tool_type in _OPENAI_CODE_TOOL_TYPES:
        return [{"codeExecution": {}}]
    if tool_type in _OPENAI_URL_CONTEXT_TOOL_TYPES:
        return [{"urlContext": {}}]
    if tool_type in _OPENAI_FILE_SEARCH_TOOL_TYPES:
        if isinstance(tool.get("retrieval"), dict) and tool["retrieval"]:
            return [{"retrieval": copy.deepcopy(tool["retrieval"])}]
        return [{"functionDeclarations": [_fallback_openai_tool_declaration(tool, tool_type)]}]
    if tool_type in _OPENAI_FALLBACK_TOOL_TYPES or tool_type:
        return [{"functionDeclarations": [_fallback_openai_tool_declaration(tool, tool_type)]}]
    return []


def _fallback_openai_tool_declaration(tool: dict[str, Any], tool_type: str) -> dict[str, Any]:
    """把 Gemini 不支持的 OpenAI 工具安全降级为普通函数声明。"""
    raw_name = tool.get("name") or tool.get("function") or tool_type or "tool"
    name = _sanitize_gemini_function_name(str(raw_name))
    description = tool.get("description") or f"OpenAI-compatible tool `{tool_type or name}`."
    parameters = (
        tool.get("parameters")
        or tool.get("input_schema")
        or tool.get("inputSchema")
        or tool.get("schema")
        or {"type": "object", "properties": {}}
    )
    decl: dict[str, Any] = {
        "name": name,
        "description": str(description),
        "parameters": _ensure_object_schema(_sanitize_schema_for_gemini(parameters)) if isinstance(parameters, dict) else {"type": "object", "properties": {}},
    }
    return decl


def _ensure_synthetic_function_tool(gemini_payload: dict[str, Any], fn_name: str) -> None:
    """tool_choice 指向未声明函数时补一个最小函数声明，避免上游 allowedFunctionNames 报错。"""
    tools = gemini_payload.setdefault("tools", [])
    if not isinstance(tools, list):
        gemini_payload["tools"] = tools = []
    for tool in tools:
        if isinstance(tool, dict) and isinstance(tool.get("functionDeclarations"), list):
            declarations = tool["functionDeclarations"]
            if any(isinstance(decl, dict) and decl.get("name") == fn_name for decl in declarations):
                return
            declarations.append({"name": fn_name, "parameters": {"type": "object", "properties": {}}})
            return
    tools.insert(0, {"functionDeclarations": [{"name": fn_name, "parameters": {"type": "object", "properties": {}}}]})


def _sanitize_gemini_function_name(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_object_schema(schema: Any) -> dict[str, Any]:
    if not isinstance(schema, dict):
        return {"type": "object", "properties": {}}
    ensured = copy.deepcopy(schema)
    if not ensured.get("type"):
        ensured["type"] = "object"
    if str(ensured.get("type")).lower() == "object" and not isinstance(ensured.get("properties"), dict):
        ensured["properties"] = {}
    return ensured


def _sanitize_schema_for_gemini(schema: Any) -> Any:
    """清理 OpenAI/JSON Schema 中 Gemini 工具声明不稳定支持的字段。"""
    if isinstance(schema, list):
        return [_sanitize_schema_for_gemini(item) for item in schema]
    if not isinstance(schema, dict):
        return schema

    unsupported_keys = {
        "$schema", "$id", "$defs", "definitions", "additionalProperties",
        "patternProperties", "unevaluatedProperties", "dependentSchemas",
        "if", "then", "else", "not",
        "examples", "default", "nullable",
    }
    sanitized: dict[str, Any] = {}
    for key, value in schema.items():
        if key in unsupported_keys:
            continue
        if key in {"allOf", "anyOf", "oneOf"} and isinstance(value, list):
            flattened = _flatten_composed_schema(key, value)
            for flattened_key, flattened_value in flattened.items():
                sanitized[flattened_key] = _sanitize_schema_for_gemini(flattened_value)
            continue
        if key == "const":
            sanitized["enum"] = [value]
            continue
        if key == "type" and isinstance(value, list):
            non_null_types = [item for item in value if item != "null"]
            sanitized[key] = non_null_types[0] if non_null_types else "string"
            continue
        sanitized[key] = _sanitize_schema_for_gemini(value)
    if sanitized.get("type") == "object" and "properties" not in sanitized:
        sanitized["properties"] = {}
    return sanitized


def _flatten_composed_schema(kind: str, variants: list[Any]) -> dict[str, Any]:
    """把 OpenAI 常见组合 schema 尽量降级成 Gemini 可接受的 schema。"""
    dict_variants = [copy.deepcopy(v) for v in variants if isinstance(v, dict)]
    if not dict_variants:
        return {}

    if kind == "allOf":
        merged: dict[str, Any] = {}
        required: list[Any] = []
        properties: dict[str, Any] = {}
        for variant in dict_variants:
            for key, value in variant.items():
                if key == "properties" and isinstance(value, dict):
                    properties.update(value)
                elif key == "required" and isinstance(value, list):
                    required.extend(item for item in value if item not in required)
                else:
                    merged[key] = value
        if properties:
            merged["properties"] = properties
            merged.setdefault("type", "object")
        if required:
            merged["required"] = required
        return merged

    first = dict_variants[0]
    if all(v.get("type") == "object" for v in dict_variants):
        properties = {}
        for variant in dict_variants:
            if isinstance(variant.get("properties"), dict):
                properties.update(variant["properties"])
        if properties:
            return {"type": "object", "properties": properties}
    return first


def _parse_data_uri(uri: str) -> tuple[str, str]:
    """解析 data:mime;base64,DATA 格式"""
    try:
        header, data = uri.split(",", 1)
        mime = header.split(":")[1].split(";")[0]
        return mime, data
    except (ValueError, IndexError):
        return "", ""


def _image_url_to_part(url: str) -> dict[str, Any] | None:
    if not url:
        return None
    if url.startswith("data:"):
        mime, b64 = _parse_data_uri(url)
        if mime and b64:
            return {"inlineData": {"mimeType": mime, "data": _normalize_base64(b64)}}
        return None
    if url.startswith(("http://", "https://", "gs://")):
        return {"fileData": {"mimeType": _guess_mime_from_url(url), "fileUri": url}}
    return None


def _parse_gemini_sse(chunk: str) -> dict[str, Any] | None:
    """从 Gemini SSE 行解析 JSON"""
    s = chunk.strip()
    if s.startswith("data: "):
        s = s[6:]
    if not s:
        return None
    try:
        return json.loads(s)
    except json.JSONDecodeError:
        return None


def _extract_parts(parts: list[dict[str, Any]], for_stream: bool = False) -> tuple[str, list[dict[str, Any]] | None, str]:
    """从 Gemini parts 提取 (text_content, tool_calls, reasoning_content)"""
    texts: list[str] = []
    thoughts: list[str] = []
    tool_calls: list[dict[str, Any]] = []

    for i, part in enumerate(parts):
        if part.get("thought") and "text" in part:
            thoughts.append(str(part["text"]))
        elif "text" in part and not part.get("thought"):
            texts.append(str(part["text"]))
            
        if "executableCode" in part or "executable_code" in part:
            code_obj = part.get("executableCode") or part.get("executable_code")
            if isinstance(code_obj, dict) and "code" in code_obj:
                lang = code_obj.get("language", "python")
                texts.append(f"\n```{lang}\n{code_obj['code']}\n```\n")
                
        if "codeExecutionResult" in part or "code_execution_result" in part:
            res_obj = part.get("codeExecutionResult") or part.get("code_execution_result")
            if isinstance(res_obj, dict) and "output" in res_obj:
                texts.append(f"\n```output\n{res_obj['output']}\n```\n")

        if "functionCall" in part or "function_call" in part:
            fc = part.get("functionCall") or part.get("function_call") or {}
            if not isinstance(fc, dict):
                continue
            args = _coerce_function_args(fc.get("args", {}))
            thought_signature = part.get("thoughtSignature") or part.get("thought_signature")
            call_id = fc.get("id") or fc.get("callId") or fc.get("call_id") or part.get("id") or part.get("callId") or part.get("call_id")
            tool_call: dict[str, Any] = {
                "index": len(tool_calls),
                "id": str(call_id) if call_id else f"call_{uuid.uuid4().hex[:24]}",
                "type": "function",
                "function": {
                    "name": fc.get("name", ""),
                    "arguments": json.dumps(args, ensure_ascii=False),
                },
            }
            if thought_signature:
                # 非标准扩展字段：OpenAI 协议本身没有 thoughtSignature，
                # 但部分代理客户端会原样回传 unknown fields，可用于 Gemini 3 工具调用续写。
                tool_call["thought_signature"] = thought_signature
                tool_call["function"]["thought_signature"] = thought_signature
            if not for_stream:
                tool_call.pop("index", None)
            tool_calls.append(tool_call)

    text_content = "".join(texts)
    reasoning = "".join(thoughts)
    return text_content, tool_calls if tool_calls else None, reasoning


def _map_finish_reason(finish: Any, has_tool_calls: bool = False) -> str:
    """Gemini finishReason → OpenAI finish_reason"""
    if has_tool_calls:
        return "tool_calls"
    if not finish:
        return "stop"
    return FINISH_REASON_MAP.get(str(finish).upper(), "stop")


def _convert_usage(meta: dict[str, Any]) -> dict[str, int]:
    """Gemini usageMetadata → OAI usage"""
    prompt = meta.get("promptTokenCount", 0)
    completion = meta.get("candidatesTokenCount", 0)
    return {
        "prompt_tokens": prompt,
        "completion_tokens": completion,
        "total_tokens": meta.get("totalTokenCount", prompt + completion),
    }


def _sse_line(obj: dict[str, Any]) -> str:
    return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n"


def _convert_modality(value: Any) -> str | None:
    normalized = str(value).lower()
    if normalized in {"text", "message"}:
        return "TEXT"
    if normalized in {"image", "images"}:
        return "IMAGE"
    return None


def _looks_like_image_model(model: str) -> bool:
    model_l = model.lower()
    return "image" in model_l or model_l in OPENAI_IMAGE_MODEL_ALIASES


def _coerce_positive_int(value: Any, default: int, maximum: int) -> int:
    try:
        parsed = int(value)
    except (TypeError, ValueError):
        parsed = default
    return max(1, min(parsed, maximum))


def _normalize_base64(data: str) -> str:
    value = data.strip()
    if "," in value and value.startswith("data:"):
        _, value = value.split(",", 1)
    value = value.replace("-", "+").replace("_", "/")
    padding = len(value) % 4
    if padding:
        value += "=" * (4 - padding)
    return value


def _guess_mime_from_url(url: str) -> str:
    lower = url.lower().split("?", 1)[0].split("#", 1)[0]
    if lower.endswith(".jpg") or lower.endswith(".jpeg"):
        return "image/jpeg"
    if lower.endswith(".webp"):
        return "image/webp"
    if lower.endswith(".gif"):
        return "image/gif"
    return "image/png"


def _output_format_to_mime(output_format: Any) -> str | None:
    if not output_format:
        return None
    value = str(output_format).lower().strip()
    if value in {"png", "image/png"}:
        return "image/png"
    if value in {"jpeg", "jpg", "image/jpeg"}:
        return "image/jpeg"
    if value in {"webp", "image/webp"}:
        return "image/webp"
    return None


def _build_image_prompt(
    prompt: str,
    size: Any,
    quality: Any,
    style: Any,
    background: Any,
    mode: str,
    has_mask: bool,
) -> str:
    lines = [prompt.strip()]
    if mode == "edit":
        lines.append("Edit the provided image according to the prompt while preserving unaffected details.")
    elif mode == "variation":
        lines.append("Create a faithful variation of the provided image.")
    if has_mask:
        lines.append("Respect the provided mask as the editable region.")
    if size and str(size).lower() != "auto":
        lines.append(f"Target output size/aspect: {size}.")
    if quality and str(quality).lower() != "auto":
        lines.append(f"Quality preference: {quality}.")
    if style and str(style).lower() != "auto":
        lines.append(f"Style preference: {style}.")
    if background and str(background).lower() != "auto":
        lines.append(f"Background preference: {background}.")
    return "\n".join(line for line in lines if line)


def _append_negative_prompt(prompt: str, negative_prompt: Any) -> str:
    if negative_prompt is None or str(negative_prompt).strip() == "":
        return prompt
    return f"{prompt.strip()}\nAvoid: {negative_prompt}".strip()


def _size_to_aspect_ratio(size: Any) -> str | None:
    if not size:
        return None
    value = str(size).lower().strip()
    if value in {"auto", ""}:
        return None
    if value in {"1024x1024", "1536x1536"}:
        return "1:1"
    if value in {"1536x1024", "1792x1024"}:
        return "3:2" if value.startswith("1536") else "16:9"
    if value in {"1024x1536", "1024x1792"}:
        return "2:3" if value.endswith("1536") else "9:16"
    try:
        width_str, height_str = value.split("x", 1)
        width = int(width_str)
        height = int(height_str)
        if width <= 0 or height <= 0:
            return None
        gcd = math.gcd(width, height)
        ratio = f"{width // gcd}:{height // gcd}"
        supported = {"1:1", "3:4", "4:3", "9:16", "16:9", "2:3", "3:2"}
        return ratio if ratio in supported else None
    except (ValueError, TypeError):
        return None


def _size_to_image_size(size: Any) -> str | None:
    if not size:
        return None
    value = str(size).lower().strip()
    try:
        width_str, height_str = value.split("x", 1)
        max_side = max(int(width_str), int(height_str))
    except (ValueError, TypeError):
        return None
    if max_side >= 3000:
        return "4K"
    if max_side >= 1500:
        return "2K"
    return "1K"


def _iter_response_parts(gemini_response: dict[str, Any]) -> list[dict[str, Any]]:
    parts: list[dict[str, Any]] = []
    for candidate in gemini_response.get("candidates", []) or []:
        if not isinstance(candidate, dict):
            continue
        content = candidate.get("content") or {}
        if not isinstance(content, dict):
            continue
        for part in content.get("parts", []) or []:
            if isinstance(part, dict):
                parts.append(part)
    return parts


def _image_to_oai_chat_image(image: dict[str, str]) -> dict[str, str]:
    item = {"type": "image", "mime_type": image.get("mime_type", "image/png")}
    if image.get("kind") == "file":
        item["url"] = image["url"]
    else:
        item["b64_json"] = image["data"]
    return item


def _extract_image_from_part(part: dict[str, Any]) -> dict[str, str] | None:
    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():
            return {"kind": "inline", "mime_type": str(mime_type), "data": _normalize_base64(data)}

    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():
            return {"kind": "file", "mime_type": str(mime_type), "url": file_uri.strip()}
    return None


def _coerce_function_args(args: Any) -> Any:
    if isinstance(args, str):
        try:
            return json.loads(args)
        except json.JSONDecodeError:
            return {"raw": args}
    return args if args is not None else {}