college-infos/pronote_sync/synthesis/__init__.py

"""Factory de sélection du fournisseur de synthèse IA."""

from __future__ import annotations

import logging
from urllib.parse import parse_qsl, urlparse

from pronote_sync.config.settings import AISettings
from pronote_sync.synthesis.openai import OpenAISynthesisProvider
from pronote_sync.synthesis.provider import SynthesisProvider
from pronote_sync.utils.redaction import redact_url

logger = logging.getLogger(__name__)

__all__ = ["get_synthesis_provider", "SynthesisProvider", "OpenAISynthesisProvider"]


_SENSITIVE_QUERY_PARAMS = {"token", "key", "api_key", "secret", "password", "auth"}


def _validate_base_url(provider: str, url: str, allow_insecure_http: bool) -> str | None:
    """Valide structurellement une URL de base IA, partagée entre providers.

    Applique les règles structurelles identiques aux trois providers
    (``openai``, ``litellm`` et ``openai-compatible``) : URL parsable par
    ``urlparse`` (``ValueError`` rejeté), hostname non vide, schéma limité
    à ``http``/``https`` (HTTP refusé sauf si ``allow_insecure_http`` vaut
    ``True``), absence d'identifiants dans le netloc et de paramètres
    sensibles dans la requête (y compris les paramètres sans valeur). L'URL
    est retournée strictement inchangée : aucune manipulation automatique
    du suffixe ``/v1`` n'est effectuée. En cas de violation, un
    avertissement est journalisé (l'URL est toujours masquée via
    :func:`redact_url`) et ``None`` est retourné ; la fonction ne lève
    jamais d'exception et n'effectue aucun appel réseau.

    :param provider: Nom du provider (utilisé pour le message d'avertissement).
    :param url: URL de base à valider (non vide).
    :param allow_insecure_http: Autorise ou non les URLs en HTTP.
    :return: L'URL validée, strictement inchangée, ou ``None`` si invalide.
    :rtype: str | None
    """
    try:
        parsed = urlparse(url)
        if not parsed.scheme:
            logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
            return None
        if not parsed.hostname:
            logger.warning("URL sans hostname pour le provider %s : %s", provider, redact_url(url))
            return None
        if parsed.scheme not in ("http", "https"):
            logger.warning(
                "Schéma d'URL non supporté pour le provider %s : %s", provider, redact_url(url)
            )
            return None
        if parsed.scheme == "http" and not allow_insecure_http:
            logger.warning(
                "URL HTTP non autorisée sans AI_ALLOW_INSECURE_HTTP=true pour le provider %s : %s",
                provider,
                redact_url(url),
            )
            return None
        if parsed.username is not None or parsed.password is not None:
            logger.warning(
                "Credentials dans l'URL refusés pour le provider %s : %s",
                provider,
                redact_url(url),
            )
            return None
        param_names = [name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)]
        if any(name in _SENSITIVE_QUERY_PARAMS for name in param_names):
            logger.warning(
                "Paramètres sensibles dans l'URL refusés pour le provider %s : %s",
                provider,
                redact_url(url),
            )
            return None
        # Accéder à parsed.port peut lever ValueError (port invalide/hors bornes).
        parsed.port  # noqa: B018
    except ValueError:
        logger.warning("URL invalide pour le provider %s : %s", provider, redact_url(url))
        return None
    return url


def _validate_openai_compatible_config(
    url: str | None, model: str | None, allow_insecure_http: bool
) -> str | None:
    """Valide la configuration du provider ``openai-compatible``.

    Vérifie d'abord la présence de l'URL de base et du modèle (spécifique
    à ``openai-compatible``), puis délègue les règles structurelles
    partagées à :func:`_validate_base_url`. En cas d'échec, un
    avertissement est journalisé et ``None`` est retourné : la synthèse IA
    se dégrade silencieusement, sans jamais lever d'exception.

    :param url: URL de base de l'API compatible OpenAI.
    :param model: Identifiant du modèle à utiliser.
    :param allow_insecure_http: Autorise ou non les URLs en HTTP.
    :return: L'URL validée, inchangée (aucune manipulation du chemin ou du
        suffixe ``/v1``), ou ``None`` si la configuration est invalide.
    :rtype: str | None
    """
    if not url:
        logger.warning("URL de base requise pour le provider openai-compatible")
        return None
    if not model:
        logger.warning("Modèle requis pour le provider openai-compatible")
        return None
    return _validate_base_url("openai-compatible", url, allow_insecure_http)


def get_synthesis_provider(settings: AISettings) -> SynthesisProvider | None:
    """Sélectionne le fournisseur de synthèse IA selon la configuration.

    Retourne ``None`` lorsque la synthèse IA est désactivée ou qu'aucune clé
    API n'est configurée. Pour le provider ``litellm``, le paquet ``litellm``
    (extra ``ai-litellm``) est requis : s'il est absent, un avertissement est
    journalisé et ``None`` est retourné. Pour ``openai`` et ``litellm``,
    une ``base_url`` éventuelle est validée par :func:`_validate_base_url` ;
    pour le provider ``openai-compatible``, la configuration (URL de base
    et modèle) est validée par :func:`_validate_openai_compatible_config` ;
    en cas de rejet, ``None`` est retourné avec un avertissement.

    :param settings: Paramètres IA.
    :return: Le fournisseur configuré, ou ``None`` si désactivé, sans clé API
        ou avec une configuration invalide.
    :rtype: SynthesisProvider | None
    """
    if not settings.enabled:
        return None
    if not settings.api_key:
        return None

    base_url = settings.base_url
    model = settings.model or "gpt-4o-mini"

    if settings.provider == "litellm":
        try:
            from pronote_sync.synthesis.litellm import LiteLLMSynthesisProvider
        except ImportError:
            logger.warning("Extra 'ai-litellm' requis pour le provider litellm")
            return None
        if base_url is not None and (
            _validate_base_url("litellm", base_url, settings.allow_insecure_http) is None
        ):
            return None
        return LiteLLMSynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)

    if settings.provider == "openai-compatible":
        url = _validate_openai_compatible_config(
            settings.base_url, settings.model, settings.allow_insecure_http
        )
        if url is None:
            return None
        return OpenAISynthesisProvider(api_key=settings.api_key, base_url=url, model=model)

    if base_url is not None and (
        _validate_base_url("openai", base_url, settings.allow_insecure_http) is None
    ):
        return None
    return OpenAISynthesisProvider(api_key=settings.api_key, base_url=base_url, model=model)