your_project/
        tools/                            # ou accounts/, toolbox/, etc.
        management/
        __init__.py
        commands/
        __init__.py
        compress_images.py          # <= notre script

# -*- coding: utf-8 -*-
import os
import json
from collections import defaultdict

from PIL import Image, features
from django.conf import settings
from django.core.management.base import BaseCommand, CommandError


DEFAULT_EXTS = (".png", ".jpg", ".jpeg")
WEBP_EXT = ".webp"


class Command(BaseCommand):
help = (
"Compresse les images en WebP dans un ou plusieurs répertoires.\n\n"
"Exemple d'utilisation :\n"
"  python manage.py compress_images --include-static-root --min_size 800 --quality 80\n"
)

def add_arguments(self, parser):
parser.add_argument(
"--dir", action="append", default=[],
help="Chemin absolu d'un répertoire à balayer (répéter l'option pour en ajouter plusieurs)."
)
parser.add_argument(
"--include-static-dirs", action="store_true",
help="Inclure settings.STATICFILES_DIRS (racines) dans le scan."
)
parser.add_argument(
"--include-static-root", action="store_true",
help="Inclure settings.STATIC_ROOT dans le scan (si défini)."
)
parser.add_argument(
"--min_size", type=int, default=800,
help="Seuil en Ko au-dessus duquel compresser (défaut 800)."
)
parser.add_argument(
"--quality", type=int, default=80,
help="Qualité WebP (60–95 recommandé, défaut 80)."
)
parser.add_argument(
"--extensions", type=str, default="png,jpg,jpeg",
help="Extensions sources séparées par des virgules (défaut: png,jpg,jpeg)."
)
parser.add_argument(
"--refresh", action="store_true",
help="Re-génère le .webp même s'il existe (sinon on skip)."
)
parser.add_argument(
"--dry-run", action="store_true",
help="N'écrit rien: montre simplement ce qui serait fait."
)
parser.add_argument(
"--report-json", type=str, default="",
help="Chemin d'un fichier JSON pour sauvegarder le rapport détaillé."
)
parser.add_argument(
"--verbose-paths", action="store_true",
help="Affiche les chemins complets lors du traitement."
)

def handle(self, *args, **opts):
# 0) Vérif support WebP
if not features.check("webp"):
raise CommandError(
"Pillow n'a pas le support WebP activé. "
"Recompiler/install Pillow avec WebP (libwebp)."
)

min_bytes = opts["min_size"] * 1024
quality = opts["quality"]
exts = tuple(
"." + e.strip().lower().lstrip(".")
for e in opts["extensions"].split(",")
if e.strip()
)
if not exts:
exts = DEFAULT_EXTS

# 1) Construire la liste des répertoires à analyser
scan_dirs = []

for d in opts["dir"]:
ad = os.path.abspath(d)
if not os.path.isdir(ad):
raise CommandError(f"--dir non valide (pas un répertoire): {ad}")
scan_dirs.append(ad)

if opts["include_static_dirs"]:
for d in getattr(settings, "STATICFILES_DIRS", []):
scan_dirs.append(os.path.abspath(str(d)))

if opts["include_static_root"] and getattr(settings, "STATIC_ROOT", None):
scan_dirs.append(os.path.abspath(str(settings.STATIC_ROOT)))

# Unicité + existence
final_dirs = []
seen = set()
for d in scan_dirs:
if d not in seen and os.path.isdir(d):
final_dirs.append(d)
seen.add(d)

if not final_dirs:
raise CommandError(
"Aucun répertoire valide à scanner. Utilise --dir /chemin/abs "
"et/ou --include-static-dirs / --include-static-root."
)

# 2) Structures de stats
stats = defaultdict(lambda: {
"scanned_files": 0,
"webp_files": 0,
"candidates": 0,
"converted": 0,
"skipped_existing": 0,
"converted_list": [],
"existing_webp_list": [],
"skipped_list": [],
})
all_existing_webp = []

dry = opts["dry_run"]
verbose = opts["verbose_paths"]

self.stdout.write(self.style.MIGRATE_HEADING("Compress Images – démarrage"))
self.stdout.write(
f"Extensions sources: {', '.join(exts)} | seuil: {opts['min_size']} Ko | "
f"qualité WebP: {quality}"
)
self.stdout.write("Répertoires scannés :")
for d in final_dirs:
self.stdout.write(f"  - {d}")
self.stdout.write("")

# 3) Parcours des répertoires
for base in final_dirs:
for root, dirs, files in os.walk(base):
for fname in files:
fpath = os.path.join(root, fname)
rel = os.path.relpath(fpath, base)
st = stats[base]
st["scanned_files"] += 1

lower = fname.lower()
# déjà un .webp
if lower.endswith(WEBP_EXT):
st["webp_files"] += 1
st["existing_webp_list"].append(fpath)
all_existing_webp.append(fpath)
continue

# Extension source non ciblée
if not lower.endswith(exts):
continue

try:
size = os.path.getsize(fpath)
except OSError:
continue

if size <= min_bytes:
# Trop petit, pas intéressant
st["skipped_list"].append(f"{fpath} (trop petit)")
continue

st["candidates"] += 1
dst = os.path.splitext(fpath)[0] + WEBP_EXT

# WebP déjà présent
if os.path.exists(dst) and not opts["refresh"]:
st["skipped_existing"] += 1
st["skipped_list"].append(f"{dst} (existe déjà)")
if verbose:
self.stdout.write(
f"[SKIP] {rel} → .webp existe déjà (utiliser --refresh pour régénérer)"
)
continue

# Conversion
msg_prefix = "[DRY-RUN]" if dry else "[WEBP]"
if verbose:
self.stdout.write(f"{msg_prefix} {rel}")

if dry:
st["converted"] += 1
st["converted_list"].append(f"{fpath} (dry-run)")
continue

# Écriture réelle
tmp = dst + ".tmp"
try:
with Image.open(fpath) as im:
if im.mode not in ("RGB", "RGBA"):
im = im.convert("RGB")
im.save(
tmp,
"WEBP",
quality=quality,
method=6,
)
os.replace(tmp, dst)
st["converted"] += 1
st["converted_list"].append(dst)

new_size = os.path.getsize(dst)
gain = 100.0 * (1 - new_size / float(size))
self.stdout.write(
self.style.SUCCESS(
f"✓ {rel} → {os.path.basename(dst)} "
f"(gain {gain:.1f} %)"
)
)
except Exception as e:
if os.path.exists(tmp):
try:
os.remove(tmp)
except OSError:
pass
self.stdout.write(
self.style.ERROR(f"Erreur sur {rel}: {e}")
)

# 4) Résumé
self.stdout.write("")
self.stdout.write(self.style.MIGRATE_HEADING("Résumé par répertoire"))
for base in final_dirs:
st = stats[base]
self.stdout.write(self.style.NOTICE(f"[{base}]"))
self.stdout.write(f"  Fichiers scannés      : {st['scanned_files']}")
self.stdout.write(f"  Fichiers .webp trouvés: {st['webp_files']}")
self.stdout.write(f"  Candidats > min_size : {st['candidates']}")
self.stdout.write(f"  Convertis en .webp    : {st['converted']}")
self.stdout.write(f"  Skip (webp existant)  : {st['skipped_existing']}")
self.stdout.write("")

if opts["report_json"]:
try:
payload = {
"min_size_ko": opts["min_size"],
"quality": quality,
"extensions": list(exts),
"directories_scanned": final_dirs,
"global_existing_webp": all_existing_webp,
"stats_by_dir": stats,
}
with open(opts["report_json"], "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False, default=str)
self.stdout.write(
self.style.SUCCESS(f"Rapport JSON écrit: {opts['report_json']}")
)
except Exception as e:
self.stdout.write(
self.style.ERROR(f"Impossible d'écrire le rapport JSON: {e}")
)

python manage.py compress_images \
--dir /opt/ideo-lab/static/img/supercomputers/elcapitan \
--min_size 400 \
--quality 85 \
--dry-run \
--verbose-paths

python manage.py compress_images \
--include-static-root \
--min_size 800 \
--quality 80

python manage.py compress_images \
--include-static-dirs \
--include-static-root \
--min_size 600 \
--quality 80 \
--extensions png,jpg,jpeg

python manage.py compress_images \
--include-static-root \
--min_size 500 \
--quality 90 \
--refresh

python manage.py compress_images \
--include-static-root \
--min_size 700 \
--quality 80 \
--report-json /var/log/ideo-lab/compress_images_report.json

0 4 * * * /opt/ideo-lab/venv/bin/python \
/opt/ideo-lab/manage.py compress_images \
--include-static-root \
--min_size 800 \
--quality 80 \
--report-json /var/log/ideo-lab/compress_images_report.json \
>> /var/log/ideo-lab/compress_images.log 2>&1