feat: add management command to import exceptional transport data for structures from CSV

loko/assets/management/commands/import_exceptional_transport_fod.py

@@ -0,0 +1,256 @@
+import os
+import csv
+import re
+from decimal import Decimal
+from django.core.management.base import BaseCommand
+from django.db import transaction
+
+from assets.models import Structure, StructureExceptionalTransport, StructureExceptionalTransportNotice
+
+
+def parse_float_safe(val):
+    if val is None or val == "":
+        return None
+    try:
+        f = float(str(val).replace(",", ".").strip())
+        return round(f, 2)
+    except (ValueError, TypeError):
+        return None
+
+
+def parse_int_safe(val):
+    if val is None or val == "":
+        return None
+    try:
+        return int(float(str(val).replace(",", ".").strip()))
+    except (ValueError, TypeError):
+        return None
+
+
+def parse_height(val):
+    if val is None or val == "" or val == 0:
+        return None
+    try:
+        f = float(str(val).replace(",", ".").strip())
+        if f <= 0 or f >= 999:
+            return None
+        if f > 100:  # e.g. 443 cm -> 4.43 m
+            return round(f / 100.0, 2)
+        if f > 15:   # e.g. 45 dm -> 4.50 m
+            return round(f / 10.0, 2)
+        return round(f, 2)
+    except (ValueError, TypeError):
+        return None
+
+
+class Command(BaseCommand):
+    help = "Importe les données de transport exceptionnel (FOD/SPF) sur les ouvrages d'art correspondants."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--file",
+            type=str,
+            default="/home/kdt/Downloads/matching_structures_transports_exceptionnels.csv",
+            help="Chemin vers le fichier CSV de matching."
+        )
+        parser.add_argument(
+            "--confidence",
+            type=str,
+            default="HAUTE",
+            choices=["HAUTE", "MOYENNE", "TOUS"],
+            help="Niveau de confiance à importer (défaut: HAUTE)."
+        )
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="Exécute sans enregistrer les modifications en base de données."
+        )
+
+    def handle(self, *args, **options):
+        csv_file = options["file"]
+        confidence_filter = options["confidence"]
+        dry_run = options["dry_run"]
+
+        if not os.path.exists(csv_file):
+            self.stderr.write(self.style.ERROR(f"Fichier introuvable : {csv_file}"))
+            return
+
+        self.stdout.write(self.style.NOTICE(f"=== Import Transport Exceptionnel FOD ==="))
+        self.stdout.write(f"Source : {csv_file}")
+        self.stdout.write(f"Filtre de confiance : {confidence_filter}")
+        self.stdout.write(f"Mode simulation (dry-run) : {dry_run}\n")
+
+        with open(csv_file, mode="r", encoding="utf-8-sig") as f:
+            reader = csv.DictReader(f, delimiter=";")
+            rows = list(reader)
+
+        # Filtrer par niveau de confiance
+        if confidence_filter == "HAUTE":
+            target_rows = [r for r in rows if r.get("confidence_level") == "HAUTE"]
+        elif confidence_filter == "MOYENNE":
+            target_rows = [r for r in rows if r.get("confidence_level") in ("HAUTE", "MOYENNE")]
+        else:
+            target_rows = [r for r in rows if r.get("matched_structure_code")]
+
+        self.stdout.write(f"Total de lignes éligibles : {len(target_rows)}")
+
+        seen_structures = set()
+        to_process = []
+        duplicates_skipped = 0
+
+        for r in target_rows:
+            code = (r.get("user_validation_choice") or r.get("matched_structure_code") or "").strip()
+            if not code:
+                continue
+            if code in seen_structures:
+                duplicates_skipped += 1
+                continue
+            seen_structures.add(code)
+            to_process.append(r)
+
+        self.stdout.write(f"Ouvrages cibles uniques à mettre à jour : {len(to_process)}")
+        self.stdout.write(f"Lignes doublons ignorées (première conservée) : {duplicates_skipped}\n")
+
+        created_count = 0
+        updated_count = 0
+        skipped_count = 0
+        notices_created_count = 0
+
+        with transaction.atomic():
+            for r in to_process:
+                code = (r.get("user_validation_choice") or r.get("matched_structure_code") or "").strip()
+                structure = Structure.objects.filter(code__iexact=code).first()
+                if not structure:
+                    self.stdout.write(self.style.WARNING(f"  [IGNORE] Structure avec code '{code}' introuvable en DB."))
+                    skipped_count += 1
+                    continue
+
+                fod_vnr = parse_int_safe(r.get("fod_vnr"))
+                fod_idn = (r.get("fod_idn") or "").strip() or None
+                bridge_class = (r.get("fod_bridge_class") or "").strip() or None
+                maxwa = parse_float_safe(r.get("fod_maxwa_tonnes"))
+                minwr = parse_float_safe(r.get("fod_minwr_tonnes"))
+                mx = parse_float_safe(r.get("fod_mx_tonnes"))
+                usable_width = parse_float_safe(r.get("fod_usable_width_m"))
+
+                # Spoorlijn
+                spoorlijn_raw = str(r.get("fod_spoorlijn") or "").strip().lower()
+                is_railway_metro = spoorlijn_raw in ("true", "1", "t", "oui", "yes")
+
+                # Hauteurs
+                h_str = r.get("fod_clearance_heights") or ""
+                h1_m, h2_m, h3_m = None, None, None
+                if h_str and "/" in h_str:
+                    parts = h_str.split("/")
+                    if len(parts) >= 3:
+                        h1_m = parse_height(parts[0])
+                        h2_m = parse_height(parts[1])
+                        h3_m = parse_height(parts[2])
+
+                # Prescriptions textuelles
+                p1 = (r.get("fod_prescriptions_1") or "").strip()
+                p2 = (r.get("fod_prescriptions_2") or "").strip()
+                p3 = (r.get("fod_prescriptions_3") or "").strip()
+
+                prescriptions_fr_list = []
+                prescriptions_nl_list = []
+
+                for p in (p1, p2, p3):
+                    if not p or len(p) < 3:
+                        continue
+                    p_lower = p.lower()
+                    if any(w in p_lower for w in ("passage", "autorisé", "condition", "vitesse", "limitée", "freinage", "chaussée", "ouvrage", "surplombe")):
+                        if p not in prescriptions_fr_list:
+                            prescriptions_fr_list.append(p)
+                    if any(w in p_lower for w in ("toegestaan", "voorwaarde", "snelheid", "beperkt", "metrotunnel", "overbelastingen", "structuur", "rijbaan")):
+                        if p not in prescriptions_nl_list:
+                            prescriptions_nl_list.append(p)
+
+                # Si W14 était en FR et non classé, fallback
+                if p1 and not prescriptions_fr_list and not prescriptions_nl_list:
+                    prescriptions_fr_list.append(p1)
+
+                presc_fr = "\n\n".join(prescriptions_fr_list) if prescriptions_fr_list else None
+                presc_nl = "\n\n".join(prescriptions_nl_list) if prescriptions_nl_list else None
+
+                # Détection des règles d'exploitation
+                all_text = f"{p1} {p2} {p3}".lower()
+                is_braking_forbidden = any(w in all_text for w in ("frein", "remmen", "sans freinage", "niet remmen"))
+                is_isolated_crossing = any(w in all_text for w in ("convoi seul", "seul", "alleen", "geen andere gelijktijdige", "pas simultanément d'autres surcharges", "pas d'autre surcharge"))
+                has_speed_5 = any(w in all_text for w in ("5km/h", "5 km/h", "5km/u", "5 km/u"))
+                max_speed = 5 if has_speed_5 else None
+
+                # Statut d'admissibilité
+                if presc_fr or presc_nl or is_braking_forbidden or is_isolated_crossing or max_speed:
+                    passage_status = "conditional"
+                elif maxwa or mx or minwr:
+                    passage_status = "authorized"
+                else:
+                    passage_status = "not_evaluated"
+
+                # Détection référence d'étude
+                last_study_ref = None
+                study_match = re.search(r"(avis\s*DITP[_\s\w\d]*|Mission\s*\d+|note\s*de\s*calcul[_\s\w\d]*)", all_text, re.IGNORECASE)
+                if study_match:
+                    last_study_ref = study_match.group(0).strip()
+
+                te_obj, created = StructureExceptionalTransport.objects.get_or_create(structure=structure)
+
+                te_obj.fod_vnr = fod_vnr
+                te_obj.national_id = fod_idn
+                if bridge_class:
+                    te_obj.bridge_class = bridge_class
+                if maxwa is not None:
+                    te_obj.max_authorized_weight_tonnes = maxwa
+                if minwr is not None:
+                    te_obj.min_guaranteed_capacity_tonnes = minwr
+                if mx is not None:
+                    te_obj.max_calculated_weight_tonnes = mx
+                if usable_width is not None:
+                    te_obj.usable_width_m = usable_width
+                if h1_m is not None:
+                    te_obj.clearance_height_left_m = h1_m
+                if h2_m is not None:
+                    te_obj.clearance_height_center_m = h2_m
+                if h3_m is not None:
+                    te_obj.clearance_height_right_m = h3_m
+                if is_railway_metro:
+                    te_obj.is_railway_metro_crossing = True
+                if is_braking_forbidden:
+                    te_obj.is_braking_forbidden = True
+                if is_isolated_crossing:
+                    te_obj.is_isolated_crossing_only = True
+                if max_speed:
+                    te_obj.max_speed_kmh = max_speed
+                if passage_status:
+                    te_obj.passage_status = passage_status
+                if presc_fr:
+                    te_obj.prescriptions_fr = presc_fr
+                if presc_nl:
+                    te_obj.prescriptions_nl = presc_nl
+                if last_study_ref:
+                    te_obj.last_study_ref = last_study_ref
+
+                if not dry_run:
+                    te_obj.save()
+
+                if created:
+                    created_count += 1
+                else:
+                    updated_count += 1
+
+                self.stdout.write(
+                    f"  [OK] {structure.code} ({structure.get_name()}) : VNR={fod_vnr}, BKL={bridge_class or '-'}, MAXWA={maxwa or '-'}t, Statut={passage_status}"
+                )
+
+            if dry_run:
+                transaction.set_rollback(True)
+
+        self.stdout.write("\n" + self.style.SUCCESS("=== Résultat de l'import ==="))
+        self.stdout.write(f"Ouvrages créés / initialisés : {created_count}")
+        self.stdout.write(f"Ouvrages mis à jour : {updated_count}")
+        self.stdout.write(f"Ouvrages ignorés : {skipped_count}")
+        if dry_run:
+            self.stdout.write(self.style.WARNING("Mode simulation : aucune modification n'a été enregistrée en base."))
+        else:
+            self.stdout.write(self.style.SUCCESS("Toutes les données ont été enregistrées avec succès en base de données."))