pyssh-client/app/ai/provider.py

"""Abstraction du fournisseur IA : une API compatible OpenAI (endpoint chat
completions), avec URL de base et modele configurables - pas de liste de
fournisseurs ou de modeles figee en dur, pour supporter OpenAI, OpenRouter, un
proxy interne ou un LLM local exposant la meme API.

AIProvider porte les 4 usages metier (explain_output/diagnose/generate_command/
ask), tous construits au-dessus de chat(), seule methode que doit implementer
un fournisseur concret (OpenAICompatibleProvider)."""
from __future__ import annotations

import abc
import json
import urllib.error
import urllib.request


class AIError(Exception):
    """Message deja comprehensible, pret a etre affiche a l'utilisateur."""


_SYSTEM_PROMPT_TROUBLESHOOT = """Tu es un assistant specialise en administration systeme, Linux, Windows,
reseau, SSH, Docker, virtualisation et troubleshooting.

Analyse la sortie terminal fournie par l'utilisateur.

Explique :
1. ce que signifie la sortie ;
2. la cause probable ;
3. les verifications recommandees ;
4. les commandes eventuellement utiles.

Ne suppose jamais qu'une commande a ete executee.
Ne pretends jamais connaitre des elements qui ne sont pas presents.
Signale explicitement les hypotheses."""

_SYSTEM_PROMPT_DIAGNOSE = _SYSTEM_PROMPT_TROUBLESHOOT + """

Structure ta reponse ainsi :
Cause probable
Explication
Verifications
Commandes proposees
Niveau de risque"""

_SYSTEM_PROMPT_COMMAND = """Tu es un assistant specialise en administration systeme, Linux, Windows,
reseau, SSH, Docker et virtualisation.

L'utilisateur decrit ce qu'il veut accomplir en langage naturel. Propose UNE
commande (ou une courte suite de commandes) qui repond a la demande, dans un
bloc de code, avec une breve explication de ce qu'elle fait.

Ne suppose jamais le systeme d'exploitation ou l'environnement si l'utilisateur
ne l'a pas precise : demande-le, ou propose une variante par systeme si
pertinent. La commande proposee ne sera jamais executee automatiquement : elle
sera seulement affichee a l'utilisateur pour verification."""


class AIProvider(abc.ABC):
    @abc.abstractmethod
    def chat(self, messages: list[dict], *, temperature: float | None = None) -> str:
        """Renvoie le texte de la reponse. Leve AIError en cas d'echec."""

    def explain_output(self, output: str) -> str:
        return self.chat([
            {"role": "system", "content": _SYSTEM_PROMPT_TROUBLESHOOT},
            {"role": "user", "content": f"Explique cette sortie terminal :\n\n{output}"},
        ])

    def diagnose(self, output: str) -> str:
        return self.chat([
            {"role": "system", "content": _SYSTEM_PROMPT_DIAGNOSE},
            {"role": "user", "content": f"Diagnostique cette erreur :\n\n{output}"},
        ])

    def propose_solution(self, output: str) -> str:
        return self.chat([
            {"role": "system", "content": _SYSTEM_PROMPT_TROUBLESHOOT},
            {"role": "user", "content": (
                "Propose une solution concrete a ce probleme, avec les commandes "
                f"eventuellement necessaires :\n\n{output}"
            )},
        ])

    def generate_command(self, request: str) -> str:
        return self.chat([
            {"role": "system", "content": _SYSTEM_PROMPT_COMMAND},
            {"role": "user", "content": request},
        ])

    def ask(self, question: str, context: str = "") -> str:
        messages = [{"role": "system", "content": _SYSTEM_PROMPT_TROUBLESHOOT}]
        if context:
            messages.append({"role": "user", "content": f"Contexte (sortie terminal selectionnee) :\n{context}"})
        messages.append({"role": "user", "content": question})
        return self.chat(messages)


class OpenAICompatibleProvider(AIProvider):
    def __init__(
        self, base_url: str, api_key: str, model: str,
        timeout: float = 30.0, temperature: float = 0.3,
    ):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.model = model
        self.timeout = timeout
        self.temperature = temperature

    def chat(self, messages: list[dict], *, temperature: float | None = None) -> str:
        if not self.base_url:
            raise AIError("Aucune URL de base configuree pour l'assistant IA.")
        if not self.model:
            raise AIError("Aucun modele configure pour l'assistant IA.")

        url = f"{self.base_url}/chat/completions"
        payload = {
            "model": self.model,
            "messages": messages,
            "temperature": self.temperature if temperature is None else temperature,
        }
        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        request = urllib.request.Request(
            url, data=json.dumps(payload).encode("utf-8"), headers=headers, method="POST",
        )

        try:
            with urllib.request.urlopen(request, timeout=self.timeout) as response:
                raw = response.read()
        except urllib.error.HTTPError as exc:
            raise AIError(self._describe_http_error(exc)) from exc
        except urllib.error.URLError as exc:
            if isinstance(exc.reason, TimeoutError):
                raise AIError("Le fournisseur IA n'a pas repondu a temps (delai depasse).") from exc
            raise AIError(f"Impossible de joindre le fournisseur IA : {exc.reason}") from exc

        try:
            parsed = json.loads(raw.decode("utf-8"))
            return parsed["choices"][0]["message"]["content"]
        except (json.JSONDecodeError, KeyError, IndexError, TypeError, UnicodeDecodeError) as exc:
            raise AIError(f"Reponse du fournisseur IA invalide ou inattendue : {exc}") from exc

    @staticmethod
    def _describe_http_error(exc: urllib.error.HTTPError) -> str:
        if exc.code == 401:
            return "Authentification refusee (401) : verifiez la cle API."
        if exc.code == 429:
            return "Trop de requetes (429) : le fournisseur limite le debit, reessayez plus tard."
        try:
            body = exc.read().decode("utf-8", errors="replace")
        except Exception:  # noqa: BLE001
            body = ""
        return f"Erreur du fournisseur IA ({exc.code}) : {(body or exc.reason)[:300]}"