feat: add external reference fields and implement migration script for asset data synchronization

loko/sign/management/commands/copy_sign_data.py

@@ -1,6 +1,10 @@
+import os
 from django.core.management.base import BaseCommand
 from django.db import transaction
+from django.contrib.gis.db.models.functions import Distance
+from django.conf import settings
 
+from common.models import AdrStreet
 from assets.models.sign import (
     SignStreet as AssetsSignStreet,
     SignStreetDocument as AssetsSignStreetDocument,
@@ -13,6 +17,7 @@ from assets.models.sign import (
 )
 
 from sign.models import (
+    Organization,
     SignStreet,
     SignStreetDocument,
     SignPoleModel,
@@ -27,10 +32,74 @@ from sign.models import (
 class Command(BaseCommand):
     help = "Copy sign-related data from assets app into sign app models."
 
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--organization",
+            type=str,
+            default=None,
+            help="Code or name of the Organization to associate with copied assets.",
+        )
+        parser.add_argument(
+            "--clean",
+            action="store_true",
+            default=False,
+            help="Clear existing sign app data before copying.",
+        )
+
     @transaction.atomic
     def handle(self, *args, **options):
         self.stdout.write(self.style.WARNING("Starting copy of sign data from assets to sign..."))
 
+        if options.get("clean"):
+            self.stdout.write(self.style.WARNING("Cleaning existing sign app data..."))
+            ComplementaryPoliceRegulation.objects.all().delete()
+            SignStreetDocument.objects.all().delete()
+            SignPanel.objects.all().delete()
+            SignPole.objects.all().delete()
+            SignStreet.objects.all().delete()
+            SignPanelModel.objects.all().delete()
+            SignPoleModel.objects.all().delete()
+            SignPanelType.objects.all().delete()
+            self.stdout.write(self.style.SUCCESS("Existing sign data deleted."))
+
+        # 0) Resolve default organization
+        default_org = None
+        org_arg = options.get("organization")
+        if org_arg:
+            default_org = (
+                Organization.objects.filter(code__iexact=org_arg).first()
+                or Organization.objects.filter(name_fr__icontains=org_arg).first()
+                or Organization.objects.filter(name_nl__icontains=org_arg).first()
+            )
+            if not default_org:
+                self.stdout.write(
+                    self.style.WARNING(f"Specified organization '{org_arg}' not found in DB.")
+                )
+        if not default_org:
+            default_org = (
+                Organization.objects.filter(code__iexact="SPRB").first()
+                or Organization.objects.filter(organization_type="region").first()
+                or Organization.objects.first()
+            )
+
+        if default_org:
+            self.stdout.write(self.style.NOTICE(f"Using default organization: {default_org.code} - {default_org.name}"))
+        else:
+            self.stdout.write(self.style.NOTICE("No default organization found; organization will be null where permitted."))
+
+        def get_organization_for_street(src_street):
+            if src_street and src_street.municipality:
+                muni = src_street.municipality
+                muni_name = muni.name_fr or muni.name_nl or ""
+                org = Organization.objects.filter(name_fr__iexact=muni_name).first()
+                if org:
+                    return org
+                if hasattr(muni, "code") and muni.code:
+                    org = Organization.objects.filter(code__iexact=muni.code).first()
+                    if org:
+                        return org
+            return default_org
+
         # Maps for ID relationships
         type_map = {}
         pole_model_map = {}
@@ -38,6 +107,60 @@ class Command(BaseCommand):
         street_map = {}
         pole_map = {}
         panel_map = {}
+        adr_street_cache = {}
+
+        def get_adr_street(src_street, geom=None):
+            """Resolve corresponding common.AdrStreet from an AssetsSignStreet or geometry."""
+            if src_street:
+                if src_street.id in adr_street_cache:
+                    return adr_street_cache[src_street.id]
+
+                adr = None
+                # Try match by French name
+                if src_street.name_fr:
+                    qs = AdrStreet.objects.filter(name_fr__iexact=src_street.name_fr.strip())
+                    if src_street.municipality_id:
+                        adr = qs.filter(municipality_id=src_street.municipality_id).first()
+                    if not adr:
+                        adr = qs.first()
+
+                # Try match by Dutch name
+                if not adr and src_street.name_nl:
+                    qs = AdrStreet.objects.filter(name_nl__iexact=src_street.name_nl.strip())
+                    if src_street.municipality_id:
+                        adr = qs.filter(municipality_id=src_street.municipality_id).first()
+                    if not adr:
+                        adr = qs.first()
+
+                # Try spatial proximity with src_street geometry
+                if not adr and src_street.geom:
+                    try:
+                        adr = (
+                            AdrStreet.objects.filter(geom__isnull=False)
+                            .annotate(distance=Distance("geom", src_street.geom))
+                            .order_by("distance")
+                            .first()
+                        )
+                    except Exception:
+                        pass
+
+                adr_street_cache[src_street.id] = adr
+                if adr:
+                    return adr
+
+            # Fallback to asset point geometry
+            if geom:
+                try:
+                    return (
+                        AdrStreet.objects.filter(geom__isnull=False)
+                        .annotate(distance=Distance("geom", geom))
+                        .order_by("distance")
+                        .first()
+                    )
+                except Exception:
+                    pass
+
+            return None
 
         # 1) Panel Types
         for src in AssetsSignPanelType.objects.all():
@@ -48,10 +171,13 @@ class Command(BaseCommand):
                     "name_nl": src.name_nl,
                     "description": src.description,
                     "sgv_picture": src.sgv_picture,
+                    "default_width": getattr(src, "default_width", None),
+                    "default_height": getattr(src, "default_height", None),
+                    "default_diameter": getattr(src, "default_diameter", None),
                 },
             )
             type_map[src.id] = dst
-        self.stdout.write(self.style.SUCCESS(f"Copied {len(type_map)} panel types"))
+        self.stdout.write(self.style.SUCCESS(f"Copied/Updated {len(type_map)} panel types"))
 
         # 2) Pole Models
         for src in AssetsSignPoleModel.objects.all():
@@ -68,10 +194,11 @@ class Command(BaseCommand):
                     "diameter": src.diameter,
                     "material": src.material,
                     "category": src.category,
+                    "galvanized": getattr(src, "galvanized", False),
                 },
             )
             pole_model_map[src.id] = dst
-        self.stdout.write(self.style.SUCCESS(f"Copied {len(pole_model_map)} pole models"))
+        self.stdout.write(self.style.SUCCESS(f"Copied/Updated {len(pole_model_map)} pole models"))
 
         # 3) Panel Models
         for src in AssetsSignPanelModel.objects.all():
@@ -90,123 +217,197 @@ class Command(BaseCommand):
                     "diameter": src.diameter,
                     "material": src.material,
                     "category": src.category,
+                    "reflective": getattr(src, "reflective", False),
+                    "illuminated": getattr(src, "illuminated", False),
+                    "dynamic": getattr(src, "dynamic", False),
                 },
             )
             panel_model_map[src.id] = dst
-        self.stdout.write(self.style.SUCCESS(f"Copied {len(panel_model_map)} panel models"))
+        self.stdout.write(self.style.SUCCESS(f"Copied/Updated {len(panel_model_map)} panel models"))
 
-        # 4) Streets
+        # 4) Streets (SignStreet)
         for src in AssetsSignStreet.objects.all():
-            dst = SignStreet.objects.create(
+            street_code = src.code or f"STR-{src.id}"
-                code=src.code,
+            dst = SignStreet.objects.filter(code=street_code).first()
-                name_fr=src.name_fr,
+            if not dst:
-                name_nl=src.name_nl,
+                dst = SignStreet(code=street_code)
-                lon=src.lon,
-                lat=src.lat,
-                geom=src.geom,
-                geojson=src.geojson,
-                municipality=src.municipality,
-                road_code=getattr(src, "road_code", None),
-                section_code=getattr(src, "section_code", None),
-                street_level=getattr(src, "street_level", None),
 
-            )
+            dst.name_fr = src.name_fr
+            dst.name_nl = src.name_nl
+            dst.lon = src.lon
+            dst.lat = src.lat
+            dst.geom = src.geom
+            dst.geojson = src.geojson
+            dst.municipality = src.municipality
+            dst.road_code = getattr(src, "road_code", None)
+            dst.section_code = getattr(src, "section_code", None)
+            dst.street_level = getattr(src, "street_level", None)
+            dst.save()
             street_map[src.id] = dst
-        self.stdout.write(self.style.SUCCESS(f"Copied {len(street_map)} streets"))
+        self.stdout.write(self.style.SUCCESS(f"Copied/Updated {len(street_map)} streets"))
 
-        # 5) Poles
+        # 5) Poles (SignPole)
         for src in AssetsSignPole.objects.all():
-            dst = SignPole.objects.create(
+            dst_code = src.code or f"POLE-{src.id}"
-                code=src.code,
+            dst = SignPole.objects.filter(code=dst_code).first()
-                name_fr=src.name_fr,
+            if not dst:
-                name_nl=src.name_nl,
+                dst = SignPole(code=dst_code)
-                status=src.status,
-                installation_date=src.installation_date,
-                uninstallation_date=src.uninstallation_date,
-                serial_number=src.serial_number,
-                warranty_duration=src.warranty_duration,
-                last_inspection_date=src.last_inspection_date,
-                last_preventive_date=src.last_preventive_date,
-                last_corrective_date=src.last_corrective_date,
-                last_survey_date=src.last_survey_date,
-                category=src.category,
-                brand=src.brand,
-                lon=src.lon,
-                lat=src.lat,
-                geom=src.geom,
-                geojson=src.geojson,
-                model= pole_model_map.get(getattr(src.model, "id", None)),
-                street= street_map.get(getattr(src.street, "id", None)),
-                # organization intentionally left None; define mapping if available
-            )
-            pole_map[src.id] = dst
-        self.stdout.write(self.style.SUCCESS(f"Copied {len(pole_map)} poles"))
 
-        # 6) Panels
+            dst_model = pole_model_map.get(getattr(src.model, "id", None))
+            dst_org = get_organization_for_street(src.street)
+            adr_street = get_adr_street(src.street, src.geom)
+
+            dst.name_fr = src.name_fr
+            dst.name_nl = src.name_nl
+            dst.status = src.status or "active"
+            dst.installation_date = src.installation_date
+            dst.uninstallation_date = src.uninstallation_date
+            dst.serial_number = src.serial_number
+            dst.warranty_duration = src.warranty_duration
+            dst.last_inspection_date = src.last_inspection_date
+            dst.last_preventive_date = src.last_preventive_date
+            dst.last_corrective_date = src.last_corrective_date
+            dst.last_survey_date = src.last_survey_date
+            dst.category = src.category
+            dst.brand = src.brand
+            dst.lon = src.lon
+            dst.lat = src.lat
+            dst.geom = src.geom
+            dst.geojson = src.geojson
+            dst.model = dst_model
+            dst.street = adr_street
+            dst.organization = dst_org
+            dst.orientation = getattr(src, "orientation", 90) or 90
+            dst.external_reference = getattr(src, "external_reference", None)
+            dst.funding_program = getattr(src, "funding_program", None)
+            dst.inspection_status = getattr(src, "inspection_status", "unknown") or "unknown"
+            dst.note = getattr(src, "note", None)
+            dst.validation_status = getattr(src, "validation_status", None)
+            dst.save()
+            pole_map[src.id] = dst
+        self.stdout.write(self.style.SUCCESS(f"Copied/Updated {len(pole_map)} poles"))
+
+        # 6) Panels (SignPanel)
         for src in AssetsSignPanel.objects.all():
-            dst = SignPanel.objects.create(
+            dst_code = src.code or f"PANEL-{src.id}"
-                code=src.code,
+            dst = SignPanel.objects.filter(code=dst_code).first()
-                name_fr=src.name_fr,
+            if not dst:
-                name_nl=src.name_nl,
+                dst = SignPanel(code=dst_code)
-                status=src.status,
+
-                installation_date=src.installation_date,
+            dst_model = panel_model_map.get(getattr(src.model, "id", None))
-                uninstallation_date=src.uninstallation_date,
+            dst_type = type_map.get(getattr(src.signpanel_type, "id", None))
-                serial_number=src.serial_number,
+            dst_pole = pole_map.get(getattr(src.pole, "id", None))
-                warranty_duration=src.warranty_duration,
+            dst_org = (
-                last_inspection_date=src.last_inspection_date,
+                dst_pole.organization
-                last_preventive_date=src.last_preventive_date,
+                if dst_pole and dst_pole.organization
-                last_corrective_date=src.last_corrective_date,
+                else get_organization_for_street(src.street)
-                last_survey_date=src.last_survey_date,
-                category=src.category,
-                brand=src.brand,
-                lon=src.lon,
-                lat=src.lat,
-                geom=src.geom,
-                geojson=src.geojson,
-                vertical_order=src.vertical_order,
-                signpanel_text=src.signpanel_text,
-                signpanel_value=src.signpanel_value,
-                orientation=src.orientation,
-                support_type=src.support_type,
-                is_temporary=src.is_temporary,
-                model=panel_model_map.get(getattr(src.model, "id", None)),
-                signpanel_type=type_map.get(getattr(src.signpanel_type, "id", None)),
-                pole=pole_map.get(getattr(src.pole, "id", None)),
-                street=street_map.get(getattr(src.street, "id", None)),
-                # organization intentionally left None
             )
+
+            # Use pole's street if available, otherwise resolve from src.street or point geom
+            adr_street = (
+                dst_pole.street
+                if (dst_pole and dst_pole.street)
+                else get_adr_street(src.street, src.geom)
+            )
+
+            width = getattr(src, "width", None)
+            if not width and hasattr(src, "get_width"):
+                try:
+                    width = src.get_width()
+                except Exception:
+                    pass
+
+            height = getattr(src, "height", None)
+            if not height and hasattr(src, "get_height"):
+                try:
+                    height = src.get_height()
+                except Exception:
+                    pass
+
+            dst.name_fr = src.name_fr
+            dst.name_nl = src.name_nl
+            dst.status = src.status or "active"
+            dst.installation_date = src.installation_date
+            dst.uninstallation_date = src.uninstallation_date
+            dst.serial_number = src.serial_number
+            dst.warranty_duration = src.warranty_duration
+            dst.last_inspection_date = src.last_inspection_date
+            dst.last_preventive_date = src.last_preventive_date
+            dst.last_corrective_date = src.last_corrective_date
+            dst.last_survey_date = src.last_survey_date
+            dst.category = src.category
+            dst.brand = src.brand
+            dst.lon = src.lon
+            dst.lat = src.lat
+            dst.geom = src.geom
+            dst.geojson = src.geojson
+            dst.vertical_order = src.vertical_order
+            dst.signpanel_text = src.signpanel_text
+            dst.signpanel_value = src.signpanel_value
+            dst.orientation = src.orientation
+            dst.support_type = src.support_type
+            dst.is_temporary = src.is_temporary or False
+            dst.model = dst_model
+            dst.signpanel_type = dst_type
+            dst.pole = dst_pole
+            dst.street = adr_street
+            dst.organization = dst_org
+            dst.width = width
+            dst.height = height
+            dst.external_reference = getattr(src, "external_reference", None)
+            dst.funding_program = getattr(src, "funding_program", None)
+            dst.inspection_status = getattr(src, "inspection_status", "unknown") or "unknown"
+            dst.note = getattr(src, "note", None)
+            dst.validation_status = getattr(src, "validation_status", None)
+            dst.save()
             panel_map[src.id] = dst
-        self.stdout.write(self.style.SUCCESS(f"Copied {len(panel_map)} panels"))
+        self.stdout.write(self.style.SUCCESS(f"Copied/Updated {len(panel_map)} panels"))
 
         # 7) Street Documents
+        copied_docs = 0
         for src in AssetsSignStreetDocument.objects.all():
-            dst = SignStreetDocument.objects.create(
+            dst_street = street_map.get(getattr(src.street, "id", None))
-                street=street_map.get(getattr(src.street, "id", None)),
+            if dst_street and src.file:
-                file=src.file,
+                dst = SignStreetDocument.objects.filter(
-                description=src.description,
+                    street=dst_street, file=src.file
-                upload_date=src.upload_date,
+                ).first()
-                status=src.status,
+                if not dst:
-                thumbnail=src.thumbnail,
+                    dst = SignStreetDocument(street=dst_street, file=src.file)
-                document_type=src.document_type,
+                dst.description = src.description
-                uploaded_by=src.uploaded_by,
+                dst.upload_date = src.upload_date
-                managed_document=src.managed_document,
+                dst.status = src.status
-            )
+                dst.thumbnail = src.thumbnail
-        self.stdout.write(self.style.SUCCESS("Copied street documents"))
+                dst.document_type = src.document_type
+                dst.uploaded_by = src.uploaded_by
+                dst.managed_document = getattr(src, "managed_document", None)
+                dst.organization = get_organization_for_street(src.street)
+                dst.save()
+                copied_docs += 1
+        self.stdout.write(self.style.SUCCESS(f"Copied/Updated {copied_docs} street documents"))
 
         # 8) Complementary Police Regulations
+        copied_regs = 0
         for src in AssetsComplementaryPoliceRegulation.objects.all():
-            dst = ComplementaryPoliceRegulation.objects.create(
+            ref = src.reference or f"REG-{src.id}"
-                title=src.title,
+            dst = ComplementaryPoliceRegulation.objects.filter(reference=ref).first()
-                reference=src.reference,
+            if not dst:
-                decision_date=src.decision_date,
+                dst = ComplementaryPoliceRegulation(reference=ref)
-                file=src.file,
-                thumbnail=src.thumbnail,
-                uploaded_by=src.uploaded_by,
-                managed_document=src.managed_document,
-            )
-            # M2M panels mapping
-            if dst:
-                dst.panels.set([panel_map.get(p.id) for p in src.panels.all() if panel_map.get(p.id)])
-        self.stdout.write(self.style.SUCCESS("Copied complementary police regulations"))
 
-        self.stdout.write(self.style.SUCCESS("Copy completed successfully."))
+            dst.title = src.title or ref
+            dst.description = getattr(src, "description", None)
+            dst.decision_date = src.decision_date
+            dst.status = "approved" if src.decision_date else "in_preparation"
+            dst.file = src.file
+            dst.thumbnail = src.thumbnail
+            dst.uploaded_by = src.uploaded_by
+            dst.organization = default_org
+            dst.save()
+
+            # M2M panels mapping (excluding any panels not in panel_map)
+            dst_panels = [panel_map[p.id] for p in src.panels.all() if p.id in panel_map]
+            if dst_panels:
+                dst.panels.set(dst_panels)
+            copied_regs += 1
+        self.stdout.write(self.style.SUCCESS(f"Copied/Updated {copied_regs} complementary police regulations"))
+
+        self.stdout.write(self.style.SUCCESS("Copy of sign data completed successfully."))

loko/sign/migrations/0003_signpanel_external_reference_and_more.py

@@ -0,0 +1,23 @@
+# Generated by Django 6.0.7 on 2026-09-08 14:58
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('sign', '0002_signpanel_funding_program_and_more'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='signpanel',
+            name='external_reference',
+            field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
+        ),
+        migrations.AddField(
+            model_name='signpole',
+            name='external_reference',
+            field=models.CharField(blank=True, max_length=100, null=True, verbose_name='Référence externe'),
+        ),
+    ]

loko/sign/tests.py

@@ -655,3 +655,132 @@ class ESignOrganizationRestrictionsTests(TestCase):
             }
         )
         self.assertEqual(response_edit_other_rcp.status_code, 403)
+
+
+class CopySignDataCommandTests(TestCase):
+    """Tests for the copy_sign_data management command."""
+
+    def test_copy_sign_data_command(self):
+        from io import StringIO
+        from django.core.management import call_command
+        from django.contrib.gis.geos import Point
+        from common.models import Municipality, AdrStreet
+        from assets.models import AssetCategory
+        from assets.models.sign import (
+            SignPanelType as AssetsSignPanelType,
+            SignPoleModel as AssetsSignPoleModel,
+            SignPanelModel as AssetsSignPanelModel,
+            SignStreet as AssetsSignStreet,
+            SignPole as AssetsSignPole,
+            SignPanel as AssetsSignPanel,
+            ComplementaryPoliceRegulation as AssetsComplementaryPoliceRegulation,
+        )
+        from sign.models import (
+            SignPanelType,
+            SignPoleModel,
+            SignPanelModel,
+            SignStreet,
+            SignPole,
+            SignPanel,
+            ComplementaryPoliceRegulation,
+            Organization,
+        )
+
+        # 1. Setup Source Data in Assets
+        muni = Municipality.objects.create(fid="1000", name_fr="Bruxelles", name_nl="Brussel")
+        adr_street = AdrStreet.objects.create(
+            fid="STR-1",
+            name_fr="Rue Royale",
+            name_nl="Koningsstraat",
+            municipality=muni,
+        )
+        org = Organization.objects.create(
+            code="SPRB",
+            name_fr="Bruxelles Mobilité",
+            name_nl="Brussel Mobiliteit",
+            organization_type="region",
+        )
+
+        cat = AssetCategory.objects.create(code="sign_cat", name_fr="Sign Cat")
+        src_ptype = AssetsSignPanelType.objects.create(
+            code="C1",
+            name_fr="Sens interdit",
+            name_nl="Verboden richting",
+            default_width=60.0,
+            default_height=60.0,
+        )
+        src_pmodel = AssetsSignPoleModel.objects.create(
+            code="PM-1",
+            name_fr="Poteau Acier 3m",
+            category=cat,
+            color="Gris",
+            height=3.0,
+        )
+        src_panel_model = AssetsSignPanelModel.objects.create(
+            code="PANM-1",
+            name_fr="Modèle C1 60",
+            category=cat,
+            shape="rond",
+        )
+        src_street = AssetsSignStreet.objects.create(
+            code="STR-ROYALE",
+            name_fr="Rue Royale",
+            name_nl="Koningsstraat",
+            municipality=muni,
+        )
+        src_pole = AssetsSignPole.objects.create(
+            code="POL-001",
+            name_fr="Poteau 1",
+            street=src_street,
+            model=src_pmodel,
+            category=cat,
+            geom=Point(149000, 171000, srid=3812),
+        )
+        src_panel = AssetsSignPanel.objects.create(
+            code="PAN-001",
+            name_fr="Panneau C1",
+            street=src_street,
+            pole=src_pole,
+            model=src_panel_model,
+            signpanel_type=src_ptype,
+            category=cat,
+            geom=Point(149000, 171000, srid=3812),
+        )
+        src_reg = AssetsComplementaryPoliceRegulation.objects.create(
+            title="Règlement Test",
+            reference="RCP-TEST-001",
+        )
+        src_reg.panels.add(src_panel)
+
+        # 2. Execute command
+        out = StringIO()
+        call_command("copy_sign_data", stdout=out)
+        output = out.getvalue()
+        self.assertIn("Copy of sign data completed successfully", output)
+
+        # 3. Assertions on copied destination models in Sign app
+        dst_ptype = SignPanelType.objects.filter(code="C1").first()
+        self.assertIsNotNone(dst_ptype)
+        self.assertEqual(dst_ptype.default_width, 60.0)
+
+        dst_pole = SignPole.objects.filter(code="POL-001").first()
+        self.assertIsNotNone(dst_pole)
+        self.assertEqual(dst_pole.street, adr_street)
+        self.assertEqual(dst_pole.organization, org)
+
+        dst_panel = SignPanel.objects.filter(code="PAN-001").first()
+        self.assertIsNotNone(dst_panel)
+        self.assertEqual(dst_panel.pole, dst_pole)
+        self.assertEqual(dst_panel.street, adr_street)
+        self.assertEqual(dst_panel.organization, org)
+
+        dst_reg = ComplementaryPoliceRegulation.objects.filter(reference="RCP-TEST-001").first()
+        self.assertIsNotNone(dst_reg)
+        self.assertIn(dst_panel, dst_reg.panels.all())
+
+        # 4. Verify idempotence
+        out2 = StringIO()
+        call_command("copy_sign_data", stdout=out2)
+        self.assertEqual(SignPole.objects.filter(code="POL-001").count(), 1)
+        self.assertEqual(SignPanel.objects.filter(code="PAN-001").count(), 1)
+