your_project/
  accounts/                         # ou tools/, ou autre app utilitaire
    management/
      __init__.py
      commands/
        __init__.py
        django_migration_doctor_2025.py   # <= notre script

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Django Migration Doctor 2025
Outil multi-usages pour auditer et réparer les migrations "cassées".
"""

import os
import re
import glob
import subprocess
from datetime import datetime

from django.apps import apps
from django.conf import settings
from django.core.management.base import BaseCommand, CommandError
from django.core.management import call_command
from django.db import DEFAULT_DB_ALIAS, connections
from django.db.migrations.recorder import MigrationRecorder


class Command(BaseCommand):
    help = (
        "Django Migration Doctor 2025 : audit et réparation avancée des migrations.\n\n"
        "Modes principaux :\n"
        "  --check          Audit des apps (migrations <> tables)\n"
        "  --plan           Affiche le plan de migration\n"
        "  --autofix-1050   Tente de réparer automatiquement les erreurs MySQL 1050\n"
        "  --restore        Restore ciblé d'une app (reset migrations pour 1 app)\n"
        "  --cleanup        Cleanup global multi-apps (reset migrations)\n\n"
        "Sélection des apps :\n"
        "  --app CV                Une seule app\n"
        "  --apps CV,accounts,...  Plusieurs apps\n"
        "  --all-local             Toutes les apps locales (hors django.contrib)\n"
    )

    def add_arguments(self, parser):
        # Sélection d'apps
        parser.add_argument("--app", help="App unique à traiter (ex: CV)")
        parser.add_argument(
            "--apps",
            help="Liste d'apps séparées par des virgules (ex: accounts,portfolio,CV)",
        )
        parser.add_argument(
            "--all-local",
            action="store_true",
            help="Cibler automatiquement toutes les apps locales du projet.",
        )

        parser.add_argument(
            "--database",
            default=DEFAULT_DB_ALIAS,
            help="Alias de base Django (par défaut: 'default').",
        )

        # Modes
        parser.add_argument(
            "--check",
            action="store_true",
            help="Audit : compare fichiers de migrations / django_migrations / tables DB.",
        )
        parser.add_argument(
            "--plan",
            action="store_true",
            help="Affiche le plan de migrations pour les apps cibles.",
        )
        parser.add_argument(
            "--autofix-1050",
            action="store_true",
            help="Tente de réparer automatiquement les erreurs MySQL 1050 'table already exists'.",
        )
        parser.add_argument(
            "--restore",
            action="store_true",
            help="Restore ciblé d'une app (reset migrations d'UNE seule app).",
        )
        parser.add_argument(
            "--cleanup",
            action="store_true",
            help="Cleanup global multi-apps (reset migrations pour plusieurs apps).",
        )

        # Sécurité / confirmations
        parser.add_argument(
            "--dry-run",
            action="store_true",
            help="Afficher le plan d'action sans exécuter les opérations destructives.",
        )
        parser.add_argument(
            "--yes-i-know",
            action="store_true",
            help="Confirmer explicitement les opérations destructives (prod, CI, etc.).",
        )
        parser.add_argument(
            "--force",
            action="store_true",
            help="Ne pas poser de questions interactives.",
        )

        # Backup DB (MySQL/MariaDB)
        parser.add_argument(
            "--backup",
            action="store_true",
            help="Effectuer un backup SQL avant les opérations destructives (mysqldump).",
        )
        parser.add_argument("--db", help="Nom de la base (override settings).")
        parser.add_argument("--db-user", help="Utilisateur DB (override settings).")
        parser.add_argument("--db-pass", help="Mot de passe DB (override settings).")
        parser.add_argument("--db-host", help="Host DB (override settings).")
        parser.add_argument("--db-port", help="Port DB (override settings).")
        parser.add_argument(
            "--backup-continue-on-fail",
            action="store_true",
            help="Continuer même si le backup mysqldump échoue.",
        )

    # --- helpers généraux -------------------------------------------------

    def _resolve_target_apps(self, options):
        """Détermine la liste d'AppConfig ciblées."""
        labels = []

        if options["app"]:
            labels.append(options["app"])
        if options["apps"]:
            labels.extend([x.strip() for x in options["apps"].split(",") if x.strip()])
        if options["all_local"]:
            for cfg in apps.get_app_configs():
                if cfg.name.startswith("django.") or "site-packages" in (cfg.path or ""):
                    continue
                labels.append(cfg.label)

        if not labels:
            raise CommandError(
                "Aucune app cible. Utilise --app, --apps ou --all-local."
            )

        resolved = []
        seen = set()
        for label in labels:
            try:
                cfg = apps.get_app_config(label)
            except LookupError:
                # tenter lower / upper
                tried = [label, label.lower(), label.upper()]
                cfg = None
                for cand in tried:
                    try:
                        cfg = apps.get_app_config(cand)
                        break
                    except LookupError:
                        continue
                if cfg is None:
                    raise CommandError(f"App '{label}' introuvable dans INSTALLED_APPS.")
            if cfg.label not in seen:
                resolved.append(cfg)
                seen.add(cfg.label)

        return resolved

    def _confirm_destructive(self, msg, options):
        if options["dry_run"]:
            self.stdout.write(self.style.WARNING(f"[DRY-RUN] {msg}"))
            return False

        if options["force"] or options["yes_i_know"]:
            self.stdout.write(self.style.WARNING(msg + " (confirmation forcée)."))
            return True

        answer = input(f"{msg} Continuer ? [yes/no] ").strip().lower()
        return answer in ("yes", "y")

    def _backup_db(self, options):
        db_settings = settings.DATABASES[options["database"]]
        engine = db_settings["ENGINE"]

        if "mysql" not in engine:
            self.stdout.write(
                self.style.WARNING(
                    "Backup automatique prévu pour MySQL/MariaDB uniquement. Skip."
                )
            )
            return

        name = options["db"] or db_settings["NAME"]
        user = options["db_user"] or db_settings.get("USER") or "root"
        password = options["db_pass"] or db_settings.get("PASSWORD") or ""
        host = options["db_host"] or db_settings.get("HOST") or "127.0.0.1"
        port = options["db_port"] or db_settings.get("PORT") or "3306"

        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"backup_{name}_{timestamp}.sql"
        cmd = [
            "mysqldump",
            f"-h{host}",
            f"-P{port}",
            f"-u{user}",
            f"-p{password}",
            name,
        ]

        self.stdout.write(self.style.NOTICE(f"Backup DB dans {filename}..."))
        with open(filename, "w", encoding="utf-8") as f:
            rc = subprocess.call(cmd, stdout=f)

        if rc != 0:
            msg = f"mysqldump a échoué (code {rc})"
            if options["backup_continue_on_fail"]:
                self.stdout.write(self.style.WARNING(msg + " mais on continue."))
            else:
                raise CommandError(msg)
        else:
            self.stdout.write(self.style.SUCCESS(f"Backup terminé : {filename}"))

    # --- audit de base ----------------------------------------------------

    def _audit_app(self, app_config, connection):
        """Retourne un dict avec les infos d'audit pour une app."""
        label = app_config.label
        migrations_dir = os.path.join(app_config.path, "migrations")

        # fichiers de migrations
        file_names = []
        if os.path.isdir(migrations_dir):
            pattern = os.path.join(migrations_dir, "[0-9][0-9][0-9][0-9]_*.py")
            for path in glob.glob(pattern):
                file_names.append(os.path.basename(path).replace(".py", ""))

        recorder = MigrationRecorder(connection)
        db_migrations = list(
            recorder.migration_qs.filter(app=label).values_list("name", flat=True)
        )

        # tables liées aux modèles de l'app
        model_tables = set()
        for model in app_config.get_models():
            model_tables.add(model._meta.db_table)

        # tables DB réelles
        all_tables = set(connection.introspection.table_names())
        prefix = label.lower() + "_"
        db_tables_for_app = {t for t in all_tables if t.startswith(prefix)}

        missing_files = [m for m in db_migrations if m not in file_names]
        unapplied_files = [f for f in file_names if f not in db_migrations]
        absent_tables = [t for t in model_tables if t not in all_tables]
        orphan_tables = [t for t in db_tables_for_app if t not in model_tables]

        return {
            "label": label,
            "file_names": sorted(file_names),
            "db_migrations": sorted(db_migrations),
            "missing_files": sorted(missing_files),
            "unapplied_files": sorted(unapplied_files),
            "absent_tables": sorted(absent_tables),
            "orphan_tables": sorted(orphan_tables),
        }

    # --- modes ------------------------------------------------------------

    def handle(self, *args, **options):
        database = options["database"]
        connection = connections[database]

        modes = [
            options["check"],
            options["plan"],
            options["autofix_1050"],
            options["restore"],
            options["cleanup"],
        ]
        if sum(1 for m in modes if m) == 0:
            raise CommandError(
                "Tu dois choisir au moins un mode : "
                "--check / --plan / --autofix-1050 / --restore / --cleanup."
            )

        apps_list = self._resolve_target_apps(options)

        if options["restore"] and len(apps_list) != 1:
            raise CommandError(
                "--restore ne peut s'appliquer qu'à UNE seule app (utilise --app)."
            )

        if options["backup"] and not options["dry_run"]:
            self._backup_db(options)

        # 1) MODE CHECK
        if options["check"]:
            self.stdout.write(self.style.MIGRATE_HEADING("Audit des apps ciblées..."))
            for cfg in apps_list:
                audit = self._audit_app(cfg, connection)
                self.stdout.write(self.style.NOTICE(f"[{audit['label']}]"))
                self.stdout.write(f"  Migrations fichiers : {len(audit['file_names'])}")
                self.stdout.write(f"  Migrations DB      : {len(audit['db_migrations'])}")
                if audit["missing_files"]:
                    self.stdout.write("  - Fichiers manquants pour DB :")
                    for m in audit["missing_files"]:
                        self.stdout.write(f"      * {m}")
                if audit["unapplied_files"]:
                    self.stdout.write("  - Fichiers non appliqués en DB :")
                    for m in audit["unapplied_files"]:
                        self.stdout.write(f"      * {m}")
                if audit["absent_tables"]:
                    self.stdout.write("  - Tables attendues absentes :")
                    for t in audit["absent_tables"]:
                        self.stdout.write(f"      * {t}")
                if audit["orphan_tables"]:
                    self.stdout.write("  - Tables orphelines (DB sans modèle) :")
                    for t in audit["orphan_tables"]:
                        self.stdout.write(f"      * {t}")
                if (
                    not audit["missing_files"]
                    and not audit["unapplied_files"]
                    and not audit["absent_tables"]
                    and not audit["orphan_tables"]
                ):
                    self.stdout.write("  OK : rien à signaler.")
            self.stdout.write("")

        # 2) MODE PLAN
        if options["plan"]:
            self.stdout.write(self.style.MIGRATE_HEADING("Plan de migrations :"))
            # si une seule app, on passe l'app, sinon on laisse Django faire global
            if len(apps_list) == 1:
                call_command("migrate", apps_list[0].label, database=database, plan=True)
            else:
                call_command("migrate", database=database, plan=True)
            self.stdout.write("")

        # 3) MODE AUTOFIX-1050
        if options["autofix_1050"]:
            self.stdout.write(self.style.MIGRATE_HEADING("Auto-fix MySQL 1050..."))
            try:
                call_command("migrate", database=database)
                self.stdout.write(
                    self.style.SUCCESS("migrate terminé, aucune erreur 1050 détectée.")
                )
            except Exception as exc:
                msg = str(exc)
                if "1050" not in msg and "already exists" not in msg and "existe déjà" not in msg:
                    raise

                self.stdout.write(
                    self.style.WARNING("Erreur 1050 détectée : tentative d'auto-fix...")
                )
                tables = self._extract_tables_from_1050(msg)
                self.stdout.write(f"Tables incriminées : {', '.join(tables) or 'inconnues'}")

                affected_apps = set()
                for t in tables:
                    for cfg in apps.get_app_configs():
                        for model in cfg.get_models():
                            if model._meta.db_table == t:
                                affected_apps.add(cfg.label)

                for label in sorted(affected_apps):
                    self.stdout.write(
                        self.style.NOTICE(f"Réconciliation migrations pour app {label}...")
                    )
                    if not options["dry_run"]:
                        call_command("makemigrations", label)
                        call_command("migrate", label, "0", database=database, fake=True)
                        call_command(
                            "migrate",
                            label,
                            database=database,
                            fake_initial=True,
                        )

                if not options["dry_run"]:
                    call_command("migrate", database=database)
                    self.stdout.write(
                        self.style.SUCCESS(
                            "Auto-fix 1050 terminé, migrate repassé avec succès."
                        )
                    )

        # 4) MODE RESTORE (app unique)
        if options["restore"]:
            cfg = apps_list[0]
            label = cfg.label
            self.stdout.write(
                self.style.MIGRATE_HEADING(f"Restore ciblé pour app '{label}'...")
            )

            if not self._confirm_destructive(
                f"Restore migrations pour '{label}' (reset historique + fake-initial).",
                options,
            ):
                self.stdout.write(self.style.ERROR("Restore annulé."))
            else:
                if not options["dry_run"]:
                    call_command("makemigrations", label)
                    call_command("migrate", label, "0", database=database, fake=True)
                    call_command(
                        "migrate", label, database=database, fake_initial=True
                    )
                    self.stdout.write(
                        self.style.SUCCESS(
                            f"Restore terminé pour l'app '{label}'."
                        )
                    )

        # 5) MODE CLEANUP (multi-apps)
        if options["cleanup"]:
            labels = [cfg.label for cfg in apps_list]
            self.stdout.write(
                self.style.MIGRATE_HEADING(
                    f"Cleanup global des migrations pour : {', '.join(labels)}"
                )
            )
            if not self._confirm_destructive(
                "Cleanup global : suppression fichiers de migrations + rows django_migrations.",
                options,
            ):
                self.stdout.write(self.style.ERROR("Cleanup annulé."))
            else:
                recorder = MigrationRecorder(connection)
                if not options["dry_run"]:
                    for cfg in apps_list:
                        label = cfg.label
                        # supprimer lignes django_migrations
                        deleted = recorder.migration_qs.filter(app=label).delete()
                        self.stdout.write(
                            f"[{label}] django_migrations : {deleted[0]} ligne(s) supprimée(s)."
                        )
                        # supprimer fichiers de migrations
                        migrations_dir = os.path.join(cfg.path, "migrations")
                        if os.path.isdir(migrations_dir):
                            pattern = os.path.join(
                                migrations_dir, "[0-9][0-9][0-9][0-9]_*.py"
                            )
                            for path in glob.glob(pattern):
                                self.stdout.write(
                                    f"[{label}] suppression fichier {os.path.basename(path)}"
                                )
                                os.remove(path)

                    # recréer et rejouer les migrations
                    call_command("makemigrations", *labels)
                    call_command("migrate", database=database, fake_initial=True)
                    self.stdout.write(
                        self.style.SUCCESS("Cleanup global terminé (fake-initial).")
                    )

    # ------------------------------------------------------------------

    def _extract_tables_from_1050(self, message):
        """Essaie d'extraire les noms de tables depuis un message d'erreur 1050."""
        tables = set()
        pattern = r"Table '([^']+)' already exists|Table '([^']+)' existe déjà"
        for m in re.finditer(pattern, message):
            for group in m.groups():
                if group:
                    tables.add(group)
        return sorted(tables)

# Audit de toutes les apps locales
python manage.py django_migration_doctor_2025 --all-local --check

# Plan global
python manage.py django_migration_doctor_2025 --all-local --plan

# Plan pour une app spécifique
python manage.py django_migration_doctor_2025 --app CV --plan

# Tente un migrate, et si 1050 est détecté,
# réconcilie automatiquement les migrations des apps concernées.
python manage.py django_migration_doctor_2025 --autofix-1050 --backup \
    --db-user root --db-pass XXXXX

# Restore des migrations pour l'app CV
python manage.py django_migration_doctor_2025 \
    --app CV --restore --backup --db-user root --db-pass XXXXX --yes-i-know

# Cleanup pour quelques apps seulement
python manage.py django_migration_doctor_2025 \
    --apps accounts,portfolio,CV \
    --cleanup --backup --db-user root --db-pass XXXXX --yes-i-know

0 3 * * * /opt/ideo-lab/venv/bin/python \
  /opt/ideo-lab/manage.py django_migration_doctor_2025 --all-local --check \
  >> /var/log/ideo-lab/migration_doctor_2025.log 2>&1

# Création de l'app (si pas encore faite)
python manage.py startapp doctor

# Ajout dans settings.py
INSTALLED_APPS = [
    # ...
    "doctor",
]

from django.db import models
from django.utils import timezone

class DoctorRun(models.Model):
    """
    Un exécutable d'audit/restauration (migration_doctor).
    """
    created_at   = models.DateTimeField(default=timezone.now, db_index=True)
    app_label    = models.CharField(max_length=80, db_index=True)
    mode         = models.CharField(max_length=20, choices=[
        ("check", "check"), ("plan", "plan"), ("restore", "restore")
    ])
    ok_files     = models.BooleanField(default=False)
    ok_schema    = models.BooleanField(default=False)
    message      = models.TextField(blank=True, default="")   # résumé humain
    extra        = models.JSONField(default=dict, blank=True)  # stats diverses

    class Meta:
        indexes = [models.Index(fields=["created_at", "app_label", "mode"])]

    def __str__(self):
        return f"{self.created_at:%Y-%m-%d %H:%M} [{self.app_label}] {self.mode}"

class MigrationSnapshot(models.Model):
    """
    Instantané des migrations: côté fichiers ET côté DB (django_migrations).
    """
    run          = models.ForeignKey(DoctorRun, on_delete=models.CASCADE, related_name="migration_snaps")
    source       = models.CharField(max_length=10, choices=[("files","files"),("db","db")])  # files|db
    name         = models.CharField(max_length=150)  # ex: 0001_initial
    present      = models.BooleanField(default=True) # pour uniformité
    details      = models.JSONField(default=dict, blank=True)

    class Meta:
        indexes = [models.Index(fields=["run", "source", "name"])]

class TableSnapshot(models.Model):
    """
    Instantané des tables: côté 'model' (attendues) et côté 'db' (réelles).
    """
    run          = models.ForeignKey(DoctorRun, on_delete=models.CASCADE, related_name="table_snaps")
    source       = models.CharField(max_length=10, choices=[("model","model"),("db","db")])  # model|db
    table_name   = models.CharField(max_length=180, db_index=True)
    present      = models.BooleanField(default=True)
    details      = models.JSONField(default=dict, blank=True) # pk, fields, etc. (optionnel)

    class Meta:
        indexes = [models.Index(fields=["run", "source", "table_name"])]

class Finding(models.Model):
    """
    Une divergence ou information (warning/info/critical) trouvée par le doctor.
    """
    SEVERITY = [("info","info"), ("warning","warning"), ("critical","critical")]
    KIND = [
        ("missing_file", "Migration en DB mais pas de fichier"),
        ("unapplied_file", "Fichier présent mais pas appliqué"),
        ("absent_in_db", "Table attendue manquante en DB"),
        ("orphan_in_db", "Table orpheline en DB"),
        ("summary", "Résumé"),
    ]
    run          = models.ForeignKey(DoctorRun, on_delete=models.CASCADE, related_name="findings")
    severity     = models.CharField(max_length=10, choices=SEVERITY, db_index=True)
    kind         = models.CharField(maxlength=32, choices=KIND, db_index=True)
    label        = models.CharField(max_length=180)                   # nom court (ex: 0003_..., translation_key…)
    details      = models.JSONField(default=dict, blank=True)         # données brutes utiles
    created_at   = models.DateTimeField(default=timezone.now)

    class Meta:
        indexes = [models.Index(fields=["run", "severity", "kind"])]

python manage.py makemigrations doctor
python manage.py migrate doctor

from django.contrib import admin
from doctor.models import DoctorRun, MigrationSnapshot, TableSnapshot, Finding


@admin.register(DoctorRun)
class DoctorRunAdmin(admin.ModelAdmin):
    list_display = ("created_at", "app_label", "mode", "ok_files", "ok_schema", "message")
    list_filter = ("app_label", "mode", "ok_files", "ok_schema", "created_at")
    search_fields = ("app_label", "message")
    date_hierarchy = "created_at"
    ordering = ("-created_at",)


@admin.register(MigrationSnapshot)
class MigrationSnapshotAdmin(admin.ModelAdmin):
    list_display = ("run", "source", "name")
    list_filter = ("source", "run__app_label", "run__mode")
    search_fields = ("name", "run__app_label")


@admin.register(TableSnapshot)
class TableSnapshotAdmin(admin.ModelAdmin):
    list_display = ("run", "source", "table_name")
    list_filter = ("source", "run__app_label", "run__mode")
    search_fields = ("table_name", "run__app_label")


@admin.register(Finding)
class FindingAdmin(admin.ModelAdmin):
    list_display = ("run", "severity", "kind", "label")
    list_filter = ("severity", "kind", "run__app_label", "run__mode")
    search_fields = ("label", "kind", "run__app_label")

django.db.utils.OperationalError: (1050, "Table 'cv_cvprofile_primary_skills' already exists")