gestion_projets/serve.py

#!/usr/bin/env python3
"""Micro-serveur pour la frise multi-projets.

Sert les fichiers statiques du dossier courant et expose le fichier de données :

    GET  /api/data   -> renvoie le JSON du planning
    PUT  /api/data   -> le remplace, apres copie horodatee dans data/backups/

Bibliotheque standard uniquement, aucune dependance a installer.

Ce n'est volontairement pas un back-end : il n'y a ici aucune logique metier, le
serveur ne fait que lire et ecrire un fichier. Voir docs/decisions.md, section 3.
"""

import argparse
import json
import shutil
import sys
from datetime import datetime, timezone
from functools import partial
from http import HTTPStatus
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

RACINE = Path(__file__).resolve().parent

# Au-dela, on refuse la requete : un planning macro pese quelques kilo-octets,
# un corps de cette taille signale forcement une erreur.
TAILLE_MAX = 5 * 1024 * 1024

# Nombre de sauvegardes conservees par fichier de donnees. Les plus anciennes
# sont supprimees pour que data/backups/ ne grossisse pas indefiniment.
BACKUPS_CONSERVES = 50

DONNEES_INITIALES = {"version": 2, "projects": []}


class Handler(SimpleHTTPRequestHandler):
    """Fichiers statiques, plus deux routes pour le fichier de donnees."""

    def __init__(self, *args, fichier_donnees: Path, **kwargs):
        self.fichier_donnees = fichier_donnees
        super().__init__(*args, directory=str(RACINE), **kwargs)

    # --- routes -----------------------------------------------------------

    def do_GET(self):
        if self.path.split("?")[0] == "/api/data":
            self._lire_donnees()
            return
        super().do_GET()

    def do_PUT(self):
        if self.path.split("?")[0] == "/api/data":
            self._ecrire_donnees()
            return
        self._erreur(HTTPStatus.NOT_FOUND, "Route inconnue.")

    # --- implementation des routes ----------------------------------------

    def _lire_donnees(self):
        if not self.fichier_donnees.exists():
            # Premier lancement : on repond un planning vide sans rien ecrire
            # sur le disque. Le fichier naitra a la premiere sauvegarde.
            self._json(HTTPStatus.OK, DONNEES_INITIALES)
            return

        try:
            contenu = self.fichier_donnees.read_text(encoding="utf-8")
            json.loads(contenu)  # on refuse de servir un fichier corrompu
        except json.JSONDecodeError as err:
            self._erreur(
                HTTPStatus.INTERNAL_SERVER_ERROR,
                f"{self.fichier_donnees.name} n'est pas un JSON valide : {err}. "
                "Le corriger a la main ou repartir d'une sauvegarde de data/backups/.",
            )
            return
        except OSError as err:
            self._erreur(HTTPStatus.INTERNAL_SERVER_ERROR, f"Lecture impossible : {err}")
            return

        self._envoyer(HTTPStatus.OK, contenu.encode("utf-8"), "application/json")

    def _ecrire_donnees(self):
        try:
            taille = int(self.headers.get("Content-Length", 0))
        except ValueError:
            self._erreur(HTTPStatus.BAD_REQUEST, "En-tete Content-Length invalide.")
            return

        if taille <= 0:
            self._erreur(HTTPStatus.BAD_REQUEST, "Corps de requete vide.")
            return
        if taille > TAILLE_MAX:
            self._erreur(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, "Corps de requete trop volumineux.")
            return

        corps = self.rfile.read(taille)

        # On valide avant d'ecrire : mieux vaut refuser une sauvegarde que
        # remplacer un planning correct par du JSON casse.
        try:
            donnees = json.loads(corps.decode("utf-8"))
        except (json.JSONDecodeError, UnicodeDecodeError) as err:
            self._erreur(HTTPStatus.BAD_REQUEST, f"JSON invalide : {err}")
            return

        if not isinstance(donnees, dict) or "projects" not in donnees:
            self._erreur(HTTPStatus.BAD_REQUEST, "Le JSON doit etre un objet contenant 'projects'.")
            return

        try:
            self._sauvegarder_version_precedente()
            self._ecrire_atomiquement(donnees)
        except OSError as err:
            self._erreur(HTTPStatus.INTERNAL_SERVER_ERROR, f"Ecriture impossible : {err}")
            return

        self._json(HTTPStatus.OK, {"ok": True, "savedAt": _horodatage_iso()})

    # --- ecriture sur disque ----------------------------------------------

    def _sauvegarder_version_precedente(self):
        """Copie le fichier actuel dans data/backups/ avant de l'ecraser."""
        if not self.fichier_donnees.exists():
            return

        dossier = self.fichier_donnees.parent / "backups"
        dossier.mkdir(parents=True, exist_ok=True)

        marqueur = datetime.now().strftime("%Y%m%d-%H%M%S")
        cible = dossier / f"{self.fichier_donnees.stem}-{marqueur}.json"

        # Deux sauvegardes dans la meme seconde : on suffixe pour ne pas
        # ecraser la precedente.
        compteur = 1
        while cible.exists():
            cible = dossier / f"{self.fichier_donnees.stem}-{marqueur}-{compteur}.json"
            compteur += 1

        shutil.copy2(self.fichier_donnees, cible)
        _elaguer_backups(dossier, self.fichier_donnees.stem)

    def _ecrire_atomiquement(self, donnees):
        """Ecrit dans un fichier temporaire puis remplace, pour qu'une coupure
        en cours d'ecriture ne laisse jamais un planning tronque."""
        self.fichier_donnees.parent.mkdir(parents=True, exist_ok=True)
        temporaire = self.fichier_donnees.with_suffix(".json.tmp")
        texte = json.dumps(donnees, indent=2, ensure_ascii=False) + "\n"
        temporaire.write_text(texte, encoding="utf-8")
        temporaire.replace(self.fichier_donnees)

    # --- helpers de reponse -----------------------------------------------

    def _json(self, statut, charge):
        self._envoyer(statut, json.dumps(charge, ensure_ascii=False).encode("utf-8"),
                      "application/json")

    def _erreur(self, statut, message):
        self._json(statut, {"error": message})

    def _envoyer(self, statut, corps, type_contenu):
        self.send_response(statut)
        self.send_header("Content-Type", f"{type_contenu}; charset=utf-8")
        self.send_header("Content-Length", str(len(corps)))
        self.end_headers()  # ajoute Cache-Control: no-store
        self.wfile.write(corps)

    def end_headers(self):
        # Aucun cache, sur aucune route, et emis ici seulement pour n'avoir
        # qu'un seul endroit responsable de cet en-tete.
        #
        # Filtrer sur l'extension ne suffirait pas : la page d'accueil est
        # servie sur « / », donc elle passerait au travers et Chromium
        # resservirait un index.html perime apres chaque modification.
        self.send_header("Cache-Control", "no-store")
        super().end_headers()

    def log_message(self, format, *args):
        # Une ligne par requete suffit ; on retire l'adresse, toujours locale.
        sys.stderr.write(f"  {format % args}\n")


def _elaguer_backups(dossier: Path, prefixe: str):
    fichiers = sorted(dossier.glob(f"{prefixe}-*.json"))
    for vieux in fichiers[:-BACKUPS_CONSERVES]:
        vieux.unlink(missing_ok=True)


def _horodatage_iso():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def main():
    analyseur = argparse.ArgumentParser(description="Serveur local de la frise multi-projets.")
    analyseur.add_argument("--port", type=int, default=8000, help="port d'ecoute (defaut : 8000)")
    analyseur.add_argument("--data", default="data/projets.json",
                           help="fichier de donnees (defaut : data/projets.json)")
    arguments = analyseur.parse_args()

    fichier = Path(arguments.data)
    if not fichier.is_absolute():
        fichier = RACINE / fichier
    fichier = fichier.resolve()

    if RACINE not in fichier.parents:
        analyseur.error("--data doit designer un fichier situe dans le dossier du projet.")

    handler = partial(Handler, fichier_donnees=fichier)

    try:
        serveur = ThreadingHTTPServer(("127.0.0.1", arguments.port), handler)
    except OSError as err:
        print(f"Impossible d'ecouter sur le port {arguments.port} : {err}", file=sys.stderr)
        print("Un autre serveur tourne peut-etre deja. Essayer --port 8001.", file=sys.stderr)
        raise SystemExit(1)

    print(f"Frise multi-projets  ->  http://localhost:{arguments.port}")
    print(f"Donnees              :  {fichier.relative_to(RACINE)}")
    print("Ctrl+C pour arreter.\n")

    try:
        serveur.serve_forever()
    except KeyboardInterrupt:
        print("\nArret.")
        serveur.shutdown()


if __name__ == "__main__":
    main()