college-infos/pronote_sync/synthesis/openai.py

"""Fournisseur de synthèse IA via le SDK ``openai``.

Ce module définit :class:`OpenAISynthesisProvider`, un fournisseur de
synthèse IA qui construit un prompt utilisateur en français à partir des
données de synchronisation et appelle l'API OpenAI via le SDK ``openai``.
La méthode :meth:`OpenAISynthesisProvider.generate` ne lève jamais
d'exception : tout échec est journalisé (message rédigé) et dégradé en
retour ``None``.
"""

from __future__ import annotations

import logging

from openai import OpenAI

from pronote_sync.models.diff import AgendaChangeType
from pronote_sync.models.synthesis import SynthesisInput, SynthesisResult
from pronote_sync.utils.redaction import redact_secrets

logger = logging.getLogger(__name__)

__all__ = ["OpenAISynthesisProvider"]


class OpenAISynthesisProvider:
    """Fournisseur de synthèse IA utilisant le SDK ``openai``.

    Ne lève jamais d'exception : en cas d'échec, :meth:`generate` retourne
    ``None``.
    """

    SYSTEM_PROMPT = (
        "Tu es un assistant qui rédige des synthèses quotidiennes pour les parents d'élèves.\n"
        "Rédige une synthèse en 3 à 5 phrases maximum, dans un ton chaleureux et sobre.\n"
        "N'utilise aucun emoji, aucun titre, aucune liste.\n"
        "Ne mentionne aucun horaire sauf si l'heure est explicitement dans les données.\n"
        "N'invente rien. Base-toi uniquement sur les informations fournies.\n"
        "Si aucune information importante n'est disponible, retourne une chaîne vide."
    )
    MAX_LENGTH = 800
    TIMEOUT = 30
    TEMPERATURE = 0.3

    def __init__(
        self,
        api_key: str,
        base_url: str | None = None,
        model: str = "gpt-4o-mini",
        client: OpenAI | None = None,
    ) -> None:
        """Initialise le fournisseur OpenAI.

        :param api_key: Clé API OpenAI.
        :param base_url: URL de base de l'API (``None`` pour l'URL par défaut).
        :param model: Identifiant du modèle.
        :param client: Client ``OpenAI`` pré-configuré (utilisé par les
            tests). Si ``None``, un client est créé à partir des autres
            paramètres.
        """
        if client is not None:
            self._client = client
        elif base_url is not None:
            self._client = OpenAI(api_key=api_key, base_url=base_url, timeout=self.TIMEOUT)
        else:
            self._client = OpenAI(api_key=api_key, timeout=self.TIMEOUT)
        self._model = model

    @staticmethod
    def _build_prompt(input_data: SynthesisInput) -> str:
        """Construit le prompt utilisateur français à partir des données d'entrée.

        Les informations sont structurées par sections (date cible, changements
        d'agenda, messages non lus, événements scolaires), séparées par des
        sauts de ligne. Si aucune information importante n'est disponible
        (pas de changement, de message non lu ni d'événement), un message par
        défaut est retourné.

        :param input_data: Données de synthèse (diff agenda, messages, événements).
        :return: Prompt utilisateur formaté.
        :rtype: str
        """
        lines: list[str] = [f"Date cible : {input_data.target_date.strftime('%d/%m/%Y')}"]

        if input_data.agenda_diff is not None:
            for change in input_data.agenda_diff.changes:
                if change.type == AgendaChangeType.ADDED and change.lesson is not None:
                    lines.append(f"Cours ajouté : {change.lesson.subject}")
                elif (
                    change.type == AgendaChangeType.REMOVED
                    and change.theoretical_lesson is not None
                ):
                    lines.append(f"Cours supprimé : {change.theoretical_lesson.subject}")
                elif change.type == AgendaChangeType.MODIFIED and change.lesson is not None:
                    lines.append(f"Cours modifié : {change.lesson.subject} ({change.details})")

        for msg in input_data.messages:
            if not msg.read:
                lines.append(f"Message de {msg.author}: {msg.title}")

        for event in input_data.school_events:
            lines.append(f"{event.label} du {event.from_date.strftime('%d/%m')}")

        if len(lines) == 1:
            return "Aucune information importante à signaler."

        return "\n".join(lines)

    def generate(self, input_data: SynthesisInput) -> SynthesisResult | None:
        """Génère une synthèse IA à partir des données d'entrée.

        Construit le prompt via :meth:`_build_prompt`, appelle le modèle et
        nettoie la réponse (troncature à :attr:`MAX_LENGTH`, suppression des
        sauts de ligne en début et fin). Ne lève jamais d'exception : toute
        erreur est journalisée (message rédigé) et dégradée en retour
        ``None``.

        :param input_data: Données de synthèse (diff agenda, messages, événements).
        :return: Résultat de la synthèse, ou ``None`` en cas d'échec ou de
            réponse vide.
        :rtype: SynthesisResult | None
        """
        try:
            prompt = self._build_prompt(input_data)
            response = self._client.chat.completions.create(
                model=self._model,
                messages=[
                    {"role": "system", "content": self.SYSTEM_PROMPT},
                    {"role": "user", "content": prompt},
                ],
                max_tokens=self.MAX_LENGTH,
                temperature=self.TEMPERATURE,
            )
            content = response.choices[0].message.content
            if not content:
                return None
            synthesis_text = content[: self.MAX_LENGTH].strip()
            if not synthesis_text:
                return None
            return SynthesisResult(text=synthesis_text)
        except Exception as e:
            logger.error("Échec de la génération de la synthèse IA : %s", redact_secrets(str(e)))
            return None