Frise multi-projets : outil de planification macro
Une frise chronologique unique où plusieurs projets s'empilent en couloirs pliables, pour voir d'un coup d'œil où en est chacun et comment ils se situent les uns par rapport aux autres. Ce n'est pas un outil de suivi quotidien : une phase se compte en semaines, et il n'y a ni sous-tâches, ni tickets, ni dépendances entre phases. - Manipulation directe des barres (glisser, redimensionner), accrochées au lundi - Couloirs pliables : plié, un projet devient une barre segmentée par phase - Planning de référence figeable, avec barre fantôme et calcul de dérive - En-tête à trois bandes : année, mois, numéro de semaine ISO 8601 - Frise qui s'élargit au défilement, pour planifier dans un futur encore vide - Micro-serveur Python (bibliothèque standard) exposant GET/PUT sur /api/data, avec sauvegarde horodatée avant chaque écriture - Zéro build : modules ES natifs, aucune dépendance à installer 50 tests unitaires sur la logique métier (node --test, sans dépendance). Vérifié dans Chromium et Firefox. Les arbitrages de conception et surtout leurs raisons sont consignés dans docs/decisions.md — notamment l'abandon des dépendances entre phases, de la File System Access API, du SVG et des niveaux de zoom. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
230
serve.py
Normal file
230
serve.py
Normal file
@@ -0,0 +1,230 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user