feat: orchestrer le pipeline M11
Co-authored-by: Codex/gpt-5.6-terra <codex-gpt-5-6-terra@agents.invalid>
This commit is contained in:
281
pronote_sync/pipeline/run.py
Normal file
281
pronote_sync/pipeline/run.py
Normal file
@@ -0,0 +1,281 @@
|
||||
"""Composition root et orchestrateur du pipeline Pronote → CalDAV → XMPP."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from contextlib import AbstractContextManager, nullcontext
|
||||
from datetime import datetime
|
||||
from typing import Protocol, runtime_checkable
|
||||
|
||||
from pronote_sync.channels import get_channel
|
||||
from pronote_sync.channels.protocol import Channel
|
||||
from pronote_sync.config.settings import Settings
|
||||
from pronote_sync.errors import PipelineCriticalError, PipelineError, PipelineWarning
|
||||
from pronote_sync.models.blog import ExternalInfo
|
||||
from pronote_sync.models.pronote import PronoteData
|
||||
from pronote_sync.models.sync import CalDAVSyncResult, CalDAVSyncStatus
|
||||
from pronote_sync.models.synthesis import SynthesisInput
|
||||
from pronote_sync.models.xmpp import XmppMessage
|
||||
from pronote_sync.pipeline.steps.caldav_sync import CalDAVSynchronizer, caldav_sync_step
|
||||
from pronote_sync.pipeline.steps.compare import compare_step
|
||||
from pronote_sync.pipeline.steps.fetch import fetch_step
|
||||
from pronote_sync.pipeline.steps.fetch_blog import fetch_blog_step
|
||||
from pronote_sync.pipeline.steps.normalize import normalize_step
|
||||
from pronote_sync.pipeline.steps.send import send_step
|
||||
from pronote_sync.pipeline.steps.synthesis import synthesis_step
|
||||
from pronote_sync.sources.blog.rss import BlogRSSClient
|
||||
from pronote_sync.sources.blog.state import BlogRSSState
|
||||
from pronote_sync.sources.pronote.client import PronoteClient
|
||||
from pronote_sync.sources.pronote.fallback import PronoteFetcher, PronoteFetcherProtocol
|
||||
from pronote_sync.sources.theoretical import get_theoretical_provider
|
||||
from pronote_sync.sync.diff import AgendaComparator
|
||||
from pronote_sync.sync.synchronizer import synchronize
|
||||
from pronote_sync.synthesis import get_synthesis_provider
|
||||
from pronote_sync.synthesis.provider import SynthesisProvider
|
||||
from pronote_sync.utils.redaction import redact_exception, redact_secrets
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _synchronize_caldav(data: PronoteData, settings: Settings) -> CalDAVSyncResult:
|
||||
"""Adapte le synchroniseur CalDAV de production au protocole injecté.
|
||||
|
||||
:param data: Données Pronote normalisées à synchroniser.
|
||||
:param settings: Configuration effective de l'exécution.
|
||||
:return: Résultat de la synchronisation CalDAV.
|
||||
:rtype: CalDAVSyncResult
|
||||
"""
|
||||
return synchronize(data, settings)
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class _RunContextFetcher(PronoteFetcherProtocol, Protocol):
|
||||
"""Protocole interne d'un fetcher capable d'isoler un cache par run."""
|
||||
|
||||
def run_context(self) -> AbstractContextManager[None]:
|
||||
"""Retourne le contexte de durée de vie d'une exécution.
|
||||
|
||||
:return: Contexte éphémère associé à l'exécution.
|
||||
:rtype: AbstractContextManager[None]
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class PipelineRunner:
|
||||
"""Orchestre les étapes fetch → normalize → blog → compare → CalDAV → IA → XMPP.
|
||||
|
||||
Toutes les dépendances sont injectables. La méthode :meth:`from_settings`
|
||||
constitue la composition root de production et ne crée aucun singleton.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
settings: Settings,
|
||||
pronote_fetcher: PronoteFetcherProtocol,
|
||||
caldav_synchronizer: CalDAVSynchronizer = _synchronize_caldav,
|
||||
agenda_comparator: AgendaComparator | None = None,
|
||||
synthesis_provider: SynthesisProvider | None = None,
|
||||
channel: Channel | None = None,
|
||||
blog_client: BlogRSSClient | None = None,
|
||||
blog_state: BlogRSSState | None = None,
|
||||
dry_run: bool | None = None,
|
||||
now_provider: Callable[[], datetime] = datetime.now,
|
||||
) -> None:
|
||||
"""Initialise un pipeline entièrement injectable.
|
||||
|
||||
:param settings: Configuration de base du pipeline.
|
||||
:param pronote_fetcher: Source Pronote à utiliser.
|
||||
:param caldav_synchronizer: Service CalDAV injecté.
|
||||
:param agenda_comparator: Comparateur théorique, absent si désactivé.
|
||||
:param synthesis_provider: Fournisseur IA optionnel.
|
||||
:param channel: Canal XMPP optionnel.
|
||||
:param blog_client: Client RSS optionnel.
|
||||
:param blog_state: État RSS associé au client optionnel.
|
||||
:param dry_run: Surcharge optionnelle du mode dry-run de la configuration.
|
||||
:param now_provider: Horloge injectée pour rendre l'exécution testable.
|
||||
"""
|
||||
self._settings = settings
|
||||
self._pronote_fetcher = pronote_fetcher
|
||||
self._caldav_synchronizer = caldav_synchronizer
|
||||
self._agenda_comparator = agenda_comparator
|
||||
self._synthesis_provider = synthesis_provider
|
||||
self._channel = channel
|
||||
self._blog_client = blog_client
|
||||
self._blog_state = blog_state
|
||||
self._dry_run = settings.app.dry_run if dry_run is None else dry_run
|
||||
self._now_provider = now_provider
|
||||
self._errors: list[PipelineError] = []
|
||||
self._warnings: list[PipelineWarning] = []
|
||||
|
||||
@classmethod
|
||||
def from_settings(cls, settings: Settings, *, dry_run: bool | None = None) -> PipelineRunner:
|
||||
"""Construit les dépendances de production sans singleton global.
|
||||
|
||||
:param settings: Configuration validée de l'application.
|
||||
:param dry_run: Surcharge optionnelle du mode dry-run.
|
||||
:return: Pipeline prêt à être exécuté.
|
||||
:rtype: PipelineRunner
|
||||
"""
|
||||
effective_dry_run = settings.app.dry_run if dry_run is None else dry_run
|
||||
theoretical_provider = get_theoretical_provider(
|
||||
settings.app.theoretical_agenda_path,
|
||||
settings.app.school_holidays_path,
|
||||
settings.app.theoretical_week_anchor_date,
|
||||
settings.app.theoretical_week_anchor_type,
|
||||
)
|
||||
comparator = (
|
||||
AgendaComparator(theoretical_provider) if theoretical_provider is not None else None
|
||||
)
|
||||
blog_client = BlogRSSClient(settings.blog.rss_url) if settings.blog.enabled else None
|
||||
blog_state = BlogRSSState() if settings.blog.enabled else None
|
||||
return cls(
|
||||
settings=settings,
|
||||
pronote_fetcher=PronoteFetcher(settings, PronoteClient(settings.pronote)),
|
||||
agenda_comparator=comparator,
|
||||
synthesis_provider=get_synthesis_provider(settings.ai),
|
||||
channel=get_channel(settings.xmpp, dry_run=effective_dry_run),
|
||||
blog_client=blog_client,
|
||||
blog_state=blog_state,
|
||||
dry_run=effective_dry_run,
|
||||
)
|
||||
|
||||
def _effective_settings(self) -> Settings:
|
||||
"""Retourne la configuration dont le dry-run reflète l'exécution courante.
|
||||
|
||||
:return: Copie de configuration à passer aux dépendances.
|
||||
:rtype: Settings
|
||||
"""
|
||||
if self._settings.app.dry_run == self._dry_run:
|
||||
return self._settings
|
||||
return self._settings.model_copy(
|
||||
update={"app": self._settings.app.model_copy(update={"dry_run": self._dry_run})}
|
||||
)
|
||||
|
||||
def _run_context(self) -> AbstractContextManager[None]:
|
||||
"""Retourne le contexte isolant les éventuels caches de source.
|
||||
|
||||
:return: Contexte de durée de vie du run, vide pour un fetcher générique.
|
||||
:rtype: AbstractContextManager[None]
|
||||
"""
|
||||
if isinstance(self._pronote_fetcher, _RunContextFetcher):
|
||||
return self._pronote_fetcher.run_context()
|
||||
return nullcontext()
|
||||
|
||||
def _warn(self, step: str, message: str) -> None:
|
||||
"""Enregistre et journalise un avertissement expurgé.
|
||||
|
||||
:param step: Étape ayant échoué.
|
||||
:param message: Message déjà expurgé.
|
||||
"""
|
||||
warning = PipelineWarning(message, step=step)
|
||||
self._warnings.append(warning)
|
||||
logger.warning("Étape %s dégradée : %s", step, warning.message)
|
||||
|
||||
def run(self) -> tuple[PronoteData | None, list[PipelineError]]:
|
||||
"""Exécute le pipeline complet dans l'ordre contractuel.
|
||||
|
||||
Une erreur de récupération critique interrompt l'exécution. Les erreurs
|
||||
des étapes facultatives sont converties en :class:`PipelineWarning` afin
|
||||
que les étapes suivantes, notamment XMPP, restent exécutées.
|
||||
|
||||
:return: Données Pronote normalisées ou ``None``, puis erreurs et avertissements.
|
||||
:rtype: tuple[PronoteData | None, list[PipelineError]]
|
||||
"""
|
||||
self._errors = []
|
||||
self._warnings = []
|
||||
now = self._now_provider()
|
||||
effective_settings = self._effective_settings()
|
||||
try:
|
||||
with self._run_context():
|
||||
fetched, fetch_warnings = fetch_step(self._pronote_fetcher, today=now.date())
|
||||
self._warnings.extend(fetch_warnings)
|
||||
data = normalize_step(fetched, generated_at=now)
|
||||
|
||||
try:
|
||||
blog_articles = fetch_blog_step(self._blog_client, self._blog_state)
|
||||
except Exception as exc:
|
||||
self._warn("fetch_blog", redact_exception(exc))
|
||||
blog_articles = []
|
||||
|
||||
try:
|
||||
agenda_diff = compare_step(self._agenda_comparator, data)
|
||||
except Exception as exc:
|
||||
self._warn("compare", redact_exception(exc))
|
||||
from pronote_sync.models.diff import AgendaDiff
|
||||
|
||||
agenda_diff = AgendaDiff(target_date=data.target_date)
|
||||
|
||||
try:
|
||||
sync_result = caldav_sync_step(
|
||||
self._caldav_synchronizer, data, effective_settings
|
||||
)
|
||||
if sync_result.status is CalDAVSyncStatus.FAILED:
|
||||
caldav_errors = redact_secrets(
|
||||
"; ".join(sync_result.errors),
|
||||
extra_secrets=(effective_settings.caldav.password,)
|
||||
if effective_settings.caldav.password is not None
|
||||
else (),
|
||||
)
|
||||
self._warn("caldav_sync", caldav_errors or "Échec CalDAV")
|
||||
except Exception as exc:
|
||||
self._warn("caldav_sync", redact_exception(exc))
|
||||
|
||||
try:
|
||||
synthesis = synthesis_step(
|
||||
self._synthesis_provider,
|
||||
SynthesisInput(
|
||||
agenda_diff=agenda_diff,
|
||||
messages=data.messages,
|
||||
school_events=data.school_events,
|
||||
target_date=data.target_date,
|
||||
),
|
||||
)
|
||||
except Exception as exc:
|
||||
self._warn("synthesis", redact_exception(exc))
|
||||
synthesis = None
|
||||
|
||||
message = XmppMessage(
|
||||
target_date=data.target_date,
|
||||
synthesis=synthesis.text if synthesis is not None else None,
|
||||
homeworks=tuple(data.homeworks),
|
||||
changes=agenda_diff.changes,
|
||||
messages=tuple(data.messages),
|
||||
external_info=ExternalInfo(blog_articles=tuple(blog_articles))
|
||||
if blog_articles
|
||||
else None,
|
||||
)
|
||||
if self._channel is not None and not self._dry_run:
|
||||
try:
|
||||
if not send_step(self._channel, message):
|
||||
self._warn("send", "Le canal XMPP a refusé l'envoi")
|
||||
except Exception as exc:
|
||||
self._warn("send", redact_exception(exc))
|
||||
return data, [*self._errors, *self._warnings]
|
||||
except PipelineCriticalError as exc:
|
||||
logger.error("Erreur critique du pipeline : %s", exc.message)
|
||||
self._errors.append(exc)
|
||||
except Exception as exc:
|
||||
error = PipelineCriticalError(
|
||||
f"Erreur inattendue du pipeline : {redact_exception(exc)}", step="pipeline"
|
||||
)
|
||||
logger.error("Erreur critique du pipeline : %s", error.message)
|
||||
self._errors.append(error)
|
||||
return None, [*self._errors, *self._warnings]
|
||||
|
||||
def get_errors(self) -> list[PipelineError]:
|
||||
"""Retourne les erreurs critiques de la dernière exécution.
|
||||
|
||||
:return: Copie des erreurs critiques.
|
||||
:rtype: list[PipelineError]
|
||||
"""
|
||||
return list(self._errors)
|
||||
|
||||
def get_warnings(self) -> list[PipelineWarning]:
|
||||
"""Retourne les avertissements de la dernière exécution.
|
||||
|
||||
:return: Copie des avertissements non bloquants.
|
||||
:rtype: list[PipelineWarning]
|
||||
"""
|
||||
return list(self._warnings)
|
||||
Reference in New Issue
Block a user