AVIF q60 par défaut et sélecteur de mode de fusion par couche

scripts/transcode_q60.py

@@ -0,0 +1,54 @@
+"""Transcode les AVIF q98 des visualisations en AVIF q60 (one-shot).
+
+Usage (dans le conteneur, volume output monté) :
+    python3 /tmp/transcode_q60.py [--quality 60] [--workers N]
+
+Ne remplace un fichier que si la version q60 est plus légère (garde-fou).
+"""
+import sys
+import os
+from pathlib import Path
+from concurrent.futures import ProcessPoolExecutor
+
+from PIL import Image
+
+QUALITY = int(sys.argv[sys.argv.index('--quality') + 1]) if '--quality' in sys.argv else 60
+WORKERS = int(sys.argv[sys.argv.index('--workers') + 1]) if '--workers' in sys.argv else (os.cpu_count() or 4)
+ROOT = Path('/data/output/visualisations')
+
+
+def transcode(path):
+    try:
+        before = path.stat().st_size
+        img = Image.open(path)
+        img.load()
+        tmp = path.with_name(path.name + '.q60.tmp.avif')
+        img.save(tmp, format='AVIF', quality=QUALITY)
+        after = tmp.stat().st_size
+        if after < before:
+            os.replace(tmp, path)
+            return (path.name, before, after, 'remplacé')
+        tmp.unlink()
+        return (path.name, before, after, 'conservé (q60 plus lourd)')
+    except Exception as exc:  # noqa: BLE001
+        return (path.name, 0, 0, f'ERREUR {exc}')
+
+
+if __name__ == '__main__':
+    files = sorted(ROOT.rglob('*.avif'))
+    files = [f for f in files if '.q60.tmp' not in f.name]
+    print(f'{len(files)} AVIF à transcoder en q{QUALITY}, {WORKERS} workers', flush=True)
+    total_before = total_after = n_ok = 0
+    with ProcessPoolExecutor(max_workers=WORKERS) as pool:
+        for i, (name, before, after, status) in enumerate(pool.map(transcode, files), 1):
+            total_before += before
+            total_after += min(before, after)
+            if status.startswith('remplacé'):
+                n_ok += 1
+            if i % 50 == 0 or i == len(files):
+                print(f'  {i}/{len(files)} — {n_ok} remplacés — '
+                      f'{total_before / 2**30:.1f} → {total_after / 2**30:.1f} Gio', flush=True)
+            elif status.startswith('ERREUR'):
+                print(f'  {name}: {status}', flush=True)
+    print(f'FIN: {total_before / 2**30:.1f} → {total_after / 2**30:.1f} Gio '
+          f'({n_ok} fichiers réencodés)', flush=True)