"""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)