238 lines
11 KiB
Python
238 lines
11 KiB
Python
"""
|
|
Management command to detect, report, and safely fix duplicate and orphan road sign panels.
|
|
|
|
Usage:
|
|
# Inspection / Dry-run report:
|
|
python manage.py check_duplicate_sign_panels [--org_id 1]
|
|
|
|
# Fix duplicate orphan panels by archiving them:
|
|
python manage.py check_duplicate_sign_panels [--org_id 1] --fix
|
|
|
|
# Fix duplicate orphan panels by deleting them:
|
|
python manage.py check_duplicate_sign_panels [--org_id 1] --fix --action delete
|
|
|
|
# Also automatically attach unattached non-duplicate panels to the nearest pole (within 1m):
|
|
python manage.py check_duplicate_sign_panels [--org_id 1] --fix --attach-candidates
|
|
"""
|
|
|
|
from django.core.management.base import BaseCommand
|
|
from django.db import transaction
|
|
from django.db.models import Count, Q
|
|
|
|
from sign.models import SignPanel, SignPole, Organization
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Détecte, affiche et corrige les panneaux de signalisation en double et orphelins."
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
"--org_id",
|
|
type=int,
|
|
default=None,
|
|
help="ID de l'organisation à analyser (optionnel).",
|
|
)
|
|
parser.add_argument(
|
|
"--fix",
|
|
action="store_true",
|
|
default=False,
|
|
help="Applique les corrections en base de données. Sans cette option, la commande s'exécute en mode consultation (DRY-RUN).",
|
|
)
|
|
parser.add_argument(
|
|
"--action",
|
|
type=str,
|
|
choices=["archive", "delete"],
|
|
default="archive",
|
|
help="Action pour les doublons orphelins : 'archive' (par défaut, conserve l'historique) ou 'delete'.",
|
|
)
|
|
parser.add_argument(
|
|
"--attach-candidates",
|
|
action="store_true",
|
|
default=False,
|
|
help="Rattache automatiquement les panneaux orphelins non-doublons au poteau situé à la même position (< 1m).",
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
org_id = options.get("org_id")
|
|
fix_mode = options.get("fix", False)
|
|
action = options.get("action", "archive")
|
|
attach_candidates = options.get("attach_candidates", False)
|
|
|
|
self.stdout.write("=" * 80)
|
|
self.stdout.write(self.style.NOTICE("DIAGNOSTIC DES PANNEAUX DE SIGNALISATION (DOUBLONS ET ORPHELINS)"))
|
|
self.stdout.write("=" * 80)
|
|
|
|
if not fix_mode:
|
|
self.stdout.write(self.style.WARNING("Mode consultation (DRY-RUN) : aucune modification ne sera enregistrée.\nUtilisez --fix pour appliquer les corrections.\n"))
|
|
else:
|
|
self.stdout.write(self.style.SUCCESS(f"Mode CORRECTION actif (action doublons : {action.upper()}).\n"))
|
|
|
|
# Base querysets
|
|
panels_qs = SignPanel.objects.select_related("pole", "signpanel_type", "street", "organization")
|
|
poles_qs = SignPole.objects.select_related("street", "organization")
|
|
|
|
if org_id:
|
|
panels_qs = panels_qs.filter(organization_id=org_id)
|
|
poles_qs = poles_qs.filter(organization_id=org_id)
|
|
self.stdout.write(f"Filtrage par organisation ID : {org_id}")
|
|
|
|
total_panels = panels_qs.count()
|
|
total_poles = poles_qs.count()
|
|
self.stdout.write(f"Total poteaux analysés : {total_poles}")
|
|
self.stdout.write(f"Total panneaux analysés : {total_panels}\n")
|
|
|
|
# ---------------------------------------------------------------------
|
|
# 1. Doublons de panneaux attachés sur un même poteau
|
|
# (même pole_id, même type de panneau, statut actif)
|
|
# ---------------------------------------------------------------------
|
|
self.stdout.write("-" * 80)
|
|
self.stdout.write(self.style.NOTICE("1. RECHERCHE DES DOUBLONS ATTACHÉS AU MÊME POTEAU"))
|
|
self.stdout.write("-" * 80)
|
|
|
|
attached_dupes_query = (
|
|
panels_qs.filter(pole__isnull=False, signpanel_type__isnull=False, status="active")
|
|
.values("pole_id", "signpanel_type_id")
|
|
.annotate(count=Count("id"))
|
|
.filter(count__gt=1)
|
|
)
|
|
|
|
attached_dupe_groups = list(attached_dupes_query)
|
|
self.stdout.write(f"Groupes de doublons sur un même poteau trouvés : {len(attached_dupe_groups)}")
|
|
|
|
attached_dupes_to_fix = []
|
|
for group in attached_dupe_groups:
|
|
p_id = group["pole_id"]
|
|
type_id = group["signpanel_type_id"]
|
|
matching_panels = list(
|
|
panels_qs.filter(pole_id=p_id, signpanel_type_id=type_id, status="active").order_by("id")
|
|
)
|
|
# Keep the first (oldest), others are duplicates
|
|
kept = matching_panels[0]
|
|
dupes = matching_panels[1:]
|
|
type_code = kept.signpanel_type.code if kept.signpanel_type else "?"
|
|
self.stdout.write(
|
|
f" • Poteau ID {p_id} (Code {kept.pole.code if kept.pole else '-'}) : "
|
|
f"{len(matching_panels)} panneaux de type {type_code} (Conservé : ID {kept.id}, Doublon(s) : {[d.id for d in dupes]})"
|
|
)
|
|
attached_dupes_to_fix.extend(dupes)
|
|
|
|
# ---------------------------------------------------------------------
|
|
# 2. Panneaux orphelins (pole__isnull=True) à l'emplacement d'un poteau
|
|
# ---------------------------------------------------------------------
|
|
self.stdout.write("\n" + "-" * 80)
|
|
self.stdout.write(self.style.NOTICE("2. RECHERCHE DES PANNEAUX ORPHELINS À L'EMPLACEMENT D'UN POTEAU"))
|
|
self.stdout.write("-" * 80)
|
|
|
|
# Pre-load poles with coordinates
|
|
poles_by_coord = {}
|
|
for pole in poles_qs.iterator():
|
|
if pole.lon is not None and pole.lat is not None:
|
|
ck = (round(float(pole.lon), 4), round(float(pole.lat), 4))
|
|
if ck not in poles_by_coord:
|
|
poles_by_coord[ck] = []
|
|
poles_by_coord[ck].append(pole)
|
|
|
|
# Pre-load attached panels per pole (only active)
|
|
attached_types_by_pole = {}
|
|
for p in panels_qs.filter(pole__isnull=False, status="active").iterator():
|
|
if p.pole_id not in attached_types_by_pole:
|
|
attached_types_by_pole[p.pole_id] = set()
|
|
if p.signpanel_type_id:
|
|
attached_types_by_pole[p.pole_id].add(p.signpanel_type_id)
|
|
|
|
orphan_panels = list(
|
|
panels_qs.filter(pole__isnull=True, lon__isnull=False, lat__isnull=False, status="active")
|
|
)
|
|
self.stdout.write(f"Total panneaux orphelins actifs avec coordonnées : {len(orphan_panels)}")
|
|
|
|
orphan_exact_duplicates = []
|
|
orphan_attachable_candidates = []
|
|
|
|
for orphan in orphan_panels:
|
|
ck = (round(float(orphan.lon), 4), round(float(orphan.lat), 4))
|
|
nearby_poles = poles_by_coord.get(ck, [])
|
|
if not nearby_poles:
|
|
continue
|
|
|
|
target_pole = nearby_poles[0]
|
|
target_types = attached_types_by_pole.get(target_pole.id, set())
|
|
|
|
type_code = orphan.signpanel_type.code if orphan.signpanel_type else "?"
|
|
if orphan.signpanel_type_id and orphan.signpanel_type_id in target_types:
|
|
# This orphan panel is an exact duplicate of a panel already attached to this pole!
|
|
orphan_exact_duplicates.append((orphan, target_pole))
|
|
self.stdout.write(
|
|
f" • Doublon orphelin : Panneau ID {orphan.id} (Code {orphan.code}, Type {type_code}) "
|
|
f"partage l'emplacement du Poteau ID {target_pole.id} (Code {target_pole.code}) qui possède DÉJÀ ce type !"
|
|
)
|
|
else:
|
|
orphan_attachable_candidates.append((orphan, target_pole))
|
|
|
|
self.stdout.write(f"\nDoublons orphelins redondants détectés : {len(orphan_exact_duplicates)}")
|
|
self.stdout.write(f"Panneaux orphelins pouvant être rattachés à un poteau : {len(orphan_attachable_candidates)}")
|
|
|
|
# ---------------------------------------------------------------------
|
|
# 3. Panneaux inactifs (removed / archived)
|
|
# ---------------------------------------------------------------------
|
|
self.stdout.write("\n" + "-" * 80)
|
|
self.stdout.write(self.style.NOTICE("3. STATUTS INACTIFS"))
|
|
self.stdout.write("-" * 80)
|
|
inactive_count = panels_qs.filter(status__in=["removed", "archived"]).count()
|
|
self.stdout.write(f"Panneaux retirés ou archivés en base : {inactive_count} (exclus de la carte par défaut)")
|
|
|
|
# ---------------------------------------------------------------------
|
|
# 4. Application des corrections si --fix est demandé
|
|
# ---------------------------------------------------------------------
|
|
if not fix_mode:
|
|
self.stdout.write("\n" + "=" * 80)
|
|
self.stdout.write(self.style.WARNING("FIN DU RAPPORT. Aucune modification effectuée (lancez avec --fix pour corriger)."))
|
|
self.stdout.write("=" * 80)
|
|
return
|
|
|
|
self.stdout.write("\n" + "=" * 80)
|
|
self.stdout.write(self.style.NOTICE("APPLICATION DES CORRECTIONS EN BASE..."))
|
|
self.stdout.write("=" * 80)
|
|
|
|
with transaction.atomic():
|
|
fixed_attached = 0
|
|
fixed_orphans = 0
|
|
attached_count = 0
|
|
|
|
# Fix attached duplicates
|
|
for dupe in attached_dupes_to_fix:
|
|
if action == "delete":
|
|
dupe.delete()
|
|
else:
|
|
dupe.status = "archived"
|
|
dupe.note = (dupe.note or "") + " [Doublon archivé par check_duplicate_sign_panels]"
|
|
dupe.save(update_fields=["status", "note"])
|
|
fixed_attached += 1
|
|
|
|
# Fix orphan duplicates
|
|
for orphan, target_pole in orphan_exact_duplicates:
|
|
if action == "delete":
|
|
orphan.delete()
|
|
else:
|
|
orphan.status = "archived"
|
|
orphan.note = (orphan.note or "") + f" [Doublon orphelin du poteau {target_pole.code} archivé]"
|
|
orphan.save(update_fields=["status", "note"])
|
|
fixed_orphans += 1
|
|
|
|
# Attach candidates if requested
|
|
if attach_candidates:
|
|
for orphan, target_pole in orphan_attachable_candidates:
|
|
orphan.pole = target_pole
|
|
orphan.street = target_pole.street
|
|
orphan.geom = target_pole.geom
|
|
orphan.lon = target_pole.lon
|
|
orphan.lat = target_pole.lat
|
|
orphan.geojson = target_pole.geojson
|
|
orphan.save()
|
|
attached_count += 1
|
|
|
|
self.stdout.write(self.style.SUCCESS(
|
|
f"\n✓ Succès :\n"
|
|
f" - {fixed_attached} doublon(s) attaché(s) {action}s\n"
|
|
f" - {fixed_orphans} doublon(s) orphelin(s) {action}s\n"
|
|
f" - {attached_count} panneau(x) orphelin(s) rattaché(s) à leur poteau"
|
|
))
|