college-infos/pronote_sync/cli/main.py

"""Point d'entrée en ligne de commande du pipeline Pronote → CalDAV → XMPP."""

from __future__ import annotations

import argparse
import logging
import traceback
from collections.abc import Sequence

from pydantic import SecretStr

from pronote_sync.config.env import load_settings
from pronote_sync.config.settings import Settings
from pronote_sync.pipeline.run import PipelineRunner
from pronote_sync.utils.logging import setup_logging
from pronote_sync.utils.redaction import redact_secrets

logger = logging.getLogger(__name__)

_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")


def _parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace:
    """Analyse les options de lancement du programme.

    :param arguments: Arguments à analyser, ou ``None`` pour ceux du processus.
    :return: Options de ligne de commande validées.
    :rtype: argparse.Namespace
    """
    parser = argparse.ArgumentParser(description="Synchronise Pronote vers CalDAV et XMPP.")
    parser.add_argument(
        "--dry-run",
        action="store_true",
        default=None,
        help="Simule la synchronisation sans écrire vers CalDAV ni XMPP.",
    )
    parser.add_argument(
        "--log-level",
        choices=_LOG_LEVELS,
        type=str.upper,
        help="Niveau de verbosité des journaux.",
    )
    return parser.parse_args(arguments)


def _settings_secrets(settings: Settings) -> tuple[SecretStr | str, ...]:
    """Retourne les valeurs sensibles connues pour la rédaction des messages.

    Centraliser ces valeurs garantit que les diagnostics CLI ne divulguent pas
    les secrets configurés, y compris lorsque le niveau ``DEBUG`` est demandé.

    :param settings: Configuration validée de l'application.
    :return: Secrets connus à transmettre au mécanisme de rédaction.
    :rtype: tuple[SecretStr | str, ...]
    """
    candidates = (
        *settings.redaction_secrets(),
        settings.pronote.username,
        settings.caldav.username,
        settings.xmpp.jid,
        settings.xmpp.to,
    )
    return tuple(dict.fromkeys(secret for secret in candidates if secret is not None))


def _safe_traceback(
    exception: BaseException, *, extra_secrets: Sequence[SecretStr | str] = ()
) -> str:
    """Construit une pile complète sans inclure les messages d'exception bruts.

    Les noms de fichiers, lignes et fonctions conservent la valeur de diagnostic
    de la pile. Les messages et les chaînes de causes sont volontairement
    remplacés, car ils peuvent provenir d'une bibliothèque externe.

    :param exception: Exception à représenter sans divulguer son contenu.
    :param extra_secrets: Valeurs sensibles configurées à rédiger dans les cadres.
    :return: Représentation de la pile et de ses causes, expurgée.
    :rtype: str
    """
    lines = ["Traceback (most recent call last):"]
    current: BaseException | None = exception
    seen: set[int] = set()
    while current is not None and id(current) not in seen:
        seen.add(id(current))
        for frame in traceback.extract_tb(current.__traceback__):
            lines.append(f'  File "{frame.filename}", line {frame.lineno}, in {frame.name}')
        lines.append(f"{type(current).__name__}: erreur expurgée")
        next_exception = current.__cause__ or current.__context__
        if next_exception is not None and id(next_exception) not in seen:
            lines.append("La cause ou le contexte précédent est le suivant :")
        current = next_exception
    return redact_secrets("\n".join(lines), extra_secrets=extra_secrets)


def _log_failure(
    message: str,
    exception: BaseException,
    *,
    extra_secrets: Sequence[SecretStr | str] = (),
) -> None:
    """Journalise une erreur et sa pile expurgée uniquement en niveau DEBUG.

    :param message: Message public déjà sûr à afficher hors DEBUG.
    :param exception: Exception dont la pile doit être présentée de façon sûre.
    :param extra_secrets: Valeurs sensibles configurées à rédiger.
    :rtype: None
    """
    logger.error("%s", redact_secrets(message, extra_secrets=extra_secrets))
    if logger.isEnabledFor(logging.DEBUG):
        logger.debug("%s", _safe_traceback(exception, extra_secrets=extra_secrets))


def main(arguments: Sequence[str] | None = None) -> int:
    """Lance le pipeline configuré et retourne son code de sortie.

    En niveau ``DEBUG``, les piles sont affichées sans leurs messages externes
    bruts afin de préserver le diagnostic sans exposer de secret.

    :param arguments: Arguments optionnels, principalement utiles aux appels programmatiques.
    :return: ``0`` en cas de succès, ``1`` sinon (après analyse des arguments).
    :rtype: int
    :raises SystemExit: Si argparse rejette les arguments (code de sortie 2).
    """
    parsed_arguments = _parse_arguments(arguments)
    setup_logging(parsed_arguments.log_level or "INFO")
    try:
        settings = load_settings()
    except Exception as exception:
        _log_failure("Configuration invalide ou indisponible.", exception)
        return 1

    setup_logging(parsed_arguments.log_level or settings.app.log_level)
    try:
        runner = PipelineRunner.from_settings(settings, dry_run=parsed_arguments.dry_run)
        data, errors = runner.run()
    except Exception as exception:
        _log_failure(
            "Échec inattendu du pipeline.",
            exception,
            extra_secrets=_settings_secrets(settings),
        )
        return 1

    secrets = _settings_secrets(settings)
    for error in errors:
        logger.error("%s", redact_secrets(error.message, extra_secrets=secrets))
    if data is None:
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())