FabriceandClaude Opus 5.5
06153a227e
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
332 lines
14 KiB
Python
332 lines
14 KiB
Python
"""Analyse des sources d'un voyage et mise en forme dans le dossier du site.
|
|
|
|
Les sources (traces, photos, vidéos) peuvent être n'importe où et nommées
|
|
n'importe comment. Cette étape les range dans l'organisation que le reste
|
|
de la chaîne attend, sans jamais modifier les originaux :
|
|
|
|
routes/AAAA-MM-JJ.gpx une trace par journée (créée par RTG)
|
|
photos/ copies JPEG des photos (HEIC converties)
|
|
|
|
Les vidéos ne sont pas copiées : la compression lit directement les sources.
|
|
Un voyage dont les sources sont déjà ses propres dossiers routes/ et photos/
|
|
(organisation historique) n'est pas touché.
|
|
"""
|
|
import csv
|
|
import math
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from collections import Counter
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import gpxpy
|
|
import gpxpy.gpx
|
|
|
|
from rtg.outils import EXT_PHOTOS, EXT_VIDEOS, est_cache, lister_fichiers
|
|
|
|
CREATEUR = "RoadTripsGenerator"
|
|
DATE_DANS_NOM = re.compile(r"(\d{4})-(\d{2})-(\d{2})")
|
|
|
|
|
|
class Analyse:
|
|
def __init__(self):
|
|
self.journees = {} # date → [GPXTrack]
|
|
self.traces = [] # fichiers GPX lus
|
|
self.photos = [] # chemins des photos sources
|
|
self.videos = [] # chemins des vidéos sources
|
|
self.a_verifier = [] # messages pour l'utilisateur
|
|
self.details_traces = [] # un dict par fichier GPX (pour l'interface)
|
|
self.non_datees = {} # fichier → [GPXTrack] des traces sans jour
|
|
|
|
def km(self, jour):
|
|
total = 0.0
|
|
for t in self.journees.get(jour, []):
|
|
try:
|
|
total += t.length_2d() or 0
|
|
except Exception:
|
|
pass
|
|
return total / 1000.0
|
|
|
|
|
|
def _sources(voyage, cle, cfg):
|
|
return cfg["sources"].get(cle, [])
|
|
|
|
|
|
def _est_dossier_du_voyage(voyage, sources, nom):
|
|
"""Vrai si les sources se résument au sous-dossier `nom` du voyage."""
|
|
cible = os.path.realpath(os.path.join(voyage, nom))
|
|
res = [os.path.realpath(os.path.join(voyage, os.path.expanduser(s))) for s in sources]
|
|
return res == [cible]
|
|
|
|
|
|
def date_du_nom(nom):
|
|
m = DATE_DANS_NOM.search(nom)
|
|
if not m:
|
|
return None
|
|
try:
|
|
return datetime(*map(int, m.groups())).date()
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def jour_texte(valeur):
|
|
"""« 2025-08-04 » (ou date TOML) → date, sinon None."""
|
|
try:
|
|
return valeur if isinstance(valeur, date) else date.fromisoformat(str(valeur).strip())
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def decouper_par_journee(fichier, tz, coupure_h, jour_impose=None, pistes_sans_jour=None, infos=None):
|
|
"""Pistes d'un GPX réparties par journée locale.
|
|
|
|
Renvoie ({date: [GPXTrack]}, None) ou ({}, raison) si aucune date n'est
|
|
trouvée. `jour_impose` (choisi dans l'interface) place tout le fichier sur
|
|
ce jour. Sinon, avec horodatages : date locale dans le fuseau du voyage,
|
|
avec une coupure à `coupure_h` heures (une étape finie à 1 h reste sur la
|
|
veille) ; sans horodatage : la date du nom de fichier. Les pistes d'un
|
|
fichier qu'aucune date ne place sont ajoutées à `pistes_sans_jour`. `infos`
|
|
(dictionnaire) reçoit "datable" : le fichier se date-t-il sans aide ?
|
|
"""
|
|
with open(fichier, encoding="utf-8") as f:
|
|
gpx = gpxpy.parse(f)
|
|
a_des_heures = any(p.time for t in gpx.tracks for s in t.segments for p in s.points)
|
|
if infos is not None:
|
|
infos["datable"] = a_des_heures or date_du_nom(os.path.basename(fichier)) is not None
|
|
res = {}
|
|
if jour_impose or not a_des_heures:
|
|
jour = jour_impose or date_du_nom(os.path.basename(fichier))
|
|
if jour is None:
|
|
if pistes_sans_jour is not None:
|
|
pistes_sans_jour.extend(gpx.tracks)
|
|
return {}, "trace sans heures et sans date dans son nom"
|
|
for t in gpx.tracks:
|
|
res.setdefault(jour, []).append(t)
|
|
return res, None
|
|
|
|
decalage = timedelta(hours=coupure_h)
|
|
for t in gpx.tracks:
|
|
par_jour = {}
|
|
for s in t.segments:
|
|
courant, morceau = None, None
|
|
for p in s.points:
|
|
if p.time is not None:
|
|
heure = p.time if p.time.tzinfo else p.time.replace(tzinfo=timezone.utc)
|
|
jour = (heure.astimezone(tz) - decalage).date()
|
|
else:
|
|
jour = courant
|
|
if jour is None:
|
|
continue
|
|
if jour != courant:
|
|
morceau = gpxpy.gpx.GPXTrackSegment()
|
|
par_jour.setdefault(jour, []).append(morceau)
|
|
courant = jour
|
|
morceau.points.append(p)
|
|
for jour, segments in par_jour.items():
|
|
piste = gpxpy.gpx.GPXTrack(name=t.name)
|
|
piste.segments = segments
|
|
res.setdefault(jour, []).append(piste)
|
|
return res, None
|
|
|
|
|
|
def analyser(voyage, cfg, leger=False):
|
|
"""Inventaire des sources d'un voyage (sans rien écrire).
|
|
|
|
leger=True (pour `rtg etat`) : une trace dont le nom porte la date n'est
|
|
pas lue ; sa journée est comptée, sans les points ni les kilomètres."""
|
|
a = Analyse()
|
|
exclus = set(cfg.get("exclusions", {}).get("fichiers", []))
|
|
tz = ZoneInfo(cfg["journees"]["fuseau"])
|
|
coupure = int(cfg["journees"]["coupure_heure"])
|
|
|
|
imposes = {k: jour_texte(v) for k, v in cfg.get("dates_traces", {}).items()}
|
|
sources_traces = _sources(voyage, "traces", cfg)
|
|
a.traces = [f for f in lister_fichiers(sources_traces, (".gpx",), voyage)
|
|
if os.path.basename(f) not in exclus]
|
|
if not _est_dossier_du_voyage(voyage, sources_traces, "routes"):
|
|
# Les journées que RTG a lui-même écrites dans routes/ ne sont pas des
|
|
# sources : les relire les compterait deux fois.
|
|
routes = os.path.realpath(os.path.join(voyage, "routes"))
|
|
a.traces = [f for f in a.traces
|
|
if os.path.realpath(os.path.dirname(f)) != routes or not ecrite_par_rtg(f)]
|
|
for f in a.traces:
|
|
nom = os.path.basename(f)
|
|
impose = imposes.get(nom)
|
|
if leger and (impose or date_du_nom(nom)):
|
|
a.journees.setdefault(impose or date_du_nom(nom), [])
|
|
continue
|
|
sans_jour, infos = [], {}
|
|
try:
|
|
parts, raison = decouper_par_journee(f, tz, coupure, impose, sans_jour, infos)
|
|
except Exception as e:
|
|
a.a_verifier.append(f"Trace illisible : {nom} ({e})")
|
|
a.details_traces.append({"fichier": f, "nom": nom, "jours": [], "km": 0,
|
|
"impose": impose, "erreur": str(e)})
|
|
continue
|
|
if raison:
|
|
a.a_verifier.append(f"Trace à rattacher à un jour : {nom} ({raison})")
|
|
a.non_datees[f] = sans_jour
|
|
km = 0.0
|
|
for jour, pistes in parts.items():
|
|
a.journees.setdefault(jour, []).extend(pistes)
|
|
km += sum((p.length_2d() or 0) for p in pistes) / 1000
|
|
if sans_jour:
|
|
km = sum((p.length_2d() or 0) for p in sans_jour) / 1000
|
|
a.details_traces.append({"fichier": f, "nom": nom, "jours": sorted(parts), "km": km,
|
|
"impose": impose, "erreur": None, "datable": infos["datable"]})
|
|
|
|
a.photos = [f for f in lister_fichiers(_sources(voyage, "photos", cfg), EXT_PHOTOS, voyage)
|
|
if os.path.basename(f) not in exclus]
|
|
a.videos = [f for f in lister_fichiers(_sources(voyage, "videos", cfg), EXT_VIDEOS, voyage)
|
|
if os.path.basename(f) not in exclus]
|
|
|
|
if _est_dossier_du_voyage(voyage, _sources(voyage, "photos", cfg), "photos"):
|
|
heic = [p for p in a.photos if p.lower().endswith((".heic", ".heif"))]
|
|
if heic:
|
|
a.a_verifier.append(f"{len(heic)} photo(s) HEIC dans photos/ : indiquez ce dossier comme "
|
|
f"source d'un autre emplacement pour qu'elles soient converties")
|
|
for s in _sources(voyage, "videos", cfg) + _sources(voyage, "photos", cfg) + _sources(voyage, "traces", cfg):
|
|
p = os.path.join(voyage, os.path.expanduser(s))
|
|
if os.path.islink(p) and not os.path.exists(p):
|
|
a.a_verifier.append(f"Source introuvable (disque débranché ?) : {s} → {os.readlink(p)}")
|
|
elif not os.path.lexists(p) and s not in ("routes", "rushs", "photos"):
|
|
a.a_verifier.append(f"Source introuvable : {s}")
|
|
return a
|
|
|
|
|
|
def ecrite_par_rtg(fichier):
|
|
with open(fichier, encoding="utf-8", errors="replace") as f:
|
|
return f'creator="{CREATEUR}"' in f.read(600)
|
|
|
|
|
|
def ranger(voyage, cfg, a):
|
|
"""Range les sources dans le dossier du voyage (routes/, photos/).
|
|
Renvoie un résumé des écritures."""
|
|
resume = []
|
|
sources_traces = _sources(voyage, "traces", cfg)
|
|
if a.traces and not _est_dossier_du_voyage(voyage, sources_traces, "routes"):
|
|
resume.append(_ecrire_routes(voyage, a))
|
|
if a.photos and not _est_dossier_du_voyage(voyage, _sources(voyage, "photos", cfg), "photos"):
|
|
resume.append(_copier_photos(voyage, a))
|
|
return [r for r in resume if r]
|
|
|
|
|
|
def _ecrire_routes(voyage, a):
|
|
dossier = os.path.join(voyage, "routes")
|
|
os.makedirs(dossier, exist_ok=True)
|
|
attendus = set()
|
|
for jour, pistes in sorted(a.journees.items()):
|
|
nom = f"{jour.isoformat()}.gpx"
|
|
attendus.add(nom)
|
|
gpx = gpxpy.gpx.GPX()
|
|
gpx.creator = CREATEUR
|
|
for p in pistes:
|
|
gpx.tracks.append(p)
|
|
with open(os.path.join(dossier, nom), "w", encoding="utf-8") as f:
|
|
f.write(gpx.to_xml())
|
|
# Retire les journées que RTG avait écrites et qui n'existent plus ;
|
|
# un GPX déposé à la main n'est jamais touché.
|
|
retires = 0
|
|
for nom in os.listdir(dossier):
|
|
chemin = os.path.join(dossier, nom)
|
|
if nom in attendus or est_cache(nom) or not nom.lower().endswith(".gpx"):
|
|
continue
|
|
if ecrite_par_rtg(chemin):
|
|
os.remove(chemin)
|
|
retires += 1
|
|
return f"🧭 routes/ : {len(attendus)} journée(s) écrite(s)" + (f", {retires} retirée(s)" if retires else "")
|
|
|
|
|
|
def _copier_photos(voyage, a):
|
|
dossier = os.path.join(voyage, "photos")
|
|
os.makedirs(dossier, exist_ok=True)
|
|
copiees = converties = 0
|
|
for src in a.photos:
|
|
if os.path.realpath(os.path.dirname(src)) == os.path.realpath(dossier):
|
|
continue # déjà dans photos/
|
|
base, ext = os.path.splitext(os.path.basename(src))
|
|
heic = ext.lower() in (".heic", ".heif")
|
|
dest = os.path.join(dossier, base + (".jpeg" if heic else ext))
|
|
if os.path.exists(dest) and os.path.getmtime(dest) >= os.path.getmtime(src):
|
|
continue
|
|
if heic:
|
|
convertir_heic(src, dest)
|
|
converties += 1
|
|
else:
|
|
shutil.copy2(src, dest)
|
|
copiees += 1
|
|
if not (copiees or converties):
|
|
return None
|
|
return f"📸 photos/ : {copiees} copiée(s), {converties} HEIC convertie(s) en JPEG"
|
|
|
|
|
|
def convertir_heic(src, dest):
|
|
"""HEIC → JPEG, métadonnées EXIF (date, GPS) conservées."""
|
|
from PIL import Image
|
|
import pillow_heif
|
|
pillow_heif.register_heif_opener()
|
|
with Image.open(src) as im:
|
|
exif = im.info.get("exif")
|
|
im.convert("RGB").save(dest + ".tmp", "JPEG", quality=92, exif=exif or b"")
|
|
os.replace(dest + ".tmp", dest)
|
|
|
|
|
|
# --- Suggestion d'un jour pour une trace sans heures ---------------------------
|
|
def positions_medias(voyage, a):
|
|
"""[(date, lat, lon)] des photos et vidéos du voyage : exports CSV s'ils
|
|
existent, sinon lecture rapide des photos sources par exiftool."""
|
|
res = []
|
|
for nom in ("export_photos.csv", "export_videos.csv"):
|
|
chemin = os.path.join(voyage, nom)
|
|
if not os.path.exists(chemin):
|
|
continue
|
|
with open(chemin, encoding="utf-8", errors="replace") as f:
|
|
for ligne in csv.reader(f):
|
|
try:
|
|
res.append((date(*map(int, ligne[1][:10].replace("-", ":").split(":"))),
|
|
float(ligne[2]), float(ligne[3])))
|
|
except (ValueError, IndexError):
|
|
pass
|
|
if res or not a.photos or not shutil.which("exiftool"):
|
|
return res
|
|
r = subprocess.run(["exiftool", "-n", "-q", "-p", "$DateTimeOriginal,$GPSLatitude,$GPSLongitude"]
|
|
+ a.photos[:2000], capture_output=True, text=True, errors="replace")
|
|
for ligne in r.stdout.splitlines():
|
|
try:
|
|
d, lat, lon = ligne.split(",")
|
|
res.append((date(*map(int, d[:10].split(":"))), float(lat), float(lon)))
|
|
except ValueError:
|
|
pass
|
|
return res
|
|
|
|
|
|
def suggerer_jour(pistes, medias, rayon_km=1.5):
|
|
"""Jour le plus probable pour une trace sans heures : celui dont le plus de
|
|
photos et vidéos ont été prises à moins de `rayon_km` de la trace.
|
|
Renvoie (date, nombre de médias) ou (None, 0)."""
|
|
pts = [(p.latitude, p.longitude) for t in pistes for s in t.segments for p in s.points]
|
|
if not pts or not medias:
|
|
return None, 0
|
|
pas = max(1, len(pts) // 2000)
|
|
pts = pts[::pas]
|
|
# Grille d'environ 1 km : on ne compare un média qu'aux points voisins.
|
|
cellule = 0.01
|
|
grille = {}
|
|
for lat, lon in pts:
|
|
grille.setdefault((math.floor(lat / cellule), math.floor(lon / cellule)), []).append((lat, lon))
|
|
portee = int(rayon_km / 1.1 / (cellule * 100)) + 1
|
|
compte = Counter()
|
|
for jour, lat, lon in medias:
|
|
i, j = math.floor(lat / cellule), math.floor(lon / cellule)
|
|
cos = math.cos(math.radians(lat))
|
|
proche = any(math.hypot((la - lat) * 111.2, (lo - lon) * 111.2 * cos) <= rayon_km
|
|
for di in range(-portee, portee + 1) for dj in range(-portee, portee + 1)
|
|
for la, lo in grille.get((i + di, j + dj), ()))
|
|
if proche:
|
|
compte[jour] += 1
|
|
if not compte:
|
|
return None, 0
|
|
return compte.most_common(1)[0]
|