"""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 os import re import shutil from datetime import 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 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 decouper_par_journee(fichier, tz, coupure_h): """Pistes d'un GPX réparties par journée locale. Renvoie ({date: [GPXTrack]}, None) ou ({}, raison) si aucune date n'est trouvée. 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 vaut pour tout le fichier. """ 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) res = {} if not a_des_heures: jour = date_du_nom(os.path.basename(fichier)) if jour is None: 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): """Inventaire des sources d'un voyage (sans rien écrire).""" a = Analyse() exclus = set(cfg.get("exclusions", {}).get("fichiers", [])) tz = ZoneInfo(cfg["journees"]["fuseau"]) coupure = int(cfg["journees"]["coupure_heure"]) a.traces = [f for f in lister_fichiers(_sources(voyage, "traces", cfg), (".gpx",), voyage) if os.path.basename(f) not in exclus] for f in a.traces: try: parts, raison = decouper_par_journee(f, tz, coupure) except Exception as e: a.a_verifier.append(f"Trace illisible : {os.path.basename(f)} ({e})") continue if raison: a.a_verifier.append(f"Trace à rattacher à un jour : {os.path.basename(f)} ({raison})") for jour, pistes in parts.items(): a.journees.setdefault(jour, []).extend(pistes) 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 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 with open(chemin, encoding="utf-8", errors="replace") as f: debut = f.read(600) if f'creator="{CREATEUR}"' in debut: 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: 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)