Interface locale : rtg interface (voyages, sources, journées, récit, génération)
Serveur web local sans dépendance nouvelle (http.server, Jinja2, htmx et Leaflet fournis). Écrans : Mes voyages, nouveau voyage, un voyage (journées avec carte, rattachement des traces sans heures à un jour avec suggestion d'après les photos, sources avec choix de dossier, infos et récit, génération avec avancement et arrêt), réglages du site et vérification des outils. Chaîne : jours imposés ([dates_traces] dans voyage.toml), arrêt propre de la compression, suivi des étapes, état structuré (site.tableau_etat), titre du voyage repris de voyage.toml sur l'accueil. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Mf2kATMtqVABHSYdxJMnim
This commit is contained in:
1 parent
608334b9d8
commit
06153a227e
37 files changed
+2618
-57
No files matched your search
@@ -0,0 +1,670 @@
|
||||
"""Interface de RTG : un petit serveur web local, ouvert dans le navigateur.
|
||||
|
||||
Rien n'est installé en plus : le serveur vient de Python, les pages sont des
|
||||
modèles Jinja2, htmx (échanges avec le serveur) et Leaflet (cartes) sont
|
||||
fournis dans static/. Le serveur n'écoute que sur 127.0.0.1.
|
||||
|
||||
Sécurité : une page web quelconque ouverte dans le navigateur pourrait
|
||||
envoyer des requêtes à ce serveur. Toute requête qui modifie quelque chose
|
||||
doit donc porter l'en-tête HX-Request (qu'un autre site ne peut pas ajouter
|
||||
sans l'accord du serveur), et l'en-tête Host doit être le nôtre.
|
||||
"""
|
||||
import io
|
||||
import json
|
||||
import mimetypes
|
||||
import os
|
||||
import platform
|
||||
import re
|
||||
import shutil
|
||||
import string
|
||||
import threading
|
||||
import webbrowser
|
||||
from contextlib import redirect_stdout
|
||||
from datetime import date, timedelta
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from urllib.parse import parse_qs, quote, unquote, urlsplit
|
||||
|
||||
from jinja2 import Environment, FileSystemLoader, select_autoescape
|
||||
|
||||
from rtg import __version__, accueil, site, sources, video
|
||||
from rtg.config import (QUALITES, SITE_TOML, VOYAGE_TOML, modifier_toml, parametres_video,
|
||||
reglages_site, reglages_voyage)
|
||||
from rtg.interface import taches
|
||||
from rtg.outils import EXT_PHOTOS, EXT_VIDEOS, est_cache, lister_fichiers, taille_lisible
|
||||
|
||||
ICI = os.path.dirname(os.path.abspath(__file__))
|
||||
STATIC = os.path.join(ICI, "static")
|
||||
MOIS = ["janvier", "février", "mars", "avril", "mai", "juin", "juillet", "août",
|
||||
"septembre", "octobre", "novembre", "décembre"]
|
||||
JOURS = ["lundi", "mardi", "mercredi", "jeudi", "vendredi", "samedi", "dimanche"]
|
||||
COULEURS = ["#e6194b", "#3cb44b", "#4363d8", "#f58231", "#911eb4", "#46a0b4", "#f032e6",
|
||||
"#7a9a01", "#c47f00", "#008080", "#9a6324", "#800000", "#2f4b7c", "#808000"]
|
||||
CLES_SOURCES = {"traces": ("Traces GPS", (".gpx",), "traces GPS"),
|
||||
"videos": ("Vidéos", EXT_VIDEOS, "vidéos"),
|
||||
"photos": ("Photos", EXT_PHOTOS, "photos")}
|
||||
|
||||
|
||||
def _date_longue(d):
|
||||
return f"{JOURS[d.weekday()]} {d.day} {MOIS[d.month - 1]} {d.year}" if d else ""
|
||||
|
||||
|
||||
def _periode(debut, fin):
|
||||
if not debut:
|
||||
return ""
|
||||
if debut == fin:
|
||||
return f"{debut.day} {MOIS[debut.month - 1]} {debut.year}"
|
||||
if debut.year == fin.year:
|
||||
return f"{debut.day} {MOIS[debut.month - 1]} – {fin.day} {MOIS[fin.month - 1]} {fin.year}"
|
||||
return f"{debut.day} {MOIS[debut.month - 1]} {debut.year} – {fin.day} {MOIS[fin.month - 1]} {fin.year}"
|
||||
|
||||
|
||||
ENV = Environment(loader=FileSystemLoader(os.path.join(ICI, "modeles")),
|
||||
autoescape=select_autoescape(["html"]), trim_blocks=True, lstrip_blocks=True)
|
||||
ENV.filters["date_longue"] = _date_longue
|
||||
ENV.filters["date_courte"] = lambda d: f"{d.day} {MOIS[d.month - 1][:4].rstrip('.')}" if d else ""
|
||||
ENV.filters["taille"] = taille_lisible
|
||||
ENV.filters["url"] = lambda s: quote(str(s), safe="")
|
||||
ENV.tests["commence_par"] = lambda s, debut: str(s).startswith(debut)
|
||||
ENV.globals.update(version=__version__, periode=_periode, qualites=QUALITES,
|
||||
noms_qualites={"econome": "économe", "equilibree": "équilibrée", "haute": "haute"})
|
||||
|
||||
|
||||
class Erreur(Exception):
|
||||
def __init__(self, code, message):
|
||||
super().__init__(message)
|
||||
self.code, self.message = code, message
|
||||
|
||||
|
||||
class Reponse:
|
||||
def __init__(self, corps=b"", type_="text/html; charset=utf-8", code=200, entetes=None):
|
||||
self.corps = corps.encode("utf-8") if isinstance(corps, str) else corps
|
||||
self.type, self.code, self.entetes = type_, code, entetes or {}
|
||||
|
||||
|
||||
def page(modele, **ctx):
|
||||
return Reponse(ENV.get_template(modele).render(**ctx))
|
||||
|
||||
|
||||
def json_rep(donnees):
|
||||
return Reponse(json.dumps(donnees, ensure_ascii=False, default=str), "application/json")
|
||||
|
||||
|
||||
# --- Application ---------------------------------------------------------------
|
||||
class Application:
|
||||
def __init__(self, racine):
|
||||
self.racine = os.path.abspath(racine)
|
||||
self._cache = {} # voyage → (empreinte des sources, analyse)
|
||||
self.routes = []
|
||||
for methode, motif, fn in [
|
||||
("GET", r"/", self.accueil),
|
||||
("POST", r"/site/creer", self.creer_site),
|
||||
("GET", r"/nouveau", self.nouveau),
|
||||
("POST", r"/nouveau", self.nouveau_creer),
|
||||
("GET", r"/voyage", self.voyage),
|
||||
("POST", r"/voyage/infos", self.voyage_infos),
|
||||
("POST", r"/voyage/recit", self.voyage_recit),
|
||||
("POST", r"/voyage/sources", self.voyage_sources),
|
||||
("GET", r"/voyage/analyse", self.voyage_analyse),
|
||||
("POST", r"/voyage/date", self.voyage_date),
|
||||
("POST", r"/voyage/dates-suite", self.voyage_dates_suite),
|
||||
("GET", r"/voyage/infos-utiles", self.voyage_infos_utiles),
|
||||
("GET", r"/dossiers", self.dossiers),
|
||||
("GET", r"/reglages", self.reglages),
|
||||
("POST", r"/reglages", self.reglages_enregistrer),
|
||||
("GET", r"/systeme", self.systeme),
|
||||
("POST", r"/tache/lancer", self.tache_lancer),
|
||||
("POST", r"/tache/arreter", self.tache_arreter),
|
||||
("GET", r"/tache", self.tache),
|
||||
("GET", r"/tache/fragment", self.tache_fragment),
|
||||
("GET", r"/tache/bandeau", self.tache_bandeau),
|
||||
]:
|
||||
self.routes.append((methode, re.compile(motif + "$"), fn))
|
||||
|
||||
# --- outils
|
||||
def contexte(self, **ctx):
|
||||
ctx.setdefault("racine", self.racine)
|
||||
ctx.setdefault("site_existe", os.path.exists(os.path.join(self.racine, SITE_TOML)))
|
||||
ctx.setdefault("tache", taches.COURANTE)
|
||||
return ctx
|
||||
|
||||
def chemin_voyage(self, rel):
|
||||
if not rel:
|
||||
raise Erreur(400, "Voyage non précisé")
|
||||
chemin = os.path.realpath(os.path.join(self.racine, rel))
|
||||
if not chemin.startswith(os.path.realpath(self.racine) + os.sep) or not site.est_voyage(chemin):
|
||||
raise Erreur(404, f"Voyage introuvable : {rel}")
|
||||
return chemin
|
||||
|
||||
def cfg_voyage(self, chemin):
|
||||
return reglages_voyage(chemin, reglages_site(self.racine))
|
||||
|
||||
def fiche_voyage(self, chemin):
|
||||
return site.etat_voyage(chemin, self.racine, reglages_site(self.racine))
|
||||
|
||||
def toml_voyage(self, chemin, changements):
|
||||
modifier_toml(os.path.join(chemin, VOYAGE_TOML), changements,
|
||||
entete="Voyage RoadTripsGenerator : sources, infos et réglages propres.")
|
||||
|
||||
# --- Mes voyages
|
||||
def accueil(self, r):
|
||||
voyages = site.tableau_etat(self.racine)
|
||||
groupes = {}
|
||||
for v in sorted(voyages, key=lambda v: v["debut"] or date.min, reverse=True):
|
||||
groupes.setdefault(v["rubrique"], []).append(v)
|
||||
ordre = sorted(groupes, key=lambda g: (g != "Voyages", g))
|
||||
return page("voyages.html", **self.contexte(groupes=[(g, groupes[g]) for g in ordre],
|
||||
nb=len(voyages)))
|
||||
|
||||
def creer_site(self, r):
|
||||
chemin = os.path.join(self.racine, SITE_TOML)
|
||||
if not os.path.exists(chemin):
|
||||
modifier_toml(chemin, {"titre": "Nos Road-Trips"},
|
||||
entete="Réglages du site RoadTripsGenerator (rtg).")
|
||||
return Reponse("", entetes={"HX-Redirect": "/"})
|
||||
|
||||
def groupes_existants(self):
|
||||
res = []
|
||||
for d in sorted(os.listdir(self.racine)):
|
||||
c = os.path.join(self.racine, d)
|
||||
if os.path.isdir(c) and not d.startswith((".", "_")) and d not in site.IGNORES \
|
||||
and not site.est_voyage(c):
|
||||
res.append(d)
|
||||
return res
|
||||
|
||||
def nouveau(self, r):
|
||||
return page("nouveau.html", **self.contexte(groupes=self.groupes_existants()))
|
||||
|
||||
def nouveau_creer(self, r):
|
||||
dossier = r.f("dossier").strip()
|
||||
groupe = r.f("groupe_nouveau").strip() or r.f("groupe").strip()
|
||||
if not dossier:
|
||||
return page("nouveau.html", **self.contexte(groupes=self.groupes_existants(),
|
||||
erreur="Indiquez un nom de dossier.", form=r.form))
|
||||
try:
|
||||
chemin = site.creer_voyage(self.racine, dossier, r.f("titre").strip(), groupe, r.f("moto").strip())
|
||||
except (ValueError, OSError) as e:
|
||||
return page("nouveau.html", **self.contexte(groupes=self.groupes_existants(),
|
||||
erreur=str(e), form=r.form))
|
||||
rel = os.path.relpath(chemin, self.racine).replace(os.sep, "/")
|
||||
return Reponse("", entetes={"HX-Redirect": f"/voyage?v={quote(rel)}#sources"})
|
||||
|
||||
# --- Un voyage
|
||||
def voyage(self, r):
|
||||
chemin = self.chemin_voyage(r.q("v"))
|
||||
cfg = self.cfg_voyage(chemin)
|
||||
recit_f = os.path.join(chemin, "voyage.md")
|
||||
recit = open(recit_f, encoding="utf-8").read() if os.path.exists(recit_f) else ""
|
||||
return page("voyage.html", **self.contexte(
|
||||
v=self.fiche_voyage(chemin), cfg=cfg, recit=recit,
|
||||
sources=self.etat_sources(chemin, cfg)))
|
||||
|
||||
def voyage_infos(self, r):
|
||||
chemin = self.chemin_voyage(r.f("v"))
|
||||
self.toml_voyage(chemin, {"titre": r.f("titre").strip() or None, "moto": r.f("moto").strip() or None})
|
||||
return Reponse('<span class="ok">✓ Enregistré. Les pages seront à jour à la prochaine génération.</span>')
|
||||
|
||||
def voyage_recit(self, r):
|
||||
chemin = self.chemin_voyage(r.f("v"))
|
||||
texte = r.f("recit").replace("\r\n", "\n")
|
||||
f = os.path.join(chemin, "voyage.md")
|
||||
if texte.strip() or os.path.exists(f):
|
||||
with open(f, "w", encoding="utf-8") as fh:
|
||||
fh.write(texte)
|
||||
return Reponse('<span class="ok">✓ Récit enregistré.</span>')
|
||||
|
||||
def etat_sources(self, chemin, cfg):
|
||||
res = {}
|
||||
for cle, (libelle, ext, _) in CLES_SOURCES.items():
|
||||
lignes = []
|
||||
for s in cfg["sources"].get(cle, []):
|
||||
p = os.path.join(chemin, os.path.expanduser(s))
|
||||
fichiers = lister_fichiers([s], ext, chemin) if os.path.exists(p) else []
|
||||
lignes.append({"source": s, "absolu": os.path.abspath(p), "existe": os.path.exists(p),
|
||||
"lien_casse": os.path.islink(p) and not os.path.exists(p),
|
||||
"nb": len(fichiers)})
|
||||
res[cle] = {"libelle": libelle, "lignes": lignes}
|
||||
return res
|
||||
|
||||
def voyage_sources(self, r):
|
||||
chemin = self.chemin_voyage(r.f("v"))
|
||||
cle = r.f("cle")
|
||||
if cle not in CLES_SOURCES:
|
||||
raise Erreur(400, "Type de source inconnu")
|
||||
cfg = self.cfg_voyage(chemin)
|
||||
liste = list(cfg["sources"].get(cle, []))
|
||||
source = r.f("chemin").strip()
|
||||
if r.f("action") == "ajouter" and source:
|
||||
source = os.path.abspath(os.path.expanduser(source))
|
||||
if not os.path.exists(source):
|
||||
raise Erreur(400, f"Introuvable : {source}")
|
||||
# Dans le dossier du voyage : chemin relatif (le site reste déplaçable).
|
||||
if os.path.realpath(source).startswith(os.path.realpath(chemin) + os.sep):
|
||||
source = os.path.relpath(os.path.realpath(source), os.path.realpath(chemin))
|
||||
if source not in liste:
|
||||
liste.append(source)
|
||||
elif r.f("action") == "retirer":
|
||||
liste = [s for s in liste if s != source]
|
||||
self.toml_voyage(chemin, {"sources": {cle: liste}})
|
||||
cfg = self.cfg_voyage(chemin)
|
||||
rel = r.f("v")
|
||||
return Reponse(ENV.get_template("_sources.html").render(
|
||||
v={"chemin": rel}, sources=self.etat_sources(chemin, cfg)),
|
||||
entetes={"HX-Trigger": "sources-modifiees"})
|
||||
|
||||
def _empreinte(self, chemin, cfg):
|
||||
"""Ce dont dépend l'analyse : si rien n'a changé, on la réutilise."""
|
||||
def etat(f):
|
||||
try:
|
||||
st = os.stat(f)
|
||||
return (f, st.st_mtime, st.st_size)
|
||||
except OSError:
|
||||
return (f, None, None)
|
||||
fichiers = lister_fichiers(cfg["sources"].get("traces", []), (".gpx",), chemin)
|
||||
autres = [os.path.join(chemin, n) for n in (VOYAGE_TOML, "export_photos.csv", "export_videos.csv")]
|
||||
autres.append(os.path.join(self.racine, SITE_TOML))
|
||||
photos = cfg["sources"].get("photos", [])
|
||||
return tuple(etat(f) for f in fichiers + autres) + (tuple(photos),)
|
||||
|
||||
def analyse(self, chemin):
|
||||
cfg = self.cfg_voyage(chemin)
|
||||
cle = self._empreinte(chemin, cfg)
|
||||
if self._cache.get(chemin, (None,))[0] == cle:
|
||||
return self._cache[chemin][1]
|
||||
res = self._analyse(chemin, cfg)
|
||||
self._cache[chemin] = (cle, res)
|
||||
return res
|
||||
|
||||
def _analyse(self, chemin, cfg):
|
||||
a = sources.analyser(chemin, cfg)
|
||||
jours = []
|
||||
for i, j in enumerate(sorted(a.journees)):
|
||||
noms = sorted({d["nom"] for d in a.details_traces if j in d["jours"]})
|
||||
jours.append({"date": j, "km": a.km(j), "traces": noms, "couleur": COULEURS[i % len(COULEURS)]})
|
||||
suggestions = {}
|
||||
if a.non_datees:
|
||||
medias = sources.positions_medias(chemin, a)
|
||||
for f, pistes in a.non_datees.items():
|
||||
suggestions[os.path.basename(f)] = sources.suggerer_jour(pistes, medias)
|
||||
return cfg, a, jours, suggestions
|
||||
|
||||
def voyage_analyse(self, r):
|
||||
chemin = self.chemin_voyage(r.q("v"))
|
||||
cfg, a, jours, suggestions = self.analyse(chemin)
|
||||
total = sum(j["km"] for j in jours)
|
||||
return page("_analyse.html", v={"chemin": r.q("v")}, a=a, jours=jours, total_km=total,
|
||||
suggestions=suggestions, cfg=cfg, couches=_couches(a, jours),
|
||||
non_datees=sorted(os.path.basename(f) for f in a.non_datees))
|
||||
|
||||
def voyage_date(self, r):
|
||||
chemin = self.chemin_voyage(r.f("v"))
|
||||
nom, jour = r.f("fichier"), r.f("jour").strip()
|
||||
if jour and not sources.jour_texte(jour):
|
||||
raise Erreur(400, f"Date invalide : {jour}")
|
||||
self.toml_voyage(chemin, {"dates_traces": {nom: jour or None}})
|
||||
return self._rafraichir_analyse(r.f("v"))
|
||||
|
||||
def voyage_dates_suite(self, r):
|
||||
chemin = self.chemin_voyage(r.f("v"))
|
||||
debut = sources.jour_texte(r.f("debut"))
|
||||
if not debut:
|
||||
raise Erreur(400, "Indiquez le premier jour")
|
||||
_, a, _, _ = self.analyse(chemin)
|
||||
# Toutes les traces qui ne se datent pas seules, même déjà rattachées.
|
||||
noms = sorted(d["nom"] for d in a.details_traces if not d["erreur"] and not d["datable"])
|
||||
self.toml_voyage(chemin, {"dates_traces": {n: (debut + timedelta(days=i)).isoformat()
|
||||
for i, n in enumerate(noms)}})
|
||||
return self._rafraichir_analyse(r.f("v"))
|
||||
|
||||
def _rafraichir_analyse(self, rel):
|
||||
return self.voyage_analyse(Requete({"v": rel}, {}))
|
||||
|
||||
def voyage_infos_utiles(self, r):
|
||||
chemin = self.chemin_voyage(r.q("v"))
|
||||
cfg = self.cfg_voyage(chemin)
|
||||
a = sources.analyser(chemin, cfg, leger=True)
|
||||
|
||||
def taille_dossier(d):
|
||||
d = os.path.join(chemin, d)
|
||||
if not os.path.isdir(d):
|
||||
return 0
|
||||
return sum(e.stat().st_size for e in os.scandir(d) if e.is_file() and not est_cache(e.name))
|
||||
rushs = sum(os.path.getsize(f) for f in a.videos if os.path.exists(f))
|
||||
infos = {
|
||||
"rushs": rushs, "nb_rushs": len(a.videos),
|
||||
"videos_web": taille_dossier("videos"), "photos": taille_dossier("photos"),
|
||||
"photos_web": taille_dossier("photos_web"), "apercus": taille_dossier("apercus"),
|
||||
"libre": shutil.disk_usage(chemin).free,
|
||||
"encodage": parametres_video(cfg),
|
||||
}
|
||||
reste = [f for f in a.videos if not video.deja_faite(chemin, f, rapide=True)]
|
||||
infos["a_encoder"] = len(reste)
|
||||
infos["a_encoder_octets"] = sum(os.path.getsize(f) for f in reste if os.path.exists(f))
|
||||
return page("_infos_utiles.html", i=infos, cfg=cfg)
|
||||
|
||||
# --- Navigateur de dossiers
|
||||
def dossiers(self, r):
|
||||
chemin = r.q("chemin") or os.path.expanduser("~")
|
||||
chemin = os.path.abspath(os.path.expanduser(chemin))
|
||||
if not os.path.isdir(chemin):
|
||||
chemin = os.path.dirname(chemin)
|
||||
cle = r.q("cle")
|
||||
ext = CLES_SOURCES.get(cle, ("", (), ""))[1]
|
||||
sous, nb = [], 0
|
||||
try:
|
||||
for e in sorted(os.scandir(chemin), key=lambda e: e.name.lower()):
|
||||
if est_cache(e.name):
|
||||
continue
|
||||
if e.is_dir():
|
||||
sous.append(e.name)
|
||||
elif ext and e.name.lower().endswith(ext):
|
||||
nb += 1
|
||||
except PermissionError:
|
||||
pass
|
||||
raccourcis = [("Dossier personnel", os.path.expanduser("~")), ("Site", self.racine)]
|
||||
if r.q("v"):
|
||||
raccourcis.append(("Ce voyage", os.path.join(self.racine, r.q("v"))))
|
||||
if platform.system() == "Darwin" and os.path.isdir("/Volumes"):
|
||||
raccourcis += [(d, os.path.join("/Volumes", d)) for d in sorted(os.listdir("/Volumes"))
|
||||
if not d.startswith(".")]
|
||||
elif platform.system() == "Windows":
|
||||
raccourcis += [(f"{l}:", f"{l}:\\") for l in string.ascii_uppercase if os.path.exists(f"{l}:\\")]
|
||||
else:
|
||||
for base in ("/media", "/mnt", f"/media/{os.environ.get('USER', '')}"):
|
||||
if os.path.isdir(base):
|
||||
raccourcis += [(d, os.path.join(base, d)) for d in sorted(os.listdir(base))
|
||||
if os.path.isdir(os.path.join(base, d)) and not d.startswith(".")]
|
||||
parent = os.path.dirname(chemin)
|
||||
return page("_dossiers.html", chemin=chemin, parent=parent if parent != chemin else None,
|
||||
sous=sous, nb=nb, cle=cle, libelle=CLES_SOURCES.get(cle, ("", (), "fichiers"))[2],
|
||||
v=r.q("v"), raccourcis=raccourcis, sep=os.sep)
|
||||
|
||||
# --- Réglages du site
|
||||
def reglages(self, r):
|
||||
return page("reglages.html", **self.contexte(cfg=reglages_site(self.racine)))
|
||||
|
||||
def reglages_enregistrer(self, r):
|
||||
def entier(k, defaut):
|
||||
try:
|
||||
return int(r.f(k))
|
||||
except ValueError:
|
||||
return defaut
|
||||
cfg = reglages_site(self.racine)
|
||||
liens = [l.strip() for l in r.f("mymaps").splitlines() if l.strip()]
|
||||
modifier_toml(os.path.join(self.racine, SITE_TOML), {
|
||||
"titre": r.f("titre").strip() or cfg["titre"],
|
||||
"adresse_publique": r.f("adresse_publique").strip(),
|
||||
"accueil_nb_recents": entier("accueil_nb_recents", 2),
|
||||
"video": {"qualite": r.f("qualite") if r.f("qualite") in QUALITES else "equilibree",
|
||||
"encodeur": "materiel" if r.f("encodeur") == "materiel" else "processeur",
|
||||
"preset": r.f("preset") or "medium", "jobs": entier("jobs", 0)},
|
||||
"photos": {"grand_cote": entier("grand_cote", 1600), "qualite": entier("qualite_photos", 80),
|
||||
"positions_deduites": r.f("positions_deduites") == "on"},
|
||||
"journees": {"fuseau": r.f("fuseau").strip() or "Europe/Paris",
|
||||
"coupure_heure": entier("coupure_heure", 4)},
|
||||
"carte_globale": {"mymaps": liens},
|
||||
}, entete="Réglages du site RoadTripsGenerator (rtg).")
|
||||
return Reponse('<span class="ok">✓ Réglages enregistrés.</span>')
|
||||
|
||||
def systeme(self, r):
|
||||
from rtg.verifier import verifier
|
||||
tampon = io.StringIO()
|
||||
with redirect_stdout(tampon):
|
||||
ok = verifier()
|
||||
libre = shutil.disk_usage(self.racine)
|
||||
return page("_systeme.html", lignes=tampon.getvalue().splitlines(), ok=ok,
|
||||
libre=libre.free, total=libre.total, racine=self.racine)
|
||||
|
||||
# --- Traitements
|
||||
def tache_lancer(self, r):
|
||||
quoi, rel = r.f("quoi"), r.f("v")
|
||||
compression = r.f("sans_compression") != "on"
|
||||
force = r.f("force") == "on"
|
||||
racine = self.racine
|
||||
if quoi in ("generer", "compresser"):
|
||||
chemin = self.chemin_voyage(rel)
|
||||
if quoi == "generer":
|
||||
libelle = f"Génération de {rel}" + ("" if compression else " (sans les vidéos)")
|
||||
|
||||
def fn(suivi):
|
||||
return site.generer(racine, [chemin], compression=compression, force=force, suivi=suivi)
|
||||
elif quoi == "compresser":
|
||||
libelle = f"Compression des vidéos de {rel}"
|
||||
|
||||
def fn(suivi):
|
||||
cfg = reglages_voyage(chemin, reglages_site(racine))
|
||||
suivi({"etape": "Vidéos", "voyage": rel})
|
||||
return video.compresser(chemin, cfg, sources.analyser(chemin, cfg), force=force,
|
||||
progression=lambda n, e, p, rg: suivi(
|
||||
{"video": n, "etat": e, "pct": p, "rang": rg}))
|
||||
elif quoi == "tout":
|
||||
libelle = "Génération de tout le site" + ("" if compression else " (sans les vidéos)")
|
||||
|
||||
def fn(suivi):
|
||||
return site.generer(racine, None, compression=compression, force=force, suivi=suivi)
|
||||
elif quoi == "accueil":
|
||||
libelle = "Page d'accueil et carte globale"
|
||||
|
||||
def fn(suivi):
|
||||
suivi({"etape": "Accueil"})
|
||||
accueil.generer(racine, reglages_site(racine), r.f("mymaps_maj") == "on")
|
||||
else:
|
||||
raise Erreur(400, "Traitement inconnu")
|
||||
t = taches.lancer(libelle, fn, rel if quoi in ("generer", "compresser") else "")
|
||||
message = None if t else "Un traitement est déjà en cours : attendez qu'il se termine ou arrêtez-le."
|
||||
return page("_tache.html", tache=taches.COURANTE, message=message)
|
||||
|
||||
def tache_arreter(self, r):
|
||||
taches.arreter()
|
||||
return page("_tache.html", tache=taches.COURANTE)
|
||||
|
||||
def tache(self, r):
|
||||
return page("tache.html", **self.contexte())
|
||||
|
||||
def tache_fragment(self, r):
|
||||
t = taches.COURANTE
|
||||
rep = page("_tache.html", tache=t, complet=r.q("complet") == "1")
|
||||
if t and not t.en_cours:
|
||||
# Dernière relève : les autres zones de la page se mettent à jour.
|
||||
rep.entetes["HX-Trigger"] = "tache-finie"
|
||||
return rep
|
||||
|
||||
def tache_bandeau(self, r):
|
||||
return page("_bandeau.html", tache=taches.COURANTE)
|
||||
|
||||
|
||||
# --- HTTP ----------------------------------------------------------------------
|
||||
def _couches(a, jours, maxi=1500):
|
||||
"""Traces de chaque journée pour la carte (au plus `maxi` points par jour)."""
|
||||
def coords(pistes):
|
||||
segments = [s.points for t in pistes for s in t.segments if len(s.points) > 1]
|
||||
pas = max(1, sum(len(s) for s in segments) // maxi)
|
||||
res = []
|
||||
for pts in segments:
|
||||
seg = [[round(p.latitude, 5), round(p.longitude, 5)] for p in pts[::pas]]
|
||||
if (len(pts) - 1) % pas:
|
||||
seg.append([round(pts[-1].latitude, 5), round(pts[-1].longitude, 5)])
|
||||
res.append(seg)
|
||||
return res
|
||||
couches = [{"nom": _date_longue(j["date"]), "km": round(j["km"]), "couleur": j["couleur"],
|
||||
"segments": coords(a.journees[j["date"]])} for j in jours]
|
||||
for f, pistes in a.non_datees.items():
|
||||
couches.append({"nom": f"{os.path.basename(f)} (sans jour)", "couleur": "#555",
|
||||
"pointille": True, "km": 0, "segments": coords(pistes)})
|
||||
return couches
|
||||
|
||||
|
||||
def _valeur(d, k, defaut):
|
||||
v = d.get(k, defaut)
|
||||
if isinstance(v, list):
|
||||
return v[0] if v else defaut
|
||||
return v
|
||||
|
||||
|
||||
class Requete:
|
||||
def __init__(self, query, form):
|
||||
self.query, self.form = query, form
|
||||
|
||||
def q(self, k, defaut=""):
|
||||
return _valeur(self.query, k, defaut)
|
||||
|
||||
def f(self, k, defaut=""):
|
||||
return _valeur(self.form, k, defaut)
|
||||
|
||||
|
||||
def creer_gestionnaire(app, port):
|
||||
hotes = {f"127.0.0.1:{port}", f"localhost:{port}"}
|
||||
|
||||
class Gestionnaire(BaseHTTPRequestHandler):
|
||||
server_version = f"rtg/{__version__}"
|
||||
|
||||
def log_message(self, fmt, *args):
|
||||
pass # le terminal sert au journal des traitements
|
||||
|
||||
def do_GET(self):
|
||||
self.traiter("GET")
|
||||
|
||||
def do_HEAD(self):
|
||||
self.traiter("GET", tete=True)
|
||||
|
||||
def do_POST(self):
|
||||
self.traiter("POST")
|
||||
|
||||
def traiter(self, methode, tete=False):
|
||||
try:
|
||||
if self.headers.get("Host") not in hotes:
|
||||
raise Erreur(403, "Hôte refusé")
|
||||
url = urlsplit(self.path)
|
||||
chemin = unquote(url.path)
|
||||
if methode == "GET" and chemin.startswith("/static/"):
|
||||
return self.fichier(STATIC, chemin[len("/static/"):], tete)
|
||||
if methode == "GET" and chemin.startswith("/site/"):
|
||||
return self.fichier(app.racine, chemin[len("/site/"):], tete)
|
||||
form = {}
|
||||
if methode == "POST":
|
||||
if self.headers.get("HX-Request") != "true":
|
||||
raise Erreur(403, "Requête refusée")
|
||||
n = int(self.headers.get("Content-Length") or 0)
|
||||
form = parse_qs(self.rfile.read(n).decode("utf-8"), keep_blank_values=True)
|
||||
req = Requete(parse_qs(url.query, keep_blank_values=True), form)
|
||||
for m, motif, fn in app.routes:
|
||||
if m == methode and motif.match(chemin):
|
||||
rep = fn(req)
|
||||
break
|
||||
else:
|
||||
raise Erreur(404, "Page introuvable")
|
||||
except Erreur as e:
|
||||
rep = Reponse(f'<div class="erreur">{_echapper(e.message)}</div>', code=e.code)
|
||||
if self.headers.get("HX-Request") == "true":
|
||||
# htmx n'affiche pas les réponses en erreur : on renvoie 200
|
||||
# vers la zone de messages.
|
||||
rep.code = 200
|
||||
rep.entetes = {"HX-Retarget": "#messages", "HX-Reswap": "innerHTML"}
|
||||
except Exception as e:
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
rep = Reponse(f'<div class="erreur">Erreur : {_echapper(str(e))}</div>', code=500)
|
||||
if self.headers.get("HX-Request") == "true":
|
||||
rep.code = 200
|
||||
rep.entetes = {"HX-Retarget": "#messages", "HX-Reswap": "innerHTML"}
|
||||
self.envoyer(rep, tete)
|
||||
|
||||
def envoyer(self, rep, tete=False):
|
||||
self.send_response(rep.code)
|
||||
self.send_header("Content-Type", rep.type)
|
||||
self.send_header("Content-Length", str(len(rep.corps)))
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
for k, v in rep.entetes.items():
|
||||
self.send_header(k, v)
|
||||
self.end_headers()
|
||||
if not tete:
|
||||
self.wfile.write(rep.corps)
|
||||
|
||||
def fichier(self, base, rel, tete):
|
||||
"""Fichier statique, avec les requêtes partielles (Range) dont
|
||||
Safari a besoin pour lire les vidéos."""
|
||||
base = os.path.realpath(base)
|
||||
chemin = os.path.realpath(os.path.join(base, rel))
|
||||
if chemin != base and not chemin.startswith(base + os.sep):
|
||||
raise Erreur(403, "Chemin refusé")
|
||||
if os.path.isdir(chemin):
|
||||
if not self.path.split("?")[0].endswith("/"):
|
||||
self.send_response(301)
|
||||
self.send_header("Location", self.path.split("?")[0] + "/")
|
||||
self.end_headers()
|
||||
return
|
||||
chemin = os.path.join(chemin, "index.html")
|
||||
if not os.path.isfile(chemin):
|
||||
return self.envoyer(Reponse("Fichier introuvable", "text/plain; charset=utf-8", 404), tete)
|
||||
taille = os.path.getsize(chemin)
|
||||
type_ = mimetypes.guess_type(chemin)[0] or "application/octet-stream"
|
||||
if type_.startswith("text/") or type_ in ("application/javascript", "application/json"):
|
||||
type_ += "; charset=utf-8"
|
||||
debut, fin, code = 0, taille - 1, 200
|
||||
m = re.match(r"bytes=(\d*)-(\d*)$", self.headers.get("Range", ""))
|
||||
if m and taille:
|
||||
if m.group(1):
|
||||
debut = int(m.group(1))
|
||||
fin = min(int(m.group(2)), taille - 1) if m.group(2) else taille - 1
|
||||
else:
|
||||
debut = max(0, taille - int(m.group(2) or 0))
|
||||
if debut > fin:
|
||||
self.send_response(416)
|
||||
self.send_header("Content-Range", f"bytes */{taille}")
|
||||
self.end_headers()
|
||||
return
|
||||
code = 206
|
||||
self.send_response(code)
|
||||
self.send_header("Content-Type", type_)
|
||||
self.send_header("Accept-Ranges", "bytes")
|
||||
self.send_header("Content-Length", str(fin - debut + 1 if taille else 0))
|
||||
if code == 206:
|
||||
self.send_header("Content-Range", f"bytes {debut}-{fin}/{taille}")
|
||||
self.end_headers()
|
||||
if tete or not taille:
|
||||
return
|
||||
try:
|
||||
with open(chemin, "rb") as f:
|
||||
f.seek(debut)
|
||||
reste = fin - debut + 1
|
||||
while reste > 0:
|
||||
bloc = f.read(min(1 << 20, reste))
|
||||
if not bloc:
|
||||
break
|
||||
self.wfile.write(bloc)
|
||||
reste -= len(bloc)
|
||||
except (BrokenPipeError, ConnectionResetError):
|
||||
pass # le navigateur a interrompu (vidéo)
|
||||
|
||||
return Gestionnaire
|
||||
|
||||
|
||||
def _echapper(s):
|
||||
return str(s).replace("&", "&").replace("<", "<").replace(">", ">")
|
||||
|
||||
|
||||
def lancer(racine, port=8765, ouvrir=True):
|
||||
app = Application(racine)
|
||||
serveur = None
|
||||
for p in range(port, port + 20):
|
||||
try:
|
||||
serveur = ThreadingHTTPServer(("127.0.0.1", p), creer_gestionnaire(app, p))
|
||||
port = p
|
||||
break
|
||||
except OSError:
|
||||
continue
|
||||
if serveur is None:
|
||||
raise SystemExit(f"❌ Aucun port libre entre {port} et {port + 19}.")
|
||||
serveur.daemon_threads = True
|
||||
url = f"http://127.0.0.1:{port}/"
|
||||
print(f"🌐 Interface RTG : {url}")
|
||||
print(f" Site : {app.racine}")
|
||||
print(" (Ctrl+C pour arrêter)")
|
||||
if ouvrir:
|
||||
threading.Timer(0.5, lambda: webbrowser.open(url)).start()
|
||||
try:
|
||||
serveur.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
taches.arreter()
|
||||
print("\nInterface arrêtée.")
|
||||
finally:
|
||||
serveur.server_close()
|
||||
Reference in new issue
Block a user