diff --git a/loko/assets/management/commands/import_exceptional_transport_fod.py b/loko/assets/management/commands/import_exceptional_transport_fod.py new file mode 100644 index 0000000..c6c6bc8 --- /dev/null +++ b/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."))