FabriceandClaude Opus 5.5
ada3d76f50
- admin.sh : RTG lancé directement sur le Mac (plus de conteneur), nginx joint l'interface par host.docker.internal:8765 ; variante conteneur conservée (admin-docker.sh, docker-compose-conteneur.yml) - Compression : réglages lus dans les fichiers (CRF, codec), refaire seulement le 1080p ou le 720p, seulement si les réglages ont changé ; réglages vidéo propres à chaque voyage ; « Compression actuelle » affichée - Plus d'arrêt automatique après 2 h d'inactivité (--arret-inactif, 0 par défaut) ; arrêt propre qui attend la fin des encodages en cours - Aide contextuelle dans l'interface (rtg/interface/aide) ; README centré sur l'installation et la mise en place - Cœurs attribués au conteneur détectés (cgroup) et signalés Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Mf2kATMtqVABHSYdxJMnim
118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
"""Petits outils communs : programmes externes, fichiers, affichage."""
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
|
|
EXT_PHOTOS = (".jpg", ".jpeg", ".heic", ".heif")
|
|
EXT_VIDEOS = (".mp4", ".mov")
|
|
|
|
# Demande d'arrêt d'un traitement en cours (bouton « Arrêter » de l'interface).
|
|
ARRET = threading.Event()
|
|
|
|
|
|
class Interruption(Exception):
|
|
pass
|
|
|
|
|
|
def verifier_arret():
|
|
if ARRET.is_set():
|
|
raise Interruption()
|
|
|
|
|
|
def dire(message=""):
|
|
print(message, flush=True)
|
|
|
|
|
|
def programme(nom):
|
|
"""Chemin d'un programme externe (ffmpeg, ffprobe, exiftool), ou None."""
|
|
return shutil.which(nom)
|
|
|
|
|
|
def exiger(nom, conseil):
|
|
chemin = programme(nom)
|
|
if not chemin:
|
|
raise SystemExit(f"❌ {nom} introuvable. {conseil}")
|
|
return chemin
|
|
|
|
|
|
def est_cache(nom):
|
|
"""Fichiers cachés et « ._nom » AppleDouble que macOS dépose sur les
|
|
disques non-HFS : jamais des médias."""
|
|
return nom.startswith(".")
|
|
|
|
|
|
def lister_fichiers(sources, extensions, base="."):
|
|
"""Fichiers (chemins absolus, triés, sans doublon) trouvés dans une liste
|
|
de fichiers et de dossiers ; chemins relatifs résolus depuis `base`.
|
|
Les dossiers sont parcourus sur un niveau, comme les anciens scripts."""
|
|
vus, res = set(), []
|
|
for s in sources:
|
|
p = os.path.abspath(os.path.join(base, os.path.expanduser(s)))
|
|
if os.path.isdir(p):
|
|
noms = sorted(os.listdir(p))
|
|
candidats = [os.path.join(p, n) for n in noms if not est_cache(n)]
|
|
elif os.path.isfile(p):
|
|
candidats = [p]
|
|
else:
|
|
continue
|
|
for c in candidats:
|
|
if os.path.isfile(c) and c.lower().endswith(extensions):
|
|
cle = os.path.realpath(c)
|
|
if cle not in vus:
|
|
vus.add(cle)
|
|
res.append(c)
|
|
return res
|
|
|
|
|
|
def lisible(video):
|
|
"""Vrai si ffprobe sait lire la vidéo (un encodage interrompu laisse un
|
|
gros fichier sans index « moov »)."""
|
|
if not os.path.exists(video) or os.path.getsize(video) == 0:
|
|
return False
|
|
r = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration",
|
|
"-of", "csv=p=0", video], capture_output=True)
|
|
return r.returncode == 0 and r.stdout.strip() != b""
|
|
|
|
|
|
def a_jour(sortie, source, force=False):
|
|
return (not force and os.path.exists(sortie) and os.path.getsize(sortie) > 0
|
|
and os.path.getmtime(sortie) >= os.path.getmtime(source))
|
|
|
|
|
|
def coeurs_performance():
|
|
"""Cœurs « performance » sur Apple Silicon (les cœurs d'efficience
|
|
ralentiraient tout le pool de threads x265), sinon les cœurs utilisables
|
|
(dans un conteneur : ceux qui lui sont attribués)."""
|
|
if hasattr(os, "sched_getaffinity") and sys.platform != "darwin":
|
|
try:
|
|
n = len(os.sched_getaffinity(0))
|
|
quota = _quota_cgroup()
|
|
return max(1, min(n, quota) if quota else n)
|
|
except OSError:
|
|
pass
|
|
if sys.platform == "darwin":
|
|
for cle in ("hw.perflevel0.logicalcpu", "hw.logicalcpu"):
|
|
r = subprocess.run(["sysctl", "-n", cle], capture_output=True, text=True)
|
|
if r.returncode == 0 and r.stdout.strip().isdigit():
|
|
return int(r.stdout.strip())
|
|
return os.cpu_count() or 4
|
|
|
|
|
|
def _quota_cgroup():
|
|
"""Limite de processeurs d'un conteneur (cgroup v2 « cpu.max »), ou None."""
|
|
try:
|
|
with open("/sys/fs/cgroup/cpu.max") as f:
|
|
quota, periode = f.read().split()
|
|
return None if quota == "max" else max(1, int(int(quota) / int(periode)))
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
|
|
def taille_lisible(octets):
|
|
for unite in ("o", "Ko", "Mo", "Go"):
|
|
if octets < 1024 or unite == "Go":
|
|
return f"{octets:.0f} {unite}" if unite == "o" else f"{octets:.1f} {unite}"
|
|
octets /= 1024
|