#!/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()