49  TP récapitulatif — Parseur et analyseur de logs

Ce TP final du niveau Expert mobilise toutes les notions avancées : regex, générateurs, décorateurs, POO, exceptions, fichiers, type hints, tests unitaires. Vous allez construire un analyseur de logs digne d’un outil pro — utilisable en vrai.

🎯 Objectifs pédagogiques
  • Parser un format de log texte avec des regex.
  • Utiliser des générateurs pour traiter des fichiers potentiellement énormes (lazy).
  • Créer des classes typées avec @dataclass et type hints.
  • Écrire des décorateurs (chrono, logging).
  • Tester unitairement chaque composant.
  • Produire un rapport structuré en CLI.

Temps estimé : 3 h 30.

49.1 Contexte

Vous héritez d’un fichier access.log produit par un serveur web. Chaque ligne ressemble à :

2026-04-22 14:30:12 INFO  GET /api/users 200 42ms
2026-04-22 14:30:15 WARN  POST /login 401 15ms
2026-04-22 14:30:18 ERROR GET /api/orders 500 3200ms

Format : date heure NIVEAU METHODE URL CODE DUREEms.

Vous devez :

  • Lire ce log (potentiellement énorme : plusieurs Go).
  • Parser chaque ligne en objet structuré.
  • Filtrer, agréger, analyser les données.
  • Produire un rapport (top URLs, distribution des codes, erreurs les plus fréquentes).
Méthode
  1. Un fichier par module : parseur.py, rapport.py, cli.py, tests/.
  2. Testez chaque fonction avant de passer à la suite.
  3. Typez tout le code public — vérifiable avec mypy --strict.
  4. Utilisez les générateurs pour ne rien charger en mémoire inutilement.

Partie 1 — La classe LogEntry

Consignes

Créez une @dataclass typée représentant une ligne de log.

  • horodatage: datetime
  • niveau: Literal["DEBUG", "INFO", "WARN", "ERROR"]
  • methode: str
  • url: str
  • code: int
  • duree_ms: int

Ajoutez des properties pour :

  • est_erreur : True si code >= 400.
  • est_lent : True si duree_ms > 1000.

Exception personnalisée LigneInvalideError pour signaler un parsing qui échoue.

from dataclasses import dataclass
from datetime import datetime
from typing import Literal


NiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]


class LigneInvalideError(Exception):
    """Levée quand une ligne de log ne peut pas être parsée."""
    pass


@dataclass
class LogEntry:
    horodatage: datetime
    niveau: NiveauLog
    methode: str
    url: str
    code: int
    duree_ms: int

    @property
    def est_erreur(self) -> bool:
        return self.code >= 400

    @property
    def est_lent(self) -> bool:
        return self.duree_ms > 1000

    def __str__(self) -> str:
        emoji = "🔴" if self.est_erreur else "🟡" if self.est_lent else "🟢"
        return (f"{emoji} {self.horodatage:%Y-%m-%d %H:%M:%S} "
                f"{self.niveau:<5} {self.methode} {self.url} "
                f"[{self.code}] {self.duree_ms}ms")


# Test
entree = LogEntry(
    horodatage=datetime(2026, 4, 22, 14, 30, 12),
    niveau="INFO",
    methode="GET",
    url="/api/users",
    code=200,
    duree_ms=42,
)

print(entree)
print(f"Est erreur ? {entree.est_erreur}")
print(f"Est lent ?   {entree.est_lent}")

erreur = LogEntry(
    horodatage=datetime(2026, 4, 22, 14, 30, 18),
    niveau="ERROR",
    methode="GET",
    url="/api/orders",
    code=500,
    duree_ms=3200,
)
print(erreur)
print(f"Est erreur ? {erreur.est_erreur}")
print(f"Est lent ?   {erreur.est_lent}")
🟢 2026-04-22 14:30:12 INFO  GET /api/users [200] 42ms
Est erreur ? False
Est lent ?   False
🔴 2026-04-22 14:30:18 ERROR GET /api/orders [500] 3200ms
Est erreur ? True
Est lent ?   True

Points à retenir

  • Literal["DEBUG", ...] restreint les valeurs acceptées de niveau (mypy le vérifie).
  • @property pour des attributs calculés à la volée — sans stockage.
  • @dataclass génère __init__, __repr__, __eq__ automatiquement.
  • Exception personnalisée finissant par Error (convention PEP 8).

Partie 2 — Parser une ligne avec regex

Consignes

Écrivez parser_ligne(ligne: str) -> LogEntry qui extrait les champs d’une ligne.

Format à matcher :

YYYY-MM-DD HH:MM:SS NIVEAU METHODE URL CODE DUREEms

Levez LigneInvalideError si la ligne ne correspond pas au format.

Indice : utilisez re.compile pour performance, avec des groupes nommés (?P<nom>...).

import re
from dataclasses import dataclass
from datetime import datetime
from typing import Literal

NiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]


class LigneInvalideError(Exception):
    pass


@dataclass
class LogEntry:
    horodatage: datetime
    niveau: NiveauLog
    methode: str
    url: str
    code: int
    duree_ms: int

    @property
    def est_erreur(self) -> bool:
        return self.code >= 400


# Motif avec groupes nommés
MOTIF_LOG = re.compile(
    r"(?P<date>\d{4}-\d{2}-\d{2})\s+"
    r"(?P<heure>\d{2}:\d{2}:\d{2})\s+"
    r"(?P<niveau>DEBUG|INFO|WARN|ERROR)\s+"
    r"(?P<methode>GET|POST|PUT|DELETE|PATCH)\s+"
    r"(?P<url>\S+)\s+"
    r"(?P<code>\d{3})\s+"
    r"(?P<duree>\d+)ms"
)


def parser_ligne(ligne: str) -> LogEntry:
    """Parse une ligne de log en LogEntry, ou lève LigneInvalideError."""
    ligne = ligne.strip()
    match = MOTIF_LOG.match(ligne)

    if match is None:
        raise LigneInvalideError(f"Ligne invalide : {ligne!r}")

    try:
        horodatage = datetime.strptime(
            f"{match['date']} {match['heure']}",
            "%Y-%m-%d %H:%M:%S"
        )
    except ValueError as e:
        raise LigneInvalideError(f"Date invalide : {e}") from e

    return LogEntry(
        horodatage=horodatage,
        niveau=match["niveau"],   # type: ignore[arg-type]
        methode=match["methode"],
        url=match["url"],
        code=int(match["code"]),
        duree_ms=int(match["duree"]),
    )


# Tests
lignes_valides = [
    "2026-04-22 14:30:12 INFO  GET /api/users 200 42ms",
    "2026-04-22 14:30:15 WARN  POST /login 401 15ms",
    "2026-04-22 14:30:18 ERROR GET /api/orders 500 3200ms",
]

for l in lignes_valides:
    entry = parser_ligne(l)
    print(f"  {entry.methode:<4} {entry.url:<15} [{entry.code}]")
  GET  /api/users      [200]
  POST /login          [401]
  GET  /api/orders     [500]

Test des cas invalides

lignes_invalides = [
    "pas du tout un log",
    "2026-04-22 14:30:12 INFO",          # incomplet
    "2026-99-99 14:30:12 INFO GET /x 200 10ms",    # date invalide
]

for l in lignes_invalides:
    try:
        parser_ligne(l)
    except LigneInvalideError as e:
        print(f"  ❌ {l[:30]}... → {e}")
  ❌ pas du tout un log... → Ligne invalide : 'pas du tout un log'
  ❌ 2026-04-22 14:30:12 INFO... → Ligne invalide : '2026-04-22 14:30:12 INFO'
  ❌ 2026-99-99 14:30:12 INFO GET /... → Date invalide : time data '2026-99-99 14:30:12' does not match format '%Y-%m-%d %H:%M:%S'

Points à retenir

  • Groupes nommés (?P<nom>...) : accès par nom plutôt qu’indice.
  • match[\"nom\"] : équivalent à match.group(\"nom\").
  • re.compile une seule fois : plus rapide si la fonction est appelée souvent.
  • raise ... from e : préserve la chaîne d’exceptions pour le debug.
  • Le type: ignore[arg-type] est un petit compromis : mypy ne peut pas prouver que la regex renvoie un des 4 niveaux valides — on le sait, lui pas.

Partie 3 — Lire un fichier en générateur

Consignes

Écrivez lire_logs(chemin: Path) -> Iterator[LogEntry] qui :

  • Ouvre le fichier proprement avec with.
  • Itère ligne par ligne (pas readlines !).
  • Ignore les lignes vides.
  • Saute les lignes invalides en affichant un avertissement sur stderr.
  • Renvoie les LogEntry un par un via yield.

Indice : utilisez from collections.abc import Iterator pour l’annotation.

import re
import sys
from collections.abc import Iterator
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Literal

NiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]


class LigneInvalideError(Exception): pass


@dataclass
class LogEntry:
    horodatage: datetime
    niveau: NiveauLog
    methode: str
    url: str
    code: int
    duree_ms: int

    @property
    def est_erreur(self) -> bool:
        return self.code >= 400


MOTIF_LOG = re.compile(
    r"(?P<date>\d{4}-\d{2}-\d{2})\s+"
    r"(?P<heure>\d{2}:\d{2}:\d{2})\s+"
    r"(?P<niveau>DEBUG|INFO|WARN|ERROR)\s+"
    r"(?P<methode>GET|POST|PUT|DELETE|PATCH)\s+"
    r"(?P<url>\S+)\s+"
    r"(?P<code>\d{3})\s+"
    r"(?P<duree>\d+)ms"
)


def parser_ligne(ligne: str) -> LogEntry:
    ligne = ligne.strip()
    match = MOTIF_LOG.match(ligne)
    if match is None:
        raise LigneInvalideError(f"Ligne invalide : {ligne!r}")
    horodatage = datetime.strptime(
        f"{match['date']} {match['heure']}", "%Y-%m-%d %H:%M:%S"
    )
    return LogEntry(
        horodatage=horodatage,
        niveau=match["niveau"],     # type: ignore[arg-type]
        methode=match["methode"],
        url=match["url"],
        code=int(match["code"]),
        duree_ms=int(match["duree"]),
    )


def lire_logs(chemin: Path, verbose: bool = False) -> Iterator[LogEntry]:
    """Générateur d'entrées depuis un fichier log.

    Les lignes vides sont ignorées, les lignes invalides sont signalées
    sur stderr mais ne stoppent pas la lecture.
    """
    with open(chemin, "r", encoding="utf-8") as f:
        for numero, ligne in enumerate(f, start=1):
            if not ligne.strip():
                continue
            try:
                yield parser_ligne(ligne)
            except LigneInvalideError as e:
                if verbose:
                    print(f"[Ligne {numero}] ignorée : {e}", file=sys.stderr)


# Créer un fichier de test
chemin_test = Path("access.log")
contenu = """2026-04-22 14:30:12 INFO  GET /api/users 200 42ms
2026-04-22 14:30:15 WARN  POST /login 401 15ms

2026-04-22 14:30:18 ERROR GET /api/orders 500 3200ms
ligne_corrompue
2026-04-22 14:30:22 INFO  GET /api/products 200 78ms
2026-04-22 14:30:25 ERROR POST /api/checkout 500 250ms
"""
chemin_test.write_text(contenu, encoding="utf-8")

# Utilisation : lazy, rien n'est stocké en mémoire
for entry in lire_logs(chemin_test):
    statut = "🔴" if entry.est_erreur else "🟢"
    print(f"  {statut} {entry.methode} {entry.url} [{entry.code}]")

# Nettoyage
chemin_test.unlink()
  🟢 GET /api/users [200]
  🔴 POST /login [401]
  🔴 GET /api/orders [500]
  🟢 GET /api/products [200]
  🔴 POST /api/checkout [500]

Points à retenir

  • Iterator[LogEntry] (ou Generator[LogEntry, None, None]) pour typer un générateur.
  • with open(...) as f: garantit la fermeture.
  • for ligne in f: : lecture streamée, mémoire constante.
  • Résilience : on continue même si une ligne est invalide (avec log sur stderr).
  • enumerate(f, start=1) pour numéroter les lignes depuis 1.
  • file=sys.stderr pour les messages d’erreur (séparés de stdout).

Partie 4 — Décorateur de chronométrage

Consignes

Écrivez un décorateur @chrono qui :

  • Mesure le temps d’exécution de la fonction.
  • Affiche "{fonction} terminée en XX ms" sur stderr.
  • Préserve le nom et la docstring avec functools.wraps.
  • Est générique : fonctionne sur n’importe quelle fonction (*args, **kwargs).

Appliquez-le à une fonction d’analyse.

import time
import sys
from functools import wraps
from typing import Callable, TypeVar, ParamSpec
from collections.abc import Iterator
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Literal


P = ParamSpec("P")
R = TypeVar("R")


def chrono(fonction: Callable[P, R]) -> Callable[P, R]:
    """Chronomètre une fonction et affiche la durée."""
    @wraps(fonction)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        debut = time.perf_counter()
        resultat = fonction(*args, **kwargs)
        duree = (time.perf_counter() - debut) * 1000
        print(f"[chrono] {fonction.__name__} terminée en {duree:.1f} ms",
              file=sys.stderr)
        return resultat
    return wrapper


# Préparer un fichier de test plus gros
NiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]


@dataclass
class LogEntry:
    horodatage: datetime
    niveau: NiveauLog
    methode: str
    url: str
    code: int
    duree_ms: int

    @property
    def est_erreur(self) -> bool:
        return self.code >= 400


import re
MOTIF_LOG = re.compile(
    r"(?P<date>\d{4}-\d{2}-\d{2})\s+"
    r"(?P<heure>\d{2}:\d{2}:\d{2})\s+"
    r"(?P<niveau>DEBUG|INFO|WARN|ERROR)\s+"
    r"(?P<methode>GET|POST|PUT|DELETE|PATCH)\s+"
    r"(?P<url>\S+)\s+"
    r"(?P<code>\d{3})\s+"
    r"(?P<duree>\d+)ms"
)


class LigneInvalideError(Exception): pass


def parser_ligne(ligne: str) -> LogEntry:
    ligne = ligne.strip()
    match = MOTIF_LOG.match(ligne)
    if match is None:
        raise LigneInvalideError(f"Ligne invalide")
    horodatage = datetime.strptime(
        f"{match['date']} {match['heure']}", "%Y-%m-%d %H:%M:%S"
    )
    return LogEntry(
        horodatage=horodatage,
        niveau=match["niveau"],     # type: ignore[arg-type]
        methode=match["methode"],
        url=match["url"],
        code=int(match["code"]),
        duree_ms=int(match["duree"]),
    )


def lire_logs(chemin: Path) -> Iterator[LogEntry]:
    with open(chemin, "r", encoding="utf-8") as f:
        for ligne in f:
            if not ligne.strip():
                continue
            try:
                yield parser_ligne(ligne)
            except LigneInvalideError:
                continue


# Créer un fichier de 5000 lignes pour voir le chrono travailler
import random
random.seed(42)

chemin_test = Path("gros.log")
niveaux = ["DEBUG", "INFO", "WARN", "ERROR"]
methodes = ["GET", "POST", "PUT", "DELETE"]
urls = ["/api/users", "/api/orders", "/login", "/api/products", "/api/checkout"]

with open(chemin_test, "w", encoding="utf-8") as f:
    for i in range(5000):
        h = 10 + i // 600
        m = (i // 10) % 60
        s = i % 60
        niveau = random.choice(niveaux)
        methode = random.choice(methodes)
        url = random.choice(urls)
        code = random.choice([200, 200, 200, 401, 404, 500])
        duree = random.randint(10, 5000)
        f.write(f"2026-04-22 {h:02d}:{m:02d}:{s:02d} {niveau:<5} "
                f"{methode} {url} {code} {duree}ms\n")


@chrono
def compter_toutes_entrees(chemin: Path) -> int:
    """Compte le nombre d'entrées valides dans un log."""
    return sum(1 for _ in lire_logs(chemin))


@chrono
def compter_erreurs(chemin: Path) -> int:
    """Compte les erreurs (code >= 400)."""
    return sum(1 for e in lire_logs(chemin) if e.est_erreur)


# Tests
n_total = compter_toutes_entrees(chemin_test)
n_err = compter_erreurs(chemin_test)

print(f"\nTotal : {n_total}")
print(f"Erreurs : {n_err}")
print(f"Taux d'erreur : {n_err / n_total * 100:.1f} %")

# Nettoyage
chemin_test.unlink()

Total : 5000
Erreurs : 2498
Taux d'erreur : 50.0 %

Points à retenir

  • ParamSpec + TypeVar pour un décorateur générique typé (Python 3.10+).
  • time.perf_counter() plus précis que time.time() pour des mesures.
  • @wraps obligatoire pour préserver __name__, __doc__.
  • Les messages de chronométrage vont sur stderr (séparés du résultat sur stdout).
  • On voit ici le bénéfice des générateurs : on compte 5000 entrées sans jamais les stocker toutes en mémoire.

Partie 5 — Classe AnalyseurLogs

Consignes

Créez une classe AnalyseurLogs qui encapsule les statistiques.

Méthodes à implémenter :

  • nb_total() -> int
  • nb_par_niveau() -> dict[str, int]
  • nb_par_code() -> dict[int, int]
  • top_urls(n: int = 5) -> list[tuple[str, int]]
  • top_urls_lentes(n: int = 5) -> list[tuple[str, float]] — URL avec durée moyenne la plus élevée.
  • taux_erreur() -> float — pourcentage d’erreurs.
  • rapport() -> str — rapport textuel formaté.
from collections import Counter, defaultdict
from collections.abc import Iterable
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Literal
import re


NiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]


@dataclass
class LogEntry:
    horodatage: datetime
    niveau: NiveauLog
    methode: str
    url: str
    code: int
    duree_ms: int

    @property
    def est_erreur(self) -> bool:
        return self.code >= 400


MOTIF_LOG = re.compile(
    r"(?P<date>\d{4}-\d{2}-\d{2})\s+"
    r"(?P<heure>\d{2}:\d{2}:\d{2})\s+"
    r"(?P<niveau>DEBUG|INFO|WARN|ERROR)\s+"
    r"(?P<methode>GET|POST|PUT|DELETE|PATCH)\s+"
    r"(?P<url>\S+)\s+"
    r"(?P<code>\d{3})\s+"
    r"(?P<duree>\d+)ms"
)


class LigneInvalideError(Exception): pass


def parser_ligne(ligne: str) -> LogEntry:
    ligne = ligne.strip()
    match = MOTIF_LOG.match(ligne)
    if match is None:
        raise LigneInvalideError(f"Ligne invalide")
    horodatage = datetime.strptime(
        f"{match['date']} {match['heure']}", "%Y-%m-%d %H:%M:%S"
    )
    return LogEntry(
        horodatage=horodatage,
        niveau=match["niveau"],     # type: ignore[arg-type]
        methode=match["methode"],
        url=match["url"],
        code=int(match["code"]),
        duree_ms=int(match["duree"]),
    )


class AnalyseurLogs:
    """Calcule des statistiques sur un ensemble d'entrées de log."""

    def __init__(self, entrees: Iterable[LogEntry]) -> None:
        # On consomme l'itérable et on stocke (analyse en mémoire)
        self.entrees: list[LogEntry] = list(entrees)

    @classmethod
    def depuis_fichier(cls, chemin: Path) -> "AnalyseurLogs":
        """Constructeur alternatif : charge un fichier log."""
        from itertools import islice  # juste pour être propre
        with open(chemin, "r", encoding="utf-8") as f:
            entrees: list[LogEntry] = []
            for ligne in f:
                if not ligne.strip():
                    continue
                try:
                    entrees.append(parser_ligne(ligne))
                except LigneInvalideError:
                    continue
        return cls(entrees)

    def nb_total(self) -> int:
        return len(self.entrees)

    def nb_par_niveau(self) -> dict[str, int]:
        return dict(Counter(e.niveau for e in self.entrees))

    def nb_par_code(self) -> dict[int, int]:
        return dict(Counter(e.code for e in self.entrees))

    def top_urls(self, n: int = 5) -> list[tuple[str, int]]:
        return Counter(e.url for e in self.entrees).most_common(n)

    def top_urls_lentes(self, n: int = 5) -> list[tuple[str, float]]:
        durees: dict[str, list[int]] = defaultdict(list)
        for e in self.entrees:
            durees[e.url].append(e.duree_ms)
        moyennes = [(url, sum(d) / len(d)) for url, d in durees.items()]
        moyennes.sort(key=lambda t: t[1], reverse=True)
        return moyennes[:n]

    def taux_erreur(self) -> float:
        if not self.entrees:
            return 0.0
        erreurs = sum(1 for e in self.entrees if e.est_erreur)
        return erreurs / len(self.entrees) * 100

    def rapport(self) -> str:
        lignes: list[str] = []
        lignes.append("=" * 50)
        lignes.append("  RAPPORT D'ANALYSE DES LOGS")
        lignes.append("=" * 50)
        lignes.append(f"Total entrées   : {self.nb_total()}")
        lignes.append(f"Taux d'erreur   : {self.taux_erreur():.2f} %")

        lignes.append("\nPar niveau :")
        for niveau, n in sorted(self.nb_par_niveau().items()):
            lignes.append(f"  {niveau:<6} {n}")

        lignes.append("\nPar code HTTP :")
        for code, n in sorted(self.nb_par_code().items()):
            lignes.append(f"  {code}    {n}")

        lignes.append("\nTop 5 URLs (plus fréquentes) :")
        for url, n in self.top_urls(5):
            lignes.append(f"  {url:<25} {n}")

        lignes.append("\nTop 5 URLs (plus lentes en moyenne) :")
        for url, moy in self.top_urls_lentes(5):
            lignes.append(f"  {url:<25} {moy:.0f} ms")

        lignes.append("=" * 50)
        return "\n".join(lignes)


# Démo : générer un jeu de logs
import random
random.seed(42)

chemin = Path("demo.log")
niveaux = ["DEBUG", "INFO", "WARN", "ERROR"]
urls_sim = ["/api/users", "/api/orders", "/login", "/api/products", "/api/checkout", "/health"]

with open(chemin, "w", encoding="utf-8") as f:
    for i in range(1000):
        h = 10 + i // 200
        m = (i // 10) % 60
        s = i % 60
        niveau = random.choices(niveaux, weights=[10, 60, 20, 10])[0]
        methode = random.choice(["GET", "POST"])
        url = random.choice(urls_sim)
        code = random.choices([200, 201, 401, 404, 500], weights=[70, 10, 5, 5, 10])[0]
        duree = random.randint(5, 2000)
        f.write(f"2026-04-22 {h:02d}:{m:02d}:{s:02d} {niveau:<5} "
                f"{methode} {url} {code} {duree}ms\n")


analyseur = AnalyseurLogs.depuis_fichier(chemin)
print(analyseur.rapport())

chemin.unlink()
==================================================
  RAPPORT D'ANALYSE DES LOGS
==================================================
Total entrées   : 1000
Taux d'erreur   : 20.10 %

Par niveau :
  DEBUG  87
  ERROR  91
  INFO   619
  WARN   203

Par code HTTP :
  200    700
  201    99
  401    53
  404    55
  500    93

Top 5 URLs (plus fréquentes) :
  /api/orders               182
  /api/checkout             173
  /api/users                172
  /health                   170
  /api/products             158

Top 5 URLs (plus lentes en moyenne) :
  /api/orders               1061 ms
  /api/checkout             1055 ms
  /api/users                1052 ms
  /api/products             1008 ms
  /login                    1001 ms
==================================================

Points à retenir

  • Counter du module collections : compte les occurrences et offre most_common(n).
  • defaultdict(list) pour accumuler des durées par URL sans if url not in.
  • @classmethod depuis_fichier : constructeur alternatif typé (note le forward reference "AnalyseurLogs").
  • Méthodes courtes et spécialisées : chacune fait une seule chose.

Partie 6 — Tests unitaires

Consignes

Écrivez des tests unitaires pour les composants principaux.

  1. Tester parser_ligne sur des cas valides et invalides.
  2. Tester les property de LogEntry (est_erreur, est_lent).
  3. Tester les méthodes de AnalyseurLogs avec des données fixées.
import unittest
from datetime import datetime
from dataclasses import dataclass
from typing import Literal
import re

NiveauLog = Literal["DEBUG", "INFO", "WARN", "ERROR"]


@dataclass
class LogEntry:
    horodatage: datetime
    niveau: NiveauLog
    methode: str
    url: str
    code: int
    duree_ms: int

    @property
    def est_erreur(self) -> bool:
        return self.code >= 400

    @property
    def est_lent(self) -> bool:
        return self.duree_ms > 1000


MOTIF_LOG = re.compile(
    r"(?P<date>\d{4}-\d{2}-\d{2})\s+"
    r"(?P<heure>\d{2}:\d{2}:\d{2})\s+"
    r"(?P<niveau>DEBUG|INFO|WARN|ERROR)\s+"
    r"(?P<methode>GET|POST|PUT|DELETE|PATCH)\s+"
    r"(?P<url>\S+)\s+"
    r"(?P<code>\d{3})\s+"
    r"(?P<duree>\d+)ms"
)


class LigneInvalideError(Exception): pass


def parser_ligne(ligne: str) -> LogEntry:
    ligne = ligne.strip()
    match = MOTIF_LOG.match(ligne)
    if match is None:
        raise LigneInvalideError(f"Ligne invalide")
    horodatage = datetime.strptime(
        f"{match['date']} {match['heure']}", "%Y-%m-%d %H:%M:%S"
    )
    return LogEntry(
        horodatage=horodatage,
        niveau=match["niveau"],     # type: ignore[arg-type]
        methode=match["methode"],
        url=match["url"],
        code=int(match["code"]),
        duree_ms=int(match["duree"]),
    )


class TestLogEntry(unittest.TestCase):
    def test_est_erreur_200(self):
        e = LogEntry(datetime.now(), "INFO", "GET", "/", 200, 10)
        self.assertFalse(e.est_erreur)

    def test_est_erreur_404(self):
        e = LogEntry(datetime.now(), "WARN", "GET", "/", 404, 10)
        self.assertTrue(e.est_erreur)

    def test_est_erreur_500(self):
        e = LogEntry(datetime.now(), "ERROR", "GET", "/", 500, 10)
        self.assertTrue(e.est_erreur)

    def test_est_lent(self):
        rapide = LogEntry(datetime.now(), "INFO", "GET", "/", 200, 100)
        lent   = LogEntry(datetime.now(), "INFO", "GET", "/", 200, 1500)
        self.assertFalse(rapide.est_lent)
        self.assertTrue(lent.est_lent)


class TestParserLigne(unittest.TestCase):
    def test_ligne_valide(self):
        e = parser_ligne("2026-04-22 14:30:12 INFO  GET /api/users 200 42ms")
        self.assertEqual(e.niveau, "INFO")
        self.assertEqual(e.methode, "GET")
        self.assertEqual(e.url, "/api/users")
        self.assertEqual(e.code, 200)
        self.assertEqual(e.duree_ms, 42)

    def test_erreur_500(self):
        e = parser_ligne("2026-04-22 14:30:18 ERROR GET /api/x 500 3200ms")
        self.assertTrue(e.est_erreur)
        self.assertTrue(e.est_lent)

    def test_ligne_vide_leve(self):
        with self.assertRaises(LigneInvalideError):
            parser_ligne("")

    def test_ligne_mal_formatee_leve(self):
        with self.assertRaises(LigneInvalideError):
            parser_ligne("pas un log")

    def test_niveau_inconnu_leve(self):
        with self.assertRaises(LigneInvalideError):
            parser_ligne("2026-04-22 14:30:12 UNKNOWN GET / 200 10ms")


suite = unittest.TestLoader().loadTestsFromTestCase(TestLogEntry)
result1 = unittest.TextTestRunner(verbosity=0).run(suite)

suite = unittest.TestLoader().loadTestsFromTestCase(TestParserLigne)
result2 = unittest.TextTestRunner(verbosity=0).run(suite)

total = result1.testsRun + result2.testsRun
echecs = len(result1.failures) + len(result2.failures) + len(result1.errors) + len(result2.errors)
print(f"\n{total} tests exécutés, {echecs} échec(s)")

✅ 9 tests exécutés, 0 échec(s)

Points à retenir

  • Tester les cas frontières : codes 200 (non-erreur), 400 (frontière), 500 (erreur).
  • Tester les exceptions avec assertRaises.
  • Des noms explicites : test_ligne_vide_leve décrit précisément le test.
  • Chaque test est indépendant — pas de state partagé.

Partie 7 — CLI avec sys.argv

Consignes

Écrivez un script cli.py qui :

  • Prend un chemin de log en argument.
  • Affiche le rapport complet.
  • Accepte un flag -v pour le mode verbeux (afficher les lignes ignorées).
  • Affiche l’usage et sort avec code 2 si mauvaise utilisation.
# cli.py
import sys
from pathlib import Path

# (importer ce qui a été développé dans les parties précédentes)


def main() -> int:
    if len(sys.argv) < 2:
        print("Usage : python cli.py <fichier.log> [-v]", file=sys.stderr)
        return 2

    chemin = Path(sys.argv[1])
    verbose = "-v" in sys.argv[2:]

    if not chemin.is_file():
        print(f"Fichier introuvable : {chemin}", file=sys.stderr)
        return 1

    try:
        analyseur = AnalyseurLogs.depuis_fichier(chemin)
    except Exception as e:
        print(f"Erreur à l'analyse : {e}", file=sys.stderr)
        return 1

    print(analyseur.rapport())
    return 0


if __name__ == "__main__":
    sys.exit(main())

Usage

$ python cli.py access.log
[rapport détaillé]

$ python cli.py access.log -v
[ligne 42] ignorée : Ligne invalide : 'corrompue'
[rapport détaillé]

$ python cli.py
Usage : python cli.py <fichier.log> [-v]
$ echo $?
2

Points à retenir

  • Code de sortie 0 (succès), 1 (erreur), 2 (mauvaise utilisation) — convention Unix.
  • Messages d’erreur sur stderr, rapport sur stdout — permet de rediriger séparément.
  • if __name__ == "__main__": pour que le script soit importable sans se lancer.

🏆 Bonus — Pour aller plus loin

  1. Export CSV : exporter_csv(analyseur, chemin) avec le module csv.

  2. Filtre temporel : méthode filtrer_periode(debut, fin) qui renvoie un nouveau AnalyseurLogs.

  3. Détection d’anomalies : méthode qui renvoie les URLs avec un taux d’erreur supérieur à la moyenne (sur 3 écarts-types).

  4. Mode streaming : ne pas charger toutes les entrées en mémoire — utiliser des générateurs pour les stats. Défi : comment faire top_urls sans tout stocker ? (Indice : un seul passage + Counter).

  5. Tests pytest avec paramétrisation et fixtures : réécrire les tests en pytest.

  6. Rapport HTML : générer un rapport au format HTML stylisé avec CSS.

  7. Graphiques : utiliser matplotlib pour visualiser l’évolution des erreurs dans le temps.

import statistics

def urls_anormales(analyseur, seuil_sigma: float = 3.0) -> list[str]:
    """Renvoie les URLs dont le taux d'erreur dépasse moyenne + sigma × écart-type."""
    from collections import defaultdict

    stats: dict[str, list[bool]] = defaultdict(list)
    for e in analyseur.entrees:
        stats[e.url].append(e.est_erreur)

    taux = {url: sum(v) / len(v) for url, v in stats.items()}

    if len(taux) < 2:
        return []

    moyenne = statistics.mean(taux.values())
    ecart = statistics.stdev(taux.values())
    seuil = moyenne + seuil_sigma * ecart

    return [url for url, t in taux.items() if t > seuil]

Pattern réel : alerte opérationnelle sur les endpoints qui dégradent.


✅ Validation du niveau Expert

Bilan du TP

Si vous avez résolu les 7 parties (même partiellement), vous maîtrisez les compétences du niveau Expert TOSA :

  • Regex avec groupes nommés et re.compile
  • Générateurs pour traiter des flux (pas de chargement mémoire)
  • Décorateurs typés et génériques
  • Closures (dans les décorateurs)
  • POO avancée avec @dataclass, properties, classmethods
  • Exceptions personnalisées avec chaînage (raise ... from)
  • pathlib pour les chemins
  • Type hints complets + Literal, Iterator, Iterable, forward references
  • Tests unitaires avec unittest
  • CLI avec sys.argv et codes de sortie
  • Qualité : code documenté, typé, testé

Score TOSA estimé : 876-1000 (niveau Expert — excellence maximale).

Félicitations ! Vous êtes prêt(e) pour l’Évaluation finale qui vous mènera à l’examen TOSA réel.

← Chapitre précédent : Type hintsPartie 5 — Évaluation finale →