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
126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
"""Un traitement à la fois (génération, compression…) dans un fil à part.
|
|
|
|
Tout ce que la chaîne affiche est recopié dans le journal de la tâche, en
|
|
plus du terminal ; l'interface relit ce journal et l'avancement chaque seconde.
|
|
"""
|
|
import sys
|
|
import threading
|
|
import time
|
|
import traceback
|
|
from collections import deque
|
|
|
|
from rtg.outils import ARRET, Interruption
|
|
|
|
_verrou = threading.Lock()
|
|
COURANTE = None
|
|
|
|
|
|
class Tache:
|
|
def __init__(self, libelle, voyage=""):
|
|
self.libelle = libelle
|
|
self.voyage = voyage # chemin relatif, "" pour tout le site
|
|
self.lignes = deque(maxlen=3000)
|
|
self.etape = ""
|
|
self.etape_voyage = "" # voyage en cours quand tout le site est généré
|
|
self.video = None # {"nom", "pct", "rang": (i, n)}
|
|
self.debut = time.time()
|
|
self.fin = None
|
|
self.ok = None
|
|
self.arretee = False
|
|
|
|
@property
|
|
def en_cours(self):
|
|
return self.fin is None
|
|
|
|
@property
|
|
def duree(self):
|
|
s = int((self.fin or time.time()) - self.debut)
|
|
return f"{s // 60} min {s % 60:02d} s" if s >= 60 else f"{s} s"
|
|
|
|
def suivi(self, ev):
|
|
if "etape" in ev:
|
|
self.etape = ev["etape"]
|
|
self.etape_voyage = ev.get("voyage", "") if ev.get("voyage") != self.voyage else ""
|
|
if ev["etape"] != "Vidéos":
|
|
self.video = None
|
|
elif "video" in ev:
|
|
if ev["etat"] == "en_cours":
|
|
rang = ev.get("rang") or (self.video or {}).get("rang")
|
|
self.video = {"nom": ev["video"], "pct": ev["pct"] or 0, "rang": rang}
|
|
elif ev.get("rang"):
|
|
self.video = {"nom": ev["video"], "pct": 100, "rang": ev["rang"]}
|
|
|
|
|
|
class _Recopie:
|
|
"""Remplace sys.stdout pendant une tâche : écrit dans le terminal et dans
|
|
le journal de la tâche (ligne par ligne ; un « \\r » remplace la ligne)."""
|
|
|
|
def __init__(self, original, tache):
|
|
self.original, self.tache, self.tampon = original, tache, ""
|
|
|
|
def write(self, s):
|
|
self.original.write(s)
|
|
self.tampon += s
|
|
while "\n" in self.tampon:
|
|
ligne, self.tampon = self.tampon.split("\n", 1)
|
|
self.tache.lignes.append(ligne.rsplit("\r", 1)[-1])
|
|
return len(s)
|
|
|
|
def flush(self):
|
|
self.original.flush()
|
|
|
|
def isatty(self):
|
|
return False
|
|
|
|
def __getattr__(self, nom):
|
|
return getattr(self.original, nom)
|
|
|
|
|
|
def lancer(libelle, fonction, voyage=""):
|
|
"""Démarre `fonction(suivi)` dans un fil. Renvoie la tâche, ou None si une
|
|
autre tâche tourne encore."""
|
|
global COURANTE
|
|
with _verrou:
|
|
if COURANTE and COURANTE.en_cours:
|
|
return None
|
|
t = Tache(libelle, voyage)
|
|
COURANTE = t
|
|
ARRET.clear()
|
|
|
|
def executer():
|
|
original = sys.stdout
|
|
sys.stdout = _Recopie(original, t)
|
|
try:
|
|
res = fonction(t.suivi)
|
|
t.ok = res is not False
|
|
except Interruption:
|
|
t.ok, t.arretee = False, True
|
|
print("⏹ Traitement arrêté.")
|
|
except SystemExit as e:
|
|
t.ok = False
|
|
if e.code not in (None, 0):
|
|
print(str(e.code))
|
|
except Exception:
|
|
t.ok = False
|
|
print("❌ Erreur inattendue :")
|
|
print(traceback.format_exc())
|
|
finally:
|
|
if ARRET.is_set():
|
|
t.arretee = True
|
|
sys.stdout.flush()
|
|
sys.stdout = original
|
|
t.fin = time.time()
|
|
t.etape = "Arrêté" if t.arretee else ("Terminé" if t.ok else "Terminé avec des erreurs")
|
|
t.video = None
|
|
t.etape_voyage = ""
|
|
|
|
threading.Thread(target=executer, daemon=True, name="rtg-tache").start()
|
|
return t
|
|
|
|
|
|
def arreter():
|
|
if COURANTE and COURANTE.en_cours:
|
|
ARRET.set()
|
|
return True
|
|
return False
|