#! /usr/bin/python3.1 # -*- coding: utf-8 -*- # core/management/commands/migration_doctor.py """ Migration Doctor — audit, restore and cleanup for Django migrations. Highlights (Cron-friendly): - Daily audit: manage.py migration_doctor --all-local --check --store - Safe restore: manage.py migration_doctor --app --restore --backup --db-user --db-pass

--yes-i-know - Auto-fix 1050: manage.py migration_doctor --autofix-1050 [--backup --db-user --db-pass

--backup-continue-on-fail] What it does ------------ • Audit migration files vs django_migrations, and models vs DB tables • Safely restore one broken app (classic "table already exists" / history out-of-sync) • Cleanup multiple apps in bulk (purge migration files & history, rebuild a clean plan) Notes ----- - Messages are mostly in French to match the rest of the project. - This version includes: - mysqldump credentials flags (db-user, db-pass, db-host, db-port) - --backup-continue-on-fail to proceed even if backup fails (for CI/cron) - --autofix-1050 mode that runs migrate and, on MySQL error 1050, auto-restores the owning app and re-runs migrate """ from django.core.management.base import BaseCommand, CommandError from django.core.management import call_command from django.db import connection, transaction from django.db.utils import OperationalError from django.apps import apps from django.conf import settings from pathlib import Path import datetime, subprocess, shlex, sys, shutil, re # --- imports en haut --- import io from contextlib import redirect_stdout, redirect_stderr # (optionnel) stockage du rapport try: from doctor.models import DoctorRun, MigrationSnapshot, TableSnapshot, Finding HAVE_DOCTOR = True except Exception: HAVE_DOCTOR = False class Command(BaseCommand): help = "Contrôle / Intégrité / Restauration et Cleanup des migrations." def add_arguments(self, parser): # Sélection des apps parser.add_argument("--app", help="Label d’app (ex: translation)") parser.add_argument("--apps", help="Liste d'apps séparées par des virgules (ex: accounts,agenda,translation)") parser.add_argument("--all-local", action="store_true", help="Cibler toutes les apps locales (hors site-packages / django.contrib)") # Modes parser.add_argument("--check", action="store_true", help="Audit uniquement (ne modifie rien)") parser.add_argument("--plan", action="store_true", help="Affiche le plan (ne modifie rien)") parser.add_argument("--restore", action="store_true", help="Recrée un plan propre pour une app et réconcilie") parser.add_argument("--cleanup", action="store_true", help="Nettoyage total des migrations pour les apps visées, puis régénération") parser.add_argument("--autofix-1050", dest="autofix_1050", action="store_true", help="Exécute migrate; si MySQL 1050 survient, restaure automatiquement l'app propriétaire et relance migrate") # Sécurité / confort parser.add_argument("--force", action="store_true", help="Forcer l’action même si l’audit détecte des écarts (dangereux)") parser.add_argument("--backup", action="store_true", help="Dump MariaDB avant opérations destructrices") parser.add_argument("--backup-continue-on-fail", dest="backup_continue_on_fail", action="store_true", help="Ne pas arrêter le process si mysqldump échoue") parser.add_argument("--db", default=None, help="Nom de la DB pour mysqldump (sinon settings)") parser.add_argument("--db-user", dest="db_user", default=None, help="Utilisateur DB pour mysqldump") parser.add_argument("--db-pass", dest="db_pass", default=None, help="Mot de passe DB pour mysqldump") parser.add_argument("--db-host", dest="db_host", default=None, help="Hôte DB pour mysqldump") parser.add_argument("--db-port", dest="db_port", default=None, help="Port DB pour mysqldump") parser.add_argument("--mysqldump", default="mysqldump", help="Binaire mysqldump") parser.add_argument("--store", action="store_true", help="Stocker le rapport dans l’app 'doctor'") parser.add_argument("--yes-i-know", dest="yes_i_know", action="store_true", help="Confirmation explicite (obligatoire pour --restore/--cleanup)") parser.add_argument("--unsafe-fake", dest="unsafe_fake", action="store_true", help="Utiliser migrate --fake (NON recommandé). Par défaut: --fake-initial") parser.add_argument( "--churn-check", action="store_true", help="Détecte un 'makemigrations infini' (modèles non déterministes) via makemigrations --dry-run --check x2" ) # -------------------- utils: apps ciblées -------------------- def _resolve_target_apps(self, opts): labels = set() if opts.get("app"): labels.add(opts["app"]) if opts.get("apps"): labels.update([a.strip() for a in opts["apps"].split(",") if a.strip()]) if opts.get("all_local"): base = Path(getattr(settings, "BASE_DIR", Path.cwd())) for conf in apps.get_app_configs(): # filtre « local » : pas django.contrib, pas site-packages path = Path(conf.path) if conf.label.startswith("django_") or conf.name.startswith("django.contrib"): continue if "site-packages" in str(path): continue labels.add(conf.label) if not labels: raise CommandError("Spécifie --app, --apps, ou --all-local.") # validate resolved = [] for label in sorted(labels): try: conf = apps.get_app_config(label) except LookupError: raise CommandError(f"App inconnue: {label}") resolved.append(conf) return resolved # -------------------- utils: fichiers/DB -------------------- def _list_migration_files(self, app_config): mig_dir = Path(app_config.path) / "migrations" if not mig_dir.exists(): return [] return sorted([p.name for p in mig_dir.glob("[0-9][0-9][0-9][0-9]_*.py")]) def _list_db_migrations(self, app_label): with connection.cursor() as c: c.execute("SELECT name FROM django_migrations WHERE app=%s ORDER BY name", [app_label]) return [row[0] for row in c.fetchall()] def _list_db_tables(self): return set(connection.introspection.table_names()) def _app_model_tables(self, app_config): tables = set() for model in app_config.get_models(): if not model._meta.managed: continue tables.add(model._meta.db_table) return tables def _ensure_migrations_pkg(self, app_config): mig_dir = Path(app_config.path) / "migrations" mig_dir.mkdir(parents=True, exist_ok=True) init_py = mig_dir / "__init__.py" if not init_py.exists(): init_py.write_text("# auto-created by migration_doctor\n", encoding="utf-8") return mig_dir def _rm_app_migration_files(self, app_config): mig_dir = self._ensure_migrations_pkg(app_config) # supprime tous les fichiers de migration (garde __init__.py) for p in mig_dir.glob("[0-9][0-9][0-9][0-9]_*.py"): p.unlink(missing_ok=True) # pyc & __pycache__ for p in mig_dir.glob("[0-9][0-9][0-9][0-9]_*.pyc"): p.unlink(missing_ok=True) cache = mig_dir / "__pycache__" if cache.exists(): shutil.rmtree(cache, ignore_errors=True) def _delete_django_migration_rows(self, app_label): with connection.cursor() as c: c.execute("DELETE FROM django_migrations WHERE app=%s", [app_label]) # ---- mysqldump with credentials (from flags or settings) ---- def _backup_db(self, opts): cfg = connection.settings_dict dbname = opts.get("db") or cfg.get("NAME") user = opts.get("db_user") or cfg.get("USER") or "" password = opts.get("db_pass") or cfg.get("PASSWORD") or "" host = opts.get("db_host") or cfg.get("HOST") or "" port = opts.get("db_port") or (cfg.get("PORT") and str(cfg.get("PORT")) or "") ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") out = Path(f"./backup_{dbname}_{ts}.sql").resolve() parts = [opts["mysqldump"]] if host: parts += ["-h", host] if port: parts += ["-P", str(port)] if user: parts += ["-u", user] if password: parts += [f"--password={password}"] parts += [dbname] cmd = " ".join(shlex.quote(p) for p in parts) + f" > {shlex.quote(str(out))}" self.stdout.write(self.style.WARNING(f"[BACKUP] {cmd}")) rc = subprocess.call(cmd, shell=True) if rc != 0: raise CommandError("mysqldump a échoué") return out # -------------------- AUDIT (retourne dict) -------------------- def _audit_one(self, app_config, verbose=True, store=False, mode="check"): label = app_config.label files = self._list_migration_files(app_config) db_migs = self._list_db_migrations(label) db_tables = self._list_db_tables() model_tables = self._app_model_tables(app_config) missing_files = [m for m in db_migs if f"{m}.py" not in files] unapplied_files = [f[:-3] for f in files if f[:-3] not in db_migs] absent_in_db = sorted([t for t in model_tables if t not in db_tables]) orphans_in_db = sorted([t for t in db_tables if t.startswith(label + "_") and t not in model_tables]) if verbose: self.stdout.write(self.style.NOTICE(f"\n=== [{label}] MIGRATIONS ===")) self.stdout.write(self.style.NOTICE(f"Fichiers: {len(files)}")) for f in files: self.stdout.write(f" - {f}") self.stdout.write(self.style.NOTICE(f"Historique DB: {len(db_migs)}")) for m in db_migs: self.stdout.write(f" - {m}") if missing_files: self.stdout.write(self.style.WARNING(f"⚠ Dans DB mais PAS dans les fichiers: {missing_files}")) if unapplied_files: self.stdout.write(self.style.WARNING(f"⚠ Fichiers présents mais PAS appliqués en DB: {unapplied_files}")) if not missing_files and not unapplied_files: self.stdout.write(self.style.SUCCESS("✔ Fichiers et historique DB en phase.")) self.stdout.write(self.style.NOTICE(f"\n=== [{label}] TABLES ===")) self.stdout.write(self.style.NOTICE("Tables attendues (modèles):")) for t in sorted(model_tables): self.stdout.write(f" - {t}") self.stdout.write(self.style.NOTICE("Tables présentes en DB (MariaDB):")) for t in sorted(db_tables): if t.startswith(label + "_"): self.stdout.write(f" - {t}") if absent_in_db: self.stdout.write(self.style.WARNING(f"⚠ Déclarées dans les modèles MAIS absentes en DB: {absent_in_db}")) if orphans_in_db: self.stdout.write(self.style.WARNING(f"⚠ Présentes en DB MAIS absentes des modèles: {orphans_in_db}")) if not absent_in_db and not orphans_in_db: self.stdout.write(self.style.SUCCESS("✔ Schéma DB et modèles Django en phase.")) result = { "label": label, "files": files, "db_migs": db_migs, "model_tables": sorted(model_tables), "db_tables": sorted(db_tables), "missing_files": missing_files, "unapplied_files": unapplied_files, "absent_in_db": absent_in_db, "orphans_in_db": orphans_in_db, "ok_files": not missing_files and not unapplied_files, "ok_schema": not absent_in_db and not orphans_in_db, } if store and HAVE_DOCTOR: with transaction.atomic(): run = DoctorRun.objects.create( app_label=label, mode=mode, ok_files=result["ok_files"], ok_schema=result["ok_schema"], message="Alignement OK" if result["ok_files"] and result["ok_schema"] else "Écarts détectés", extra={"counts": { "files": len(files), "db_migs": len(db_migs), "model_tables": len(model_tables), "db_tables": len(db_tables), "missing_files": len(missing_files), "unapplied_files": len(unapplied_files), "absent_in_db": len(absent_in_db), "orphans_in_db": len(orphans_in_db), }}, ) MigrationSnapshot.objects.bulk_create( [MigrationSnapshot(run=run, source="files", name=f[:-3]) for f in files] + [MigrationSnapshot(run=run, source="db", name=m) for m in db_migs] ) TableSnapshot.objects.bulk_create( [TableSnapshot(run=run, source="model", table_name=t) for t in result["model_tables"]] + [TableSnapshot(run=run, source="db", table_name=t) for t in result["db_tables"] if t.startswith(label + "_")] ) Finding.objects.bulk_create( [Finding(run=run, severity="warning", kind="missing_file", label=m) for m in missing_files] + [Finding(run=run, severity="warning", kind="unapplied_file", label=u) for u in unapplied_files] + [Finding(run=run, severity="critical", kind="absent_in_db", label=t) for t in absent_in_db] + [Finding(run=run, severity="warning", kind="orphan_in_db", label=t) for t in orphans_in_db] ) result["run_id"] = run.id return result # -------------------- helpers: 1050 parsing & mapping -------------------- _RE_1050 = re.compile(r"table\s+[`']?([a-zA-Z0-9_\.]+)[`']?\s+(?:already\s+exists|existe)", re.IGNORECASE) def _makemigrations_global_dry_check(self): buf = io.StringIO() try: with redirect_stdout(buf), redirect_stderr(buf): call_command("makemigrations", dry_run=True, check=True, verbosity=1) out = buf.getvalue() has_changes = ("No changes detected" not in out) and (out.strip() != "") return has_changes, out except SystemExit: return True, buf.getvalue() def _extract_tables_from_1050(self, msg: str): tables = set() for m in self._RE_1050.finditer(msg): t = m.group(1) # drop schema prefix if any (db.table) t = t.split(".")[-1] tables.add(t) return sorted(tables) def _table_to_app(self, tbl: str): for m in apps.get_models(): if m._meta.db_table == tbl: return m._meta.app_label # heuristic fallback: prefix until first underscore if "_" in tbl: return tbl.split("_", 1)[0] return None # -------------------- handle -------------------- def handle(self, *args, **opts): # ==== AUTO-FIX 1050 MODE (no app selector required) ==== if opts.get("autofix_1050"): try: call_command("migrate") self.stdout.write(self.style.SUCCESS("migrate OK")) return except OperationalError as e: msg = str(e) if "1050" not in msg and "already exists" not in msg and "existe" not in msg: raise tables = self._extract_tables_from_1050(msg) if not tables: raise self.stdout.write(self.style.WARNING(f"MySQL 1050 détecté. Tables: {tables}")) # Backup optionnel if opts.get("backup"): try: self._backup_db(opts) except CommandError: if opts.get("backup_continue_on_fail"): self.stdout.write(self.style.WARNING("Backup failed; continuing as requested.")) else: raise for tbl in tables: app_label = self._table_to_app(tbl) if not app_label: raise CommandError(f"Impossible de déduire l’app pour la table {tbl}") self.stdout.write(self.style.WARNING(f"[AUTO-RESTORE] App visée: {app_label}")) call_command("makemigrations", app_label) call_command("migrate", app_label, "0", fake=True) call_command("migrate", app_label, fake_initial=True) call_command("migrate") self.stdout.write(self.style.SUCCESS("Auto-fix 1050 terminé (migrate OK).")) return # À partir d’ici, les modes nécessitent un périmètre d’apps targets = self._resolve_target_apps(opts) # --- dans handle(), juste après resolve_target_apps() ou avant les autres modes --- if opts.get("churn_check"): targets = self._resolve_target_apps(opts) any_issue = False global_changes, global_out = self._makemigrations_global_dry_check() if global_changes: self.stdout.write(self.style.WARNING("⚠ GLOBAL makemigrations détecte des changements (plan global).")) # Si tu veux: afficher seulement les blocs 'Migrations for ...' self.stdout.write(global_out.strip() or "(vide)") for conf in targets: self.stdout.write(self.style.NOTICE(f"\n=== CHURN CHECK [{conf.label}] ===")) c1, out1 = self._makemigrations_dry_check(conf.label) c2, out2 = self._makemigrations_dry_check(conf.label) if out1 != out2: any_issue = True self.stdout.write(self.style.ERROR("❌ NON-DÉTERMINISME détecté : l'output diffère entre 2 passes.")) self._print_churn_hints() # option: afficher un extrait self.stdout.write(self.style.WARNING("---- pass #1 ----")) self.stdout.write(out1.strip() or "(vide)") self.stdout.write(self.style.WARNING("---- pass #2 ----")) self.stdout.write(out2.strip() or "(vide)") continue if c1: any_issue = True self.stdout.write(self.style.WARNING("⚠ Changements détectés (mais déterministes).")) self.stdout.write(out1.strip() or "(vide)") self.stdout.write(self.style.NOTICE( "=> Ensuite lance un audit: migration_doctor --app --check\n" " et si DB/tables déjà là mais historique out-of-sync: --restore --yes-i-know (avec --backup)." )) else: self.stdout.write(self.style.SUCCESS("✔ Aucun changement détecté par makemigrations (--check).")) if any_issue: raise CommandError("CHURN CHECK: problèmes détectés (voir logs).") self.stdout.write(self.style.SUCCESS("\nCHURN CHECK terminé : rien à signaler.")) return # ==== MODE PLAN (affiche le plan d'une seule app si --app fourni) ==== if opts["plan"]: if len(targets) != 1: self.stdout.write(self.style.WARNING("Pour --plan, spécifie une seule app avec --app.")) label = targets[0].label self.stdout.write(self.style.SUCCESS(f"\n=== PLAN DE MIGRATION [{label}] ===")) call_command("migrate", label, plan=True) return # ==== AUDIT SIMPLE ==== if opts["check"] and not opts["restore"] and not opts["cleanup"]: for conf in targets: self._audit_one(conf, verbose=True, store=opts.get("store"), mode="check") self.stdout.write(self.style.SUCCESS("\nCHECK terminé (aucune modification).")) return # ==== RESTORE (monopérimètre recommandé) ==== if opts["restore"]: if len(targets) != 1: raise CommandError("RESTORE s’utilise sur UNE app (utilise --app).") if not opts["yes_i_know"]: raise CommandError("Ajoute --yes-i-know pour confirmer.") conf = targets[0] # 1) audit préalable self._audit_one(conf, verbose=True, store=opts.get("store"), mode="check") # 2) backup optionnel if opts["backup"]: try: self._backup_db(opts) except CommandError: if opts.get("backup_continue_on_fail"): self.stdout.write(self.style.WARNING("Backup failed; continuing as requested.")) else: raise # 3) makemigrations (recrée le plan) self.stdout.write(self.style.WARNING(f"makemigrations {conf.label} ...")) call_command("makemigrations", conf.label) # 4) vider historique app (sans toucher aux tables) self.stdout.write(self.style.WARNING(f"migrate {conf.label} zero --fake ...")) call_command("migrate", conf.label, "0", fake=True) # 5) ré-appliquer en fake-initial self.stdout.write(self.style.WARNING(f"migrate {conf.label} --fake-initial ...")) call_command("migrate", conf.label, fake_initial=True) # 6) plan final + audit stocké self.stdout.write(self.style.SUCCESS("\nPlan final:")) call_command("migrate", conf.label, plan=True) self._audit_one(conf, verbose=True, store=opts.get("store"), mode="restore") self.stdout.write(self.style.SUCCESS(f"[OK] Restauration terminée pour {conf.label}.")) return # ==== CLEANUP GLOBAL (plusieurs apps) ==== if opts["cleanup"]: if not opts["yes_i_know"]: raise CommandError("Ajoute --yes-i-know pour confirmer le CLEANUP.") # 0) audit global préalable overall_ok = True for conf in targets: res = self._audit_one(conf, verbose=True, store=opts.get("store"), mode="check") if not (res["ok_files"] and res["ok_schema"]): overall_ok = False if not overall_ok and not opts["force"]: raise CommandError("Des écarts existent (modèles ↔ DB ou fichiers ↔ historique). Abandon (utilise --force si tu es sûr).") # 1) backup (avant destructions) if opts["backup"]: try: self._backup_db(opts) except CommandError: if opts.get("backup_continue_on_fail"): self.stdout.write(self.style.WARNING("Backup failed; continuing as requested.")) else: raise # 2) purge historique & fichiers de migrations (apps visées) for conf in targets: self.stdout.write(self.style.WARNING(f"[CLEANUP] Purge django_migrations pour {conf.label}")) self._delete_django_migration_rows(conf.label) self.stdout.write(self.style.WARNING(f"[CLEANUP] Suppression des fichiers de migrations pour {conf.label}")) self._rm_app_migration_files(conf) # 3) makemigrations labels = [c.label for c in targets] self.stdout.write(self.style.WARNING(f"makemigrations {' '.join(labels)} ...")) if labels: call_command("makemigrations", *labels) else: call_command("makemigrations") # 4) migrate (sécurisé) if opts.get("unsafe_fake"): self.stdout.write(self.style.WARNING("migrate --fake (NON RECOMMANDÉ) ...")) call_command("migrate", fake=True) else: self.stdout.write(self.style.WARNING("migrate --fake-initial (recommandé) ...")) call_command("migrate", fake_initial=True) # 5) audit final self.stdout.write(self.style.SUCCESS("\n=== AUDIT FINAL ===")) for conf in targets: self._audit_one(conf, verbose=True, store=opts.get("store"), mode="cleanup") self.stdout.write(self.style.SUCCESS("\n[OK] CLEANUP terminé.")) return # Si aucun mode explicite : raise CommandError("Rien à faire : passe --check, --restore, --cleanup ou --autofix-1050.") # --- helpers dans la classe Command --- def _makemigrations_dry_check(self, app_label: str): """ Lance 'makemigrations --dry-run --check' et capture stdout/stderr. Retourne (has_changes: bool, output: str) """ buf = io.StringIO() try: with redirect_stdout(buf), redirect_stderr(buf): # verbosity=1 pour avoir les infos sans être trop verbeux call_command("makemigrations", app_label, dry_run=True, check=True, verbosity=1) out = buf.getvalue() # si check passe, Django ne lève pas, mais l'output peut être vide ou "No changes detected" has_changes = ("No changes detected" not in out) and (out.strip() != "") return has_changes, out except SystemExit as e: # Django utilise SystemExit(code=1) quand --check détecte des changements out = buf.getvalue() return True, out def _print_churn_hints(self): self.stdout.write(self.style.WARNING( "Pistes churn (makemigrations infini) :\n" " - choices/indexes/constraints/permissions construits via set()/dict() non triés\n" " - Meta.ordering/permissions/indexes/constraints dynamiques (ordre instable)\n" " - noms d'index/constraint dépendants de valeurs variables\n" " -> Fix: utiliser list/tuple + sorted(), figer l'ordre, éviter génération dynamique." ))