FabriceandClaude Opus 5.5
71779f6f53
Onglet « Carte et corrections » : glisser une photo ou une vidéo, placer les médias sans position, masquer des zones de trace (pour un voyage, ou zone privée pour tous). Corrections notées dans voyage.toml ([positions], [corrections]) ou site.toml ([confidentialite]), appliquées par les pages ; les originaux ne changent pas. L'extraction garde les médias datés sans position pour qu'on puisse les placer. Accès : clé d'accès aléatoire à chaque lancement, session par cookie HttpOnly/SameSite=Strict, mot de passe facultatif (rtg mot-de-passe, empreinte scrypt hors du site, blocage après 5 échecs), accès réseau seulement avec mot de passe (--distant), HTTPS possible (--certificat, --cle), Content-Security-Policy et interdiction des cadres. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Mf2kATMtqVABHSYdxJMnim
851 lines
39 KiB
Python
851 lines
39 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):
|
||
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"/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": "/"})
|
||
|
||
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 ("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"/site/{quote(rel)}/{dossier}/{quote(nom)}"
|
||
else:
|
||
base = os.path.splitext(nom)[0]
|
||
ap = os.path.join(chemin, "apercus", base + ".jpg")
|
||
vignette = f"/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},
|
||
}, 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)
|
||
|
||
|
||
# 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)
|
||
hotes = {f"127.0.0.1:{port}", f"localhost:{port}"}
|
||
|
||
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 traiter(self, methode, tete=False):
|
||
try:
|
||
if not gardien.distant and self.headers.get("Host") not in hotes:
|
||
raise Erreur(403, "Hôte refusé")
|
||
url = urlsplit(self.path)
|
||
chemin = unquote(url.path)
|
||
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(url.path or "/", gardien.entete_cookie(sid))
|
||
return self.rediriger("/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": "/connexion"}), tete)
|
||
return self.rediriger("/connexion")
|
||
if chemin == "/deconnexion" and methode == "POST":
|
||
gardien.fermer_session(self.session())
|
||
return self.envoyer(Reponse("", entetes={
|
||
"HX-Redirect": "/connexion", "Set-Cookie": gardien.entete_cookie("", 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 connexion(self, methode, form, connecte):
|
||
ip = self.client_address[0]
|
||
if methode == "POST":
|
||
ok, message = gardien.essayer(ip, _valeur(form, "mot_de_passe", ""))
|
||
if ok:
|
||
sid = gardien.ouvrir_session()
|
||
return Reponse("", entetes={"HX-Redirect": "/", "Set-Cookie": gardien.entete_cookie(sid)})
|
||
return Reponse(f'<div class="erreur">{_echapper(message)}</div>')
|
||
return page("connexion.html", mdp=acces.defini(), connecte=connecte,
|
||
distant=gardien.distant, attente=gardien.attente(ip))
|
||
|
||
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):
|
||
"""Lance l'interface. distant=True : écoute sur toutes les interfaces
|
||
réseau, mot de passe obligatoire ; certificat et cle : HTTPS."""
|
||
if distant and not acces.defini():
|
||
raise SystemExit("❌ L'accès distant 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)
|
||
serveur = None
|
||
for p in range(port, port + 20):
|
||
try:
|
||
gardien = Gardien(p, distant=distant, https=bool(contexte_tls))
|
||
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"❌ 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}/?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 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()
|