# -*- 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") # extensions sources WEBP_EXT = ".webp" class Command(BaseCommand): help = ( "Balaye un ou plusieurs répertoires, liste les images, convertit > min_size en WebP, " "ignore celles déjà .webp (sauf --refresh) et produit un résumé par répertoire." ) 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 skip)." ) parser.add_argument( "--dry-run", action="store_true", help="N'écrit rien: montre 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érifs préalables if not features.check("webp"): raise CommandError("Pillow sans support WebP. Installe libwebp-dev puis réinstalle pillow.") min_bytes = opts["min_size"] * 1024 quality = opts["quality"] exts = tuple("." + e.strip().lower().lstrip(".") for e in opts["extensions"].split(",")) # 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[dirpath] = dict(...) stats = defaultdict(lambda: { "scanned_files": 0, "webp_files": 0, # .webp rencontrés "candidates": 0, # sources > min_size "converted": 0, "skipped_existing": 0, # .webp déjà présent → skip (si pas --refresh) "converted_list": [], # chemins convertis "existing_webp_list": [], # liste des .webp (point 4) "skipped_list": [], # liste des sources skip car webp existait "large_sources_list": [], # sources > min_size (candidats) }) all_existing_webp = [] # liste globale point 4 # 3) Scan + traitement for basedir in final_dirs: self.stdout.write(self.style.NOTICE(f"Scan: {basedir}")) for root, _, files in os.walk(basedir): for fname in files: lf = fname.lower() src = os.path.join(root, fname) stats[basedir]["scanned_files"] += 1 # Cas 4: on recense tous les .webp existants (indépendamment du reste) if lf.endswith(WEBP_EXT): stats[basedir]["webp_files"] += 1 stats[basedir]["existing_webp_list"].append(src) all_existing_webp.append(src) continue # Filtre par extensions sources if not lf.endswith(exts): continue try: size = os.path.getsize(src) except OSError as e: self.stdout.write(self.style.WARNING(f"Skip (stat): {src} -> {e}")) continue if size <= min_bytes: continue # trop petit, on ignore stats[basedir]["candidates"] += 1 stats[basedir]["large_sources_list"].append(src) dst = os.path.splitext(src)[0] + WEBP_EXT # Skip si .webp existe déjà et pas --refresh if not opts["refresh"] and os.path.exists(dst): stats[basedir]["skipped_existing"] += 1 stats[basedir]["skipped_list"].append(src) continue if opts["verbose_paths"]: self.stdout.write(f"> {src}") if opts["dry_run"]: self.stdout.write(self.style.SUCCESS( f"DRY-RUN ✓ {os.path.basename(src)} ({size//1024}Ko) → {os.path.basename(dst)} (q={quality})" )) stats[basedir]["converted"] += 1 stats[basedir]["converted_list"].append(dst + " (dry-run)") continue # Conversion try: img = Image.open(src) # préserve alpha si présent, sinon RGB if img.mode not in ("RGB", "RGBA", "L"): img = img.convert("RGBA" if "A" in img.getbands() else "RGB") tmp = dst + ".tmp" img.save(tmp, "webp", quality=quality, method=6) os.replace(tmp, dst) new_size = os.path.getsize(dst) gain = 100 - int((new_size / size) * 100) self.stdout.write(self.style.SUCCESS( f"✓ {os.path.basename(fname)} {size//1024}Ko → {new_size//1024}Ko (gain {gain}%)" )) stats[basedir]["converted"] += 1 stats[basedir]["converted_list"].append(dst) except Exception as e: self.stdout.write(self.style.ERROR(f"Erreur {fname}: {e}")) # 4) Affichages finaux # Liste des répertoires analysés (point 2) self.stdout.write("") self.stdout.write(self.style.NOTICE("Répertoires analysés:")) for d in final_dirs: self.stdout.write(f" - {d}") # Liste globale des .webp existants (point 4) self.stdout.write("") self.stdout.write(self.style.NOTICE(f"Images déjà en .webp (total {len(all_existing_webp)}):")) if len(all_existing_webp) == 0: self.stdout.write("(aucune)") else: for p in all_existing_webp: self.stdout.write(p) # Résumé par répertoire (point 5) self.stdout.write("") self.stdout.write(self.style.NOTICE("Résumé par répertoire:")) for d in final_dirs: s = stats[d] self.stdout.write( f"* {d}\n" f" - fichiers scannés : {s['scanned_files']}\n" f" - .webp présents : {s['webp_files']}\n" f" - candidats (> {opts['min_size']}Ko) : {s['candidates']}\n" f" - convertis : {s['converted']}\n" f" - skip (webp exist.) : {s['skipped_existing']}" ) # Rapport JSON optionnel 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, } # default=str pour sérialiser les defaultdict 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}"))