diff --git a/loko/sign/management/commands/copy_sign_data.py b/loko/sign/management/commands/copy_sign_data.py index 044496a..71ea56d 100644 --- a/loko/sign/management/commands/copy_sign_data.py +++ b/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( - code=src.code, - name_fr=src.name_fr, - name_nl=src.name_nl, - 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), + street_code = src.code or f"STR-{src.id}" + dst = SignStreet.objects.filter(code=street_code).first() + if not dst: + dst = SignStreet(code=street_code) - ) + 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( - code=src.code, - name_fr=src.name_fr, - name_nl=src.name_nl, - 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")) + dst_code = src.code or f"POLE-{src.id}" + dst = SignPole.objects.filter(code=dst_code).first() + if not dst: + dst = SignPole(code=dst_code) - # 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( - code=src.code, - name_fr=src.name_fr, - name_nl=src.name_nl, - 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, - 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 + dst_code = src.code or f"PANEL-{src.id}" + dst = SignPanel.objects.filter(code=dst_code).first() + if not dst: + dst = SignPanel(code=dst_code) + + dst_model = panel_model_map.get(getattr(src.model, "id", None)) + dst_type = type_map.get(getattr(src.signpanel_type, "id", None)) + dst_pole = pole_map.get(getattr(src.pole, "id", None)) + dst_org = ( + dst_pole.organization + if dst_pole and dst_pole.organization + else get_organization_for_street(src.street) ) + + # 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( - street=street_map.get(getattr(src.street, "id", None)), - file=src.file, - description=src.description, - upload_date=src.upload_date, - status=src.status, - thumbnail=src.thumbnail, - document_type=src.document_type, - uploaded_by=src.uploaded_by, - managed_document=src.managed_document, - ) - self.stdout.write(self.style.SUCCESS("Copied street documents")) + dst_street = street_map.get(getattr(src.street, "id", None)) + if dst_street and src.file: + dst = SignStreetDocument.objects.filter( + street=dst_street, file=src.file + ).first() + if not dst: + dst = SignStreetDocument(street=dst_street, file=src.file) + dst.description = src.description + dst.upload_date = src.upload_date + dst.status = src.status + dst.thumbnail = src.thumbnail + 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( - title=src.title, - reference=src.reference, - decision_date=src.decision_date, - 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")) + ref = src.reference or f"REG-{src.id}" + dst = ComplementaryPoliceRegulation.objects.filter(reference=ref).first() + if not dst: + dst = ComplementaryPoliceRegulation(reference=ref) - 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.")) diff --git a/loko/sign/migrations/0003_signpanel_external_reference_and_more.py b/loko/sign/migrations/0003_signpanel_external_reference_and_more.py new file mode 100644 index 0000000..9871a98 --- /dev/null +++ b/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'), + ), + ] diff --git a/loko/sign/tests.py b/loko/sign/tests.py index 87b4020..51910b4 100644 --- a/loko/sign/tests.py +++ b/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) +