src/providers/availability.py - develop - dsboikov/aiBoardRoom

src/providers/availability.py
61 строка2 KB
Денис
GigaChat and YandexGPT added
04 авг 2026, 09:00
04 авг 2026, 09:006d6fee5
"""Detect which providers should appear in the UI."""

from __future__ import annotations

import time
from typing import Iterable

import httpx

from config import PROVIDER_BASE_URLS, PROVIDER_IDS, PROVIDER_LABELS
from utils.logger import get_logger

logger = get_logger("providers.availability")

# Providers that only make sense when a local daemon is running
LOCAL_PROVIDERS: frozenset[str] = frozenset({"ollama"})

_cache: dict[str, tuple[float, bool]] = {}
_CACHE_TTL_SEC = 30.0


def probe_ollama(timeout: float = 0.6) -> bool:
    """Return True if local Ollama responds."""
    base = PROVIDER_BASE_URLS.get("ollama", "http://127.0.0.1:11434/v1")
    root = base.rstrip("/").removesuffix("/v1")
    try:
        with httpx.Client(timeout=timeout) as client:
            # /api/tags is the lightest native health check
            response = client.get(f"{root}/api/tags")
            return response.status_code < 500
    except Exception as exc:  # noqa: BLE001
        logger.debug("Ollama unavailable: %s", exc)
        return False


def is_local_provider_available(provider: str, *, force: bool = False) -> bool:
    """Cached availability check for local-only providers."""
    if provider not in LOCAL_PROVIDERS:
        return True
    now = time.monotonic()
    if not force and provider in _cache:
        ts, ok = _cache[provider]
        if now - ts < _CACHE_TTL_SEC:
            return ok
    if provider == "ollama":
        ok = probe_ollama()
    else:
        ok = False
    _cache[provider] = (now, ok)
    return ok


def get_ui_provider_ids(providers: Iterable[str] | None = None) -> list[str]:
    """Provider ids for settings / role pickers (hide offline local backends)."""
    ids = list(providers) if providers is not None else list(PROVIDER_IDS)
    visible = [p for p in ids if is_local_provider_available(p)]
    return sorted(visible, key=lambda p: PROVIDER_LABELS.get(p, p).casefold())


def invalidate_availability_cache() -> None:
    _cache.clear()