RoadTripsGenerator/rtg/sources.py

"""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, 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"])

    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:
        if leger and date_du_nom(os.path.basename(f)):
            a.journees.setdefault(date_du_nom(os.path.basename(f)), [])
            continue
        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)