FabriceandClaude Opus 5.5
4366db94ed
site.toml [administration] adresse : l'accueil affiche un bouton « Connexion » vers l'interface. Celle-ci peut tourner derrière le serveur web du site (rtg interface --proxy --prefixe /admin) : adresses préfixées, mot de passe obligatoire, visiteur identifié par X-Forwarded-For, cookie Secure limité au préfixe quand le site est en HTTPS. Exemples nginx et Apache dans le README. Le bouton Arrêter ne remplace plus le bloc de la tâche en même temps que la relève. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Mf2kATMtqVABHSYdxJMnim
892 lines
42 KiB
Python
892 lines
42 KiB
Python
"""Interface de RTG : un petit serveur web, 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/. Par défaut le serveur n'écoute que sur 127.0.0.1 ;
|
||
`--distant` l'ouvre au réseau, avec mot de passe obligatoire.
|
||
|
||
Sécurité (voir aussi securite.py) :
|
||
- il faut la clé d'accès du lancement ou le mot de passe pour entrer ;
|
||
- 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) ;
|
||
- en local, l'en-tête Host doit être le nôtre (contre le « DNS rebinding ») ;
|
||
- les pages de l'interface interdisent leur affichage dans un cadre et
|
||
n'exécutent que les scripts servis par RTG (Content-Security-Policy).
|
||
"""
|
||
import io
|
||
import json
|
||
import ssl
|
||
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__, acces, accueil, corrections, 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.interface.securite import Gardien
|
||
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, prefixe=""):
|
||
self.racine = os.path.abspath(racine)
|
||
# Préfixe d'adresse quand l'interface est servie sous le site
|
||
# (https://…/admin/ derrière le serveur web) : « /admin ».
|
||
self.prefixe = ("/" + prefixe.strip("/")) if prefixe.strip("/") else ""
|
||
ENV.globals["base"] = self.prefixe
|
||
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"/voyage/carte.json", self.voyage_carte),
|
||
("POST", r"/voyage/position", self.voyage_position),
|
||
("POST", r"/voyage/masque", self.voyage_masque),
|
||
("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": self.prefixe + "/"})
|
||
|
||
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"{self.prefixe}/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 ("export_photos.csv", "export_videos.csv")]
|
||
# Seuls les réglages qui changent l'analyse comptent : déplacer une
|
||
# photo sur la carte ne doit pas faire relire toutes les traces.
|
||
reglages = json.dumps([cfg["sources"], cfg.get("dates_traces", {}), cfg["journees"],
|
||
cfg.get("exclusions", {})], sort_keys=True, default=str)
|
||
return tuple(etat(f) for f in fichiers + autres) + (reglages,)
|
||
|
||
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)
|
||
|
||
# --- Carte et corrections
|
||
def _media_json(self, chemin, rel, m):
|
||
nom = m["fichier"]
|
||
if m["type"] == "photo":
|
||
dossier = "photos_web" if os.path.exists(os.path.join(chemin, "photos_web", nom)) else "photos"
|
||
vignette = f"{self.prefixe}/site/{quote(rel)}/{dossier}/{quote(nom)}"
|
||
else:
|
||
base = os.path.splitext(nom)[0]
|
||
ap = os.path.join(chemin, "apercus", base + ".jpg")
|
||
vignette = f"{self.prefixe}/site/{quote(rel)}/apercus/{quote(base)}.jpg" if os.path.exists(ap) else None
|
||
d = m["date"][:10].replace(":", "-")
|
||
return dict(m, jour=d, heure=m["date"][11:16], vignette=vignette)
|
||
|
||
def _masques_json(self, cfg):
|
||
return {"masques": cfg.get("corrections", {}).get("masques", []),
|
||
"zones": cfg.get("confidentialite", {}).get("zones", [])}
|
||
|
||
def voyage_carte(self, r):
|
||
rel = r.q("v")
|
||
chemin = self.chemin_voyage(rel)
|
||
cfg, a, jours, _ = self.analyse(chemin)
|
||
cfg = self.cfg_voyage(chemin) # positions et masques à jour
|
||
medias = [self._media_json(chemin, rel, m) for m in corrections.lire_medias(chemin, cfg)]
|
||
return json_rep(dict(
|
||
self._masques_json(cfg),
|
||
couches=[dict(c, jour=j["date"].isoformat()) for c, j in zip(_couches(a, jours), jours)],
|
||
jours=self._jours_carte(jours, medias),
|
||
medias=medias))
|
||
|
||
@staticmethod
|
||
def _jours_carte(jours, medias):
|
||
"""Jours des traces, plus ceux où seuls des médias existent."""
|
||
res = {j["date"].isoformat(): {"jour": j["date"].isoformat(), "nom": _date_longue(j["date"])}
|
||
for j in jours}
|
||
for m in medias:
|
||
if m["jour"] not in res:
|
||
try:
|
||
d = date.fromisoformat(m["jour"])
|
||
except ValueError:
|
||
continue
|
||
res[m["jour"]] = {"jour": m["jour"], "nom": _date_longue(d) + " (sans trace)"}
|
||
return [res[k] for k in sorted(res)]
|
||
|
||
def voyage_position(self, r):
|
||
rel = r.f("v")
|
||
chemin = self.chemin_voyage(rel)
|
||
nom = r.f("fichier")
|
||
if not nom:
|
||
raise Erreur(400, "Média non précisé")
|
||
if r.f("lat") and r.f("lon"):
|
||
try:
|
||
lat, lon = float(r.f("lat")), float(r.f("lon"))
|
||
except ValueError:
|
||
raise Erreur(400, "Position invalide")
|
||
if not (-90 <= lat <= 90 and -180 <= lon <= 180):
|
||
raise Erreur(400, "Position invalide")
|
||
self.toml_voyage(chemin, {"positions": {nom: [round(lat, 6), round(lon, 6)]}})
|
||
else:
|
||
self.toml_voyage(chemin, {"positions": {nom: None}})
|
||
cfg = self.cfg_voyage(chemin)
|
||
m = next((m for m in corrections.lire_medias(chemin, cfg) if m["fichier"] == nom), None)
|
||
return json_rep({"media": self._media_json(chemin, rel, m) if m else None})
|
||
|
||
def voyage_masque(self, r):
|
||
chemin = self.chemin_voyage(r.f("v"))
|
||
portee = "site" if r.f("portee") == "site" else "voyage"
|
||
cfg = self.cfg_voyage(chemin)
|
||
liste = list(cfg["confidentialite"]["zones"] if portee == "site" else cfg["corrections"]["masques"])
|
||
if r.f("action") == "ajouter":
|
||
try:
|
||
lat, lon, rayon = float(r.f("lat")), float(r.f("lon")), int(float(r.f("rayon")))
|
||
except ValueError:
|
||
raise Erreur(400, "Zone invalide")
|
||
liste.append([round(lat, 6), round(lon, 6), max(10, min(rayon, 20000))])
|
||
elif r.f("action") == "supprimer":
|
||
try:
|
||
del liste[int(r.f("index"))]
|
||
except (ValueError, IndexError):
|
||
raise Erreur(400, "Zone introuvable")
|
||
if portee == "site":
|
||
modifier_toml(os.path.join(self.racine, SITE_TOML), {"confidentialite": {"zones": liste}},
|
||
entete="Réglages du site RoadTripsGenerator (rtg).")
|
||
else:
|
||
self.toml_voyage(chemin, {"corrections": {"masques": liste}})
|
||
return json_rep(self._masques_json(self.cfg_voyage(chemin)))
|
||
|
||
# --- 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},
|
||
"administration": {"adresse": r.f("administration").strip()},
|
||
}, 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):
|
||
# La relève suivante (chaque seconde) affichera l'arrêt : remplacer le
|
||
# bloc ici en même temps qu'elle ferait échouer l'une des deux.
|
||
taches.arreter()
|
||
return Reponse("")
|
||
|
||
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)
|
||
|
||
|
||
# Pages de l'interface : scripts et styles de RTG seulement, tuiles des fonds
|
||
# de carte, aucun affichage dans un cadre.
|
||
CSP = ("default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; "
|
||
"img-src 'self' data: https://tile.openstreetmap.org https://server.arcgisonline.com; "
|
||
"connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'")
|
||
PUBLIQUES = {"/connexion"}
|
||
|
||
|
||
def creer_gestionnaire(app, port, gardien=None):
|
||
gardien = gardien or Gardien(port, prefixe=app.prefixe)
|
||
hotes = {f"127.0.0.1:{port}", f"localhost:{port}"}
|
||
P = app.prefixe
|
||
# « Voir le site » : derrière le serveur web, le vrai site ; sinon le site servi par RTG.
|
||
ENV.globals["lien_site"] = (P.rsplit("/", 1)[0] + "/") if gardien.proxy else P + "/site/"
|
||
|
||
class Gestionnaire(BaseHTTPRequestHandler):
|
||
server_version = f"rtg/{__version__}"
|
||
sys_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 session(self):
|
||
for morceau in (self.headers.get("Cookie") or "").split(";"):
|
||
nom, _, valeur = morceau.strip().partition("=")
|
||
if nom == gardien.cookie:
|
||
return valeur
|
||
return ""
|
||
|
||
def rediriger(self, cible, cookie=None):
|
||
self.send_response(303)
|
||
self.send_header("Location", cible)
|
||
if cookie:
|
||
self.send_header("Set-Cookie", cookie)
|
||
self.send_header("Content-Length", "0")
|
||
self.end_headers()
|
||
|
||
def ip(self):
|
||
"""Adresse du visiteur ; derrière le serveur web, celle qu'il transmet."""
|
||
if gardien.proxy:
|
||
transmise = (self.headers.get("X-Forwarded-For") or "").split(",")[-1].strip()
|
||
if transmise:
|
||
return transmise
|
||
return self.client_address[0]
|
||
|
||
def traiter(self, methode, tete=False):
|
||
try:
|
||
if not (gardien.distant or gardien.proxy) and self.headers.get("Host") not in hotes:
|
||
raise Erreur(403, "Hôte refusé")
|
||
url = urlsplit(self.path)
|
||
chemin = unquote(url.path)
|
||
if P:
|
||
if chemin == P:
|
||
return self.rediriger(P + "/")
|
||
if not chemin.startswith(P + "/"):
|
||
raise Erreur(404, "Page introuvable")
|
||
chemin = chemin[len(P):]
|
||
query = parse_qs(url.query, keep_blank_values=True)
|
||
if methode == "GET" and chemin.startswith("/static/"):
|
||
return self.fichier(STATIC, chemin[len("/static/"):], tete)
|
||
# Adresse ouverte au lancement : la clé ouvre une session, puis
|
||
# on la retire de l'adresse (historique, copier-coller).
|
||
if "jeton" in query:
|
||
if gardien.jeton_valide(_valeur(query, "jeton", "")):
|
||
sid = gardien.ouvrir_session()
|
||
return self.rediriger(P + chemin, gardien.entete_cookie(sid, self.https()))
|
||
return self.rediriger(P + "/connexion")
|
||
connecte = gardien.session_valide(self.session())
|
||
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)
|
||
if n > 2_000_000:
|
||
raise Erreur(413, "Requête trop grosse")
|
||
form = parse_qs(self.rfile.read(n).decode("utf-8"), keep_blank_values=True)
|
||
if chemin in PUBLIQUES:
|
||
return self.envoyer(self.connexion(methode, form, connecte), tete)
|
||
if not connecte:
|
||
if self.headers.get("HX-Request") == "true":
|
||
return self.envoyer(Reponse("", entetes={"HX-Redirect": P + "/connexion"}), tete)
|
||
return self.rediriger(P + "/connexion")
|
||
if chemin == "/deconnexion" and methode == "POST":
|
||
gardien.fermer_session(self.session())
|
||
return self.envoyer(Reponse("", entetes={
|
||
"HX-Redirect": P + "/connexion",
|
||
"Set-Cookie": gardien.entete_cookie("", self.https(), effacer=True)}), tete)
|
||
if methode == "GET" and chemin.startswith("/site/"):
|
||
return self.fichier(app.racine, chemin[len("/site/"):], tete)
|
||
req = Requete(query, 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 https(self):
|
||
return gardien.https or (gardien.proxy and self.headers.get("X-Forwarded-Proto") == "https")
|
||
|
||
def connexion(self, methode, form, connecte):
|
||
ip = self.ip()
|
||
if methode == "POST":
|
||
ok, message = gardien.essayer(ip, _valeur(form, "mot_de_passe", ""))
|
||
if ok:
|
||
sid = gardien.ouvrir_session()
|
||
return Reponse("", entetes={"HX-Redirect": P + "/",
|
||
"Set-Cookie": gardien.entete_cookie(sid, self.https())})
|
||
return Reponse(f'<div class="erreur">{_echapper(message)}</div>')
|
||
return page("connexion.html", mdp=acces.defini(), connecte=connecte,
|
||
distant=gardien.distant or gardien.proxy, attente=gardien.attente(ip),
|
||
site=self.adresse_site())
|
||
|
||
def adresse_site(self):
|
||
"""Lien « Retour au site » de la page de connexion, derrière le serveur web."""
|
||
return (P.rsplit("/", 1)[0] or "") + "/" if gardien.proxy else ""
|
||
|
||
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")
|
||
self.send_header("X-Content-Type-Options", "nosniff")
|
||
self.send_header("X-Frame-Options", "DENY")
|
||
self.send_header("Content-Security-Policy", CSP)
|
||
self.send_header("Referrer-Policy", "strict-origin-when-cross-origin")
|
||
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, distant=False, certificat=None, cle=None,
|
||
proxy=False, prefixe=""):
|
||
"""Lance l'interface.
|
||
distant : écoute sur toutes les interfaces réseau, mot de passe obligatoire ;
|
||
certificat, cle : HTTPS ;
|
||
proxy : derrière le serveur web du site (qui transmet https://…/admin/ à
|
||
127.0.0.1:port), mot de passe obligatoire ; prefixe : « /admin »."""
|
||
if (distant or proxy) and not acces.defini():
|
||
raise SystemExit("❌ L'accès depuis le réseau demande un mot de passe : lancez d'abord « rtg mot-de-passe ».")
|
||
contexte_tls = None
|
||
if certificat:
|
||
contexte_tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||
contexte_tls.minimum_version = ssl.TLSVersion.TLSv1_2
|
||
contexte_tls.load_cert_chain(certificat, cle)
|
||
app = Application(racine, prefixe)
|
||
serveur = None
|
||
# Derrière le serveur web, le port doit être exactement celui de sa configuration.
|
||
for p in ([port] if proxy else range(port, port + 20)):
|
||
try:
|
||
gardien = Gardien(p, distant=distant, https=bool(contexte_tls), proxy=proxy, prefixe=app.prefixe)
|
||
serveur = ThreadingHTTPServer(("0.0.0.0" if distant else "127.0.0.1", p),
|
||
creer_gestionnaire(app, p, gardien))
|
||
port = p
|
||
break
|
||
except OSError:
|
||
continue
|
||
if serveur is None:
|
||
raise SystemExit(f"❌ Port {port} occupé." if proxy else f"❌ Aucun port libre entre {port} et {port + 19}.")
|
||
if contexte_tls:
|
||
serveur.socket = contexte_tls.wrap_socket(serveur.socket, server_side=True)
|
||
serveur.daemon_threads = True
|
||
schema = "https" if contexte_tls else "http"
|
||
url = f"{schema}://127.0.0.1:{port}{app.prefixe}/?jeton={gardien.jeton}"
|
||
print(f"🌐 Interface RTG : {url}")
|
||
print(" (cette adresse contient la clé d'accès de cette session : ne la partagez pas)")
|
||
if proxy:
|
||
print(f" Derrière le serveur web : https://<votre site>{app.prefixe}/ (mot de passe)")
|
||
if distant:
|
||
print(f" Accès depuis le réseau : {schema}://<nom ou adresse de cette machine>:{port}/ (mot de passe)")
|
||
if not contexte_tls:
|
||
print(" ⚠️ Sans HTTPS, le mot de passe circule en clair : placez l'interface derrière votre")
|
||
print(" serveur web en HTTPS, utilisez --certificat/--cle, ou un tunnel SSH.")
|
||
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()
|