chatgpt-on-wechat/agent/tools/vision/vision.py

"""
Vision tool - Analyze images using Vision API.
Supports local files (auto base64-encoded) and HTTP URLs.

Provider resolution:
  - tools.vision.model (if set) means "prefer this model first; fall back to
    other configured providers if it fails". The model name is mapped to its
    native provider (e.g. doubao-* → Doubao, kimi-* → Moonshot, gpt-* →
    OpenAI/LinkAI). That provider is tried first, then the standard auto
    chain runs as fallback (with the preferred provider de-duplicated).
  - Auto chain priority:
      1. Main model via bot.call_vision — only when the main bot is known
         to actually support vision (not just expose a call_vision method).
      2. Other models whose API key is configured.
      3. OpenAI / LinkAI raw HTTP.
    When use_linkai=true, LinkAI is promoted to #1.
"""

import base64
import ipaddress
import os
import socket
import subprocess
import tempfile
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from urllib.parse import urlparse

import requests

from agent.tools.base_tool import BaseTool, ToolResult
from common import const
from common.log import logger
from config import conf

DEFAULT_MODEL = const.GPT_41_MINI
DEFAULT_TIMEOUT = 180
MAX_TOKENS = 4000
COMPRESS_THRESHOLD = 1_048_576  # 1 MB

SUPPORTED_EXTENSIONS = {
    "jpg": "image/jpeg",
    "jpeg": "image/jpeg",
    "png": "image/png",
    "gif": "image/gif",
    "webp": "image/webp",
}

_MAIN_MODEL_PROVIDER_NAME = "MainModel"

# (config_key_for_api_key, bot_type, default_vision_model, provider_display_name)
# Auto-discovered as fallback vision providers when their API key is configured.
# OpenAI and LinkAI are handled separately (raw HTTP providers), so not listed here.
_DISCOVERABLE_MODELS = [
    ("moonshot_api_key", const.MOONSHOT, const.KIMI_K2_6, "Moonshot"),
    ("ark_api_key", const.DOUBAO, const.DOUBAO_SEED_2_PRO, "Doubao"),
    ("dashscope_api_key", const.QWEN_DASHSCOPE, const.QWEN37_PLUS, "DashScope"),
    ("claude_api_key", const.CLAUDEAPI, const.CLAUDE_4_6_SONNET, "Claude"),
    ("gemini_api_key", const.GEMINI, const.GEMINI_35_FLASH, "Gemini"),
    ("qianfan_api_key", const.QIANFAN, const.ERNIE_45_TURBO_VL, "Qianfan"),
    ("zhipu_ai_api_key", const.ZHIPU_AI, const.GLM_4_7, "ZhipuAI"),
    ("minimax_api_key", const.MiniMax, const.MINIMAX_M2_7, "MiniMax"),
    ("mimo_api_key", const.MIMO, const.MIMO_V2_5_PRO, "MiMo"),
]

# Model name prefix → discoverable provider display_name.
# Used to auto-route tools.vision.model to its native provider.
# Matched case-insensitively; longest prefix wins.
_MODEL_PREFIX_TO_PROVIDER = [
    ("doubao-", "Doubao"),
    ("kimi-", "Moonshot"),
    ("moonshot-", "Moonshot"),
    ("qwen", "DashScope"),       # qwen-*, qwen3-*, qwen3.6-*, etc.
    ("claude-", "Claude"),
    ("ernie-", "Qianfan"),
    ("gemini-", "Gemini"),
    ("glm-", "ZhipuAI"),
    ("minimax-", "MiniMax"),
    ("abab", "MiniMax"),
    ("mimo-", "MiMo"),
]

# Model prefixes that natively belong to OpenAI / LinkAI (raw HTTP providers).
_OPENAI_MODEL_PREFIXES = ("gpt-", "o1-", "o3-", "o4-", "chatgpt-")

# Maps the UI provider id (persisted in tools.vision.provider) to the internal
# display name used in VisionProvider.name. Keep in sync with _DISCOVERABLE_MODELS
# and the openai/linkai branches in _route_by_model_name.
_PROVIDER_ID_TO_DISPLAY = {
    "openai": "OpenAI",
    "linkai": "LinkAI",
    "moonshot": "Moonshot",
    "doubao": "Doubao",
    "dashscope": "DashScope",
    "claudeAPI": "Claude",
    "gemini": "Gemini",
    "qianfan": "Qianfan",
    "zhipu": "ZhipuAI",
    "minimax": "MiniMax",
    "mimo": "MiMo",
}


@dataclass
class VisionProvider:
    """A single Vision API provider configuration."""
    name: str
    api_key: str
    api_base: str
    extra_headers: dict = field(default_factory=dict)
    model_override: Optional[str] = None
    use_bot: bool = False  # When True, call via bot.call_vision instead of raw HTTP
    fallback_bot: Any = None  # Bot instance for non-main-model providers


class VisionAPIError(Exception):
    """Raised when a Vision API call fails and should trigger fallback."""
    pass


class Vision(BaseTool):
    """Analyze images using Vision API"""

    name: str = "vision"
    description: str = (
        "Analyze a local image or image URL (jpg/jpeg/png) using Vision API. "
        "Can describe content, extract text, identify objects, colors, etc. "
    )

    params: dict = {
        "type": "object",
        "properties": {
            "image": {
                "type": "string",
                "description": "Local file path or HTTP(S) URL of the image to analyze",
            },
            "question": {
                "type": "string",
                "description": "Question to ask about the image",
            },
        },
        "required": ["image", "question"],
    }

    def __init__(self, config: dict = None):
        self.config = config or {}

    @staticmethod
    def is_available() -> bool:
        return True

    def execute(self, args: Dict[str, Any]) -> ToolResult:
        image = args.get("image", "").strip()
        question = args.get("question", "").strip()

        if not image:
            return ToolResult.fail("Error: 'image' parameter is required")
        if not question:
            return ToolResult.fail("Error: 'question' parameter is required")

        providers = self._resolve_providers()
        if not providers:
            return ToolResult.fail(
                "Error: No model available for Vision.\n"
                "The main model does not support vision and no other API keys are configured.\n"
                "Options:\n"
                "  1. Switch to a multimodal model (e.g. ernie-4.5-turbo-vl, qwen3.7-plus, claude-sonnet-4-6, gemini-2.0-flash)\n"
                "  2. Configure OPENAI_API_KEY: env_config(action=\"set\", key=\"OPENAI_API_KEY\", value=\"your-key\")\n"
                "  3. Configure LINKAI_API_KEY: env_config(action=\"set\", key=\"LINKAI_API_KEY\", value=\"your-key\")"
            )

        try:
            image_content = self._build_image_content(image)
        except Exception as e:
            return ToolResult.fail(f"Error: {e}")

        # Default model is only used as a last-resort placeholder for providers
        # whose VisionProvider.model_override is None (e.g. raw OpenAI provider
        # when the user did not configure tools.vision.model).
        return self._call_with_fallback(providers, DEFAULT_MODEL, question, image_content)

    def _call_with_fallback(self, providers: List[VisionProvider], model: str,
                            question: str, image_content: dict) -> ToolResult:
        """Try each provider in order; fall back to the next one on failure."""
        errors: List[str] = []
        for i, provider in enumerate(providers):
            use_model = provider.model_override or model
            try:
                logger.info(f"[Vision] Trying provider '{provider.name}' "
                            f"with model '{use_model}' ({i + 1}/{len(providers)})")
                if provider.use_bot:
                    result = self._call_via_bot(use_model, question, image_content, provider)
                else:
                    result = self._call_api(provider, use_model, question, image_content)
                logger.info(f"[Vision] ✅ Success via {provider.name} (model={use_model})")
                return result
            except VisionAPIError as e:
                errors.append(f"[{provider.name}/{use_model}] {e}")
                logger.warning(f"[Vision] Provider '{provider.name}' failed: {e}")
            except requests.Timeout:
                errors.append(f"[{provider.name}/{use_model}] Request timed out after {DEFAULT_TIMEOUT}s")
                logger.warning(f"[Vision] Provider '{provider.name}' timed out")
            except requests.ConnectionError:
                errors.append(f"[{provider.name}/{use_model}] Connection failed")
                logger.warning(f"[Vision] Provider '{provider.name}' connection failed")
            except Exception as e:
                errors.append(f"[{provider.name}/{use_model}] {e}")
                logger.error(f"[Vision] Provider '{provider.name}' unexpected error: {e}", exc_info=True)

        return ToolResult.fail(
            "Error: All Vision API providers failed.\n" + "\n".join(f"  - {err}" for err in errors)
        )

    def _resolve_providers(self) -> List[VisionProvider]:
        """
        Build an ordered list of providers to try.

        Semantics of `tools.vision.model`:
          "Prefer this model first; fall back to other configured providers
           if it fails."

        Order:
          1. The provider that natively serves `tools.vision.model` (if any
             and its API key is configured) — using the user-specified model
             name verbatim.
          2. Auto-discovery chain as fallback:
               - use_linkai=true → [LinkAI, MainModel?, OtherModels…, OpenAI]
               - default         → [MainModel?, OtherModels…, OpenAI, LinkAI]
             MainModel is only included when the main bot is known to support
             vision (see _main_bot_supports_vision).

        Providers that share the same display name as the preferred provider
        are de-duplicated to avoid retrying the same endpoint twice.
        """
        user_model = self._resolve_user_vision_model()
        user_provider = self._resolve_user_vision_provider()
        providers: List[VisionProvider] = []

        # Step 1: preferred provider — explicit `tools.vision.provider`
        # wins so custom model names can still be routed correctly. Falls
        # through to model-name prefix inference when provider is unset.
        preferred = None
        if user_provider and user_model:
            preferred = self._route_by_provider_id(user_provider, user_model)
        if not preferred and user_model:
            preferred = self._route_by_model_name(user_model)
        if preferred:
            providers.extend(preferred)

        # Step 2: auto-discovery chain as fallback
        existing = {p.name for p in providers}
        fallback: List[VisionProvider] = []
        use_linkai = conf().get("use_linkai", False) and conf().get("linkai_api_key")

        if use_linkai:
            self._append_provider(fallback, lambda: self._build_linkai_provider(user_model))
            self._append_provider(fallback, self._build_main_model_provider)
            self._append_other_model_providers(fallback, preferred_model=user_model)
            self._append_provider(fallback, lambda: self._build_openai_provider(user_model))
        else:
            self._append_provider(fallback, self._build_main_model_provider)
            self._append_other_model_providers(fallback, preferred_model=user_model)
            self._append_provider(fallback, lambda: self._build_openai_provider(user_model))
            self._append_provider(fallback, lambda: self._build_linkai_provider(user_model))

        for p in fallback:
            if p.name in existing:
                continue
            providers.append(p)
            existing.add(p.name)

        return providers

    @staticmethod
    def _append_provider(providers: List[VisionProvider], builder) -> None:
        p = builder()
        if p:
            providers.append(p)

    @staticmethod
    def _resolve_user_vision_model() -> Optional[str]:
        """Read tools.vision.model (singular ``tool`` kept as runtime fallback)."""
        tools_conf = conf().get("tools") or conf().get("tool") or {}
        if not isinstance(tools_conf, dict):
            return None
        vision_conf = tools_conf.get("vision", {})
        if not isinstance(vision_conf, dict):
            return None
        m = vision_conf.get("model")
        if isinstance(m, str) and m.strip():
            return m.strip()
        return None

    @staticmethod
    def _resolve_user_vision_provider() -> Optional[str]:
        """Read tools.vision.provider — the UI-persisted vendor id.

        Lets users pin a vendor for custom model names that prefix-inference
        can't recognize. Returns None when unset/blank.
        """
        tools_conf = conf().get("tools") or conf().get("tool") or {}
        if not isinstance(tools_conf, dict):
            return None
        vision_conf = tools_conf.get("vision", {})
        if not isinstance(vision_conf, dict):
            return None
        p = vision_conf.get("provider")
        if isinstance(p, str) and p.strip():
            return p.strip()
        return None

    @staticmethod
    def _infer_provider_from_model(model_name: str) -> Optional[str]:
        """
        Infer the provider display name from a model name's prefix.
        Returns None when no rule matches (or for OpenAI-family names, which
        are handled separately by the caller).
        """
        if not model_name:
            return None
        lower = model_name.lower()
        # Sort by prefix length desc so e.g. "moonshot-" wins over hypothetical "moo-"
        for prefix, display_name in sorted(_MODEL_PREFIX_TO_PROVIDER, key=lambda x: -len(x[0])):
            if lower.startswith(prefix.lower()):
                return display_name
        return None

    def _route_by_provider_id(self, provider_id: str, user_model: str) -> Optional[List[VisionProvider]]:
        """Route by the UI-persisted provider id.

        Returns:
          - [provider] : provider id is known and its key is configured.
          - None       : unknown provider id, or the bot can't be created.
                         Caller falls through to model-name-based routing.
        """
        display_name = _PROVIDER_ID_TO_DISPLAY.get(provider_id)
        if not display_name:
            return None

        # OpenAI / LinkAI use raw HTTP providers, not the discoverable bot path.
        if provider_id == "openai":
            p = self._build_openai_provider(user_model)
            return [p] if p else None
        if provider_id == "linkai":
            p = self._build_linkai_provider(user_model)
            return [p] if p else None

        # Discoverable bot-backed providers.
        for config_key, bot_type, _default_model, name in _DISCOVERABLE_MODELS:
            if name != display_name:
                continue
            api_key = conf().get(config_key, "")
            if not api_key or not api_key.strip():
                logger.warning(f"[Vision] tools.vision.provider='{provider_id}' "
                               f"but '{config_key}' is not configured. Falling back.")
                return None
            try:
                from models.bot_factory import create_bot
                bot = create_bot(bot_type)
                if not hasattr(bot, 'call_vision'):
                    logger.warning(f"[Vision] '{display_name}' bot does not implement call_vision.")
                    return None
            except Exception as e:
                logger.warning(f"[Vision] Failed to create '{display_name}' bot: {e}")
                return None
            return [VisionProvider(
                name=display_name,
                api_key="",
                api_base="",
                model_override=user_model,
                use_bot=True,
                fallback_bot=bot,
            )]
        return None

    def _route_by_model_name(self, user_model: str) -> Optional[List[VisionProvider]]:
        """
        Try to build a provider list using the user-specified model name.
        Returns:
          - [provider]  : matched and the provider's key is configured
          - []          : matched but key missing → tell caller to surface this
                          as a hard error rather than silently falling back
          - None        : no rule matches → caller should fall through to auto
        """
        lower = user_model.lower()

        # OpenAI / LinkAI family
        if lower.startswith(_OPENAI_MODEL_PREFIXES):
            providers: List[VisionProvider] = []
            # Prefer LinkAI when explicitly enabled, else OpenAI first
            use_linkai = conf().get("use_linkai", False) and conf().get("linkai_api_key")
            if use_linkai:
                self._append_provider(providers, lambda: self._build_linkai_provider(user_model))
                self._append_provider(providers, lambda: self._build_openai_provider(user_model))
            else:
                self._append_provider(providers, lambda: self._build_openai_provider(user_model))
                self._append_provider(providers, lambda: self._build_linkai_provider(user_model))
            if providers:
                return providers
            logger.warning(f"[Vision] tools.vision.model='{user_model}' looks like an OpenAI "
                           f"model but neither OPENAI_API_KEY nor LINKAI_API_KEY is configured.")
            return None  # fall through to auto

        # Discoverable native providers (Doubao, Moonshot, etc.)
        target_display = self._infer_provider_from_model(user_model)
        if not target_display:
            return None  # unknown prefix → auto

        for config_key, bot_type, _default_model, display_name in _DISCOVERABLE_MODELS:
            if display_name != target_display:
                continue
            api_key = conf().get(config_key, "")
            if not api_key or not api_key.strip():
                logger.warning(f"[Vision] tools.vision.model='{user_model}' routes to "
                               f"'{display_name}' but '{config_key}' is not configured. "
                               f"Falling back to auto-discovery.")
                return None  # fall through to auto
            try:
                from models.bot_factory import create_bot
                bot = create_bot(bot_type)
                if not hasattr(bot, 'call_vision'):
                    logger.warning(f"[Vision] '{display_name}' bot does not implement call_vision.")
                    return None
            except Exception as e:
                logger.warning(f"[Vision] Failed to create '{display_name}' bot: {e}")
                return None

            return [VisionProvider(
                name=display_name,
                api_key="",
                api_base="",
                model_override=user_model,
                use_bot=True,
                fallback_bot=bot,
            )]

        return None

    def _append_other_model_providers(self, providers: List[VisionProvider],
                                       preferred_model: Optional[str] = None) -> None:
        """
        Auto-discover other models whose API key is configured.
        Skip the main model's own bot_type (already covered by MainModel
        provider), unless the main model itself does not support vision —
        in that case we still want the vendor's dedicated vision model
        as a fallback. Also skip bot_types that already appear in the
        provider list.

        If preferred_model matches a provider's family, use it instead
        of that provider's hard-coded default model.
        """
        main_bot_type = None
        main_bot_supports_vision = False
        if self.model and hasattr(self.model, '_resolve_bot_type'):
            main_bot_type = self.model._resolve_bot_type(conf().get("model", ""))
            main_bot = getattr(self.model, "bot", None)
            main_bot_supports_vision = self._main_bot_supports_vision(main_bot)

        existing_names = {p.name for p in providers}
        preferred_provider = self._infer_provider_from_model(preferred_model) if preferred_model else None

        for config_key, bot_type, default_model, display_name in _DISCOVERABLE_MODELS:
            if display_name in existing_names:
                continue
            # Same bot_type as the main model is normally handled by the
            # MainModel provider; only skip it here if the main model
            # actually supports vision. Otherwise fall through and add
            # the vendor's dedicated vision model as a fallback.
            if bot_type == main_bot_type and main_bot_supports_vision:
                continue
            api_key = conf().get(config_key, "")
            if not api_key or not api_key.strip():
                continue

            try:
                from models.bot_factory import create_bot
                bot = create_bot(bot_type)
                if not hasattr(bot, 'call_vision'):
                    continue
            except Exception:
                continue

            model_for_provider = (preferred_model
                                  if preferred_provider == display_name and preferred_model
                                  else default_model)

            provider = VisionProvider(
                name=display_name,
                api_key="",
                api_base="",
                model_override=model_for_provider,
                use_bot=True,
                fallback_bot=bot,
            )

            # Same vendor as the main bot is the most natural fallback when
            # the main model itself does not support vision — promote it to
            # the front of the list instead of relying on declaration order.
            if bot_type == main_bot_type:
                providers.insert(0, provider)
            else:
                providers.append(provider)

    def _main_bot_supports_vision(self, bot) -> bool:
        """
        Whether the main bot is known to natively support vision.

        Having a `call_vision` method is necessary but not sufficient —
        some bots implement the method against an endpoint that does not
        actually serve vision models, which causes silent failures when a
        vendor-foreign model name is forwarded.

        Resolution order:
          1. If the bot explicitly declares `supports_vision`, trust it.
             This lets bots opt in or out based on their own runtime
             configuration (e.g. the currently selected model).
          2. Otherwise, fall back to a model-name prefix heuristic: trust
             call_vision when the main model looks like an OpenAI family
             model or matches a known multimodal vendor prefix.
        """
        if bot is None:
            return False
        if hasattr(bot, "supports_vision"):
            return bool(getattr(bot, "supports_vision"))
        main_model = (conf().get("model") or "").lower()
        if not main_model:
            return False
        if main_model.startswith(_OPENAI_MODEL_PREFIXES):
            return True
        return self._infer_provider_from_model(main_model) is not None

    def _build_main_model_provider(self) -> Optional[VisionProvider]:
        """
        Use the vendor's own model for vision via bot.call_vision.
        Gated by _main_bot_supports_vision so non-vision bots (DeepSeek, etc.)
        do not get routed vendor-foreign model names.
        """
        if not (self.model and hasattr(self.model, 'bot')):
            return None
        try:
            bot = self.model.bot
        except Exception:
            return None
        if not hasattr(bot, 'call_vision'):
            return None
        if not self._main_bot_supports_vision(bot):
            return None

        # Use the configured main model name; do NOT inject tools.vision.model
        # here, because by the time we reach this branch the tools.vision.model
        # routing has already been attempted (and either matched the main bot
        # or failed to find a provider).
        main_model_name = conf().get("model") or None

        return VisionProvider(
            name=_MAIN_MODEL_PROVIDER_NAME,
            api_key="",
            api_base="",
            model_override=main_model_name,
            use_bot=True,
        )

    def _build_openai_provider(self, preferred_model: Optional[str] = None) -> Optional[VisionProvider]:
        api_key = conf().get("open_ai_api_key") or os.environ.get("OPENAI_API_KEY")
        if not api_key:
            return None
        api_base = (conf().get("open_ai_api_base") or os.environ.get("OPENAI_API_BASE", "")).rstrip("/") \
            or "https://api.openai.com/v1"
        # Only honor preferred_model when it looks like an OpenAI-family name;
        # otherwise the OpenAI endpoint would 400 on a vendor-specific name.
        model_override = preferred_model if (
            preferred_model and preferred_model.lower().startswith(_OPENAI_MODEL_PREFIXES)
        ) else None
        return VisionProvider(
            name="OpenAI",
            api_key=api_key,
            api_base=self._ensure_v1(api_base),
            model_override=model_override,
        )

    def _build_linkai_provider(self, preferred_model: Optional[str] = None) -> Optional[VisionProvider]:
        api_key = conf().get("linkai_api_key") or os.environ.get("LINKAI_API_KEY")
        if not api_key:
            return None
        api_base = (conf().get("linkai_api_base") or os.environ.get("LINKAI_API_BASE", "")).rstrip("/") \
            or "https://api.link-ai.tech"
        from common.utils import get_cloud_headers
        extra = get_cloud_headers(api_key)
        extra.pop("Authorization", None)
        extra.pop("Content-Type", None)
        # LinkAI is a multi-vendor proxy and accepts most model names, so we
        # honor any user-configured model name here.
        return VisionProvider(
            name="LinkAI",
            api_key=api_key,
            api_base=self._ensure_v1(api_base),
            extra_headers=extra,
            model_override=preferred_model,
        )

    def _call_via_bot(self, model: str, question: str, image_content: dict,
                      provider: Optional[VisionProvider] = None) -> ToolResult:
        """
        Call a model's call_vision with vendor-native API format.
        Uses the provider's _fallback_bot if set, otherwise the main model bot.
        Raises VisionAPIError on failure so fallback can proceed.
        """
        try:
            bot = (provider and provider.fallback_bot) or self.model.bot
        except Exception as e:
            raise VisionAPIError(f"Cannot access bot: {e}")

        # Extract the raw image URL from the OpenAI-format image_content block
        image_url = image_content.get("image_url", {}).get("url", "")
        if not image_url:
            raise VisionAPIError("No image URL in content block")

        try:
            response = bot.call_vision(
                image_url=image_url,
                question=question,
                model=model,
                max_tokens=MAX_TOKENS,
            )
        except Exception as e:
            raise VisionAPIError(f"call_vision failed: {e}")

        if response is NotImplemented:
            raise VisionAPIError("Bot does not support vision")

        if isinstance(response, dict) and response.get("error"):
            raise VisionAPIError(f"API error - {response.get('message', 'Unknown')}")

        content = response.get("content", "") if isinstance(response, dict) else ""
        if not content:
            raise VisionAPIError("Empty response from main model")

        usage_info = response.get("usage", {}) if isinstance(response, dict) else {}

        # Use the actual model name from the bot response if available
        actual_model = response.get("model", model) if isinstance(response, dict) else model
        provider_name = provider.name if provider else _MAIN_MODEL_PROVIDER_NAME
        return ToolResult.success({
            "model": actual_model,
            "provider": provider_name,
            "content": content,
            "usage": usage_info,
        })

    @staticmethod
    def _ensure_v1(api_base: str) -> str:
        """Append /v1 if the base URL doesn't already end with a versioned path."""
        if not api_base:
            return api_base
        # Already has /v1 or similar version suffix
        if api_base.rstrip("/").split("/")[-1].startswith("v"):
            return api_base
        return api_base.rstrip("/") + "/v1"

    @staticmethod
    def _validate_url_safe(url: str) -> None:
        """Reject URLs that target private/loopback/link-local addresses (SSRF guard).

        Resolves the hostname to its IP address(es) and blocks any that fall
        into non-public ranges.  Also rejects URLs with no host, non-HTTP(S)
        schemes, or hosts that fail DNS resolution.

        Raises:
            ValueError: if the URL targets a disallowed address.
        """
        parsed = urlparse(url)
        if parsed.scheme not in ("http", "https"):
            raise ValueError(f"Unsupported URL scheme: {parsed.scheme}")

        hostname = parsed.hostname
        if not hostname:
            raise ValueError("URL has no hostname")

        try:
            # Resolve all addresses for the hostname.
            addr_infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
        except socket.gaierror:
            raise ValueError(f"Cannot resolve hostname: {hostname}")

        for family, _, _, _, sockaddr in addr_infos:
            ip_str = sockaddr[0]
            ip = ipaddress.ip_address(ip_str)
            if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
                raise ValueError(
                    f"URL resolves to a non-public address ({ip_str}), "
                    f"request blocked for security"
                )

    def _build_image_content(self, image: str) -> dict:
        """
        Build the image_url content block.
        Both remote URLs and local files are converted to base64 data URLs
        so every bot backend can consume them without extra downloads.
        """
        if image.startswith(("http://", "https://")):
            self._validate_url_safe(image)
            return self._download_to_data_url(image)

        if not os.path.isfile(image):
            raise FileNotFoundError(f"Image file not found: {image}")

        ext = image.rsplit(".", 1)[-1].lower() if "." in image else ""
        mime_type = SUPPORTED_EXTENSIONS.get(ext)
        if not mime_type:
            raise ValueError(
                f"Unsupported image format '.{ext}'. "
                f"Supported: {', '.join(SUPPORTED_EXTENSIONS.keys())}"
            )

        file_path = self._maybe_compress(image)
        try:
            with open(file_path, "rb") as f:
                b64 = base64.b64encode(f.read()).decode("ascii")
        finally:
            if file_path != image and os.path.exists(file_path):
                os.remove(file_path)

        data_url = f"data:{mime_type};base64,{b64}"
        return {"type": "image_url", "image_url": {"url": data_url}}

    @staticmethod
    def _download_to_data_url(url: str) -> dict:
        """Download a remote image and return it as a base64 data URL."""
        resp = requests.get(url, timeout=30)
        if resp.status_code != 200:
            raise VisionAPIError(f"Failed to download image: HTTP {resp.status_code}")
        content_type = resp.headers.get("Content-Type", "image/jpeg").split(";")[0].strip()
        if not content_type.startswith("image/"):
            content_type = "image/jpeg"
        b64 = base64.b64encode(resp.content).decode("ascii")
        data_url = f"data:{content_type};base64,{b64}"
        return {"type": "image_url", "image_url": {"url": data_url}}

    @staticmethod
    def _maybe_compress(path: str) -> str:
        """Compress image to under COMPRESS_THRESHOLD with max long-edge 1536px."""
        file_size = os.path.getsize(path)
        if file_size <= COMPRESS_THRESHOLD:
            return path

        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
        tmp.close()

        def _try_sips(max_dim: str, quality: str) -> bool:
            try:
                subprocess.run(
                    ["sips", "-Z", max_dim, "-s", "formatOptions", quality,
                     path, "--out", tmp.name],
                    capture_output=True, check=True,
                )
                return True
            except (FileNotFoundError, subprocess.CalledProcessError):
                return False

        def _try_convert(max_dim: str, quality: str) -> bool:
            try:
                subprocess.run(
                    ["convert", path, "-resize", f"{max_dim}x{max_dim}>",
                     "-quality", quality, tmp.name],
                    capture_output=True, check=True,
                )
                return True
            except (FileNotFoundError, subprocess.CalledProcessError):
                return False

        attempts = [
            ("1536", "85"),
            ("1536", "70"),
            ("1536", "50"),
        ]

        for max_dim, quality in attempts:
            ok = _try_sips(max_dim, quality) or _try_convert(max_dim, quality)
            if not ok:
                continue
            new_size = os.path.getsize(tmp.name)
            logger.debug(f"[Vision] Compressed image "
                         f"({file_size // 1024}KB -> {new_size // 1024}KB, "
                         f"max_dim={max_dim}, q={quality})")
            if new_size <= COMPRESS_THRESHOLD:
                return tmp.name

        if os.path.exists(tmp.name) and os.path.getsize(tmp.name) > 0:
            return tmp.name

        os.remove(tmp.name)
        return path

    def _call_api(self, provider: VisionProvider, model: str,
                  question: str, image_content: dict) -> ToolResult:
        """
        Call a single provider's Vision API.
        Raises VisionAPIError on recoverable failures so the caller can try
        the next provider.
        """
        payload = {
            "model": model,
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": question},
                        image_content,
                    ],
                }
            ],
        }

        headers = {
            "Authorization": f"Bearer {provider.api_key}",
            "Content-Type": "application/json",
            **provider.extra_headers,
        }

        resp = requests.post(
            f"{provider.api_base}/chat/completions",
            headers=headers,
            json=payload,
            timeout=DEFAULT_TIMEOUT,
        )

        if resp.status_code != 200:
            raise VisionAPIError(f"HTTP {resp.status_code}: {resp.text[:200]}")

        data = resp.json()

        if "error" in data:
            msg = data["error"].get("message", "Unknown API error")
            raise VisionAPIError(f"API error - {msg}")

        content = ""
        choices = data.get("choices", [])
        if choices:
            content = choices[0].get("message", {}).get("content", "")

        usage = data.get("usage", {})
        result = {
            "model": model,
            "provider": provider.name,
            "content": content,
            "usage": {
                "prompt_tokens": usage.get("prompt_tokens", 0),
                "completion_tokens": usage.get("completion_tokens", 0),
                "total_tokens": usage.get("total_tokens", 0),
            },
        }
        return ToolResult.success(result)