212 lines
8.3 KiB
Python
212 lines
8.3 KiB
Python
from django.core.management.base import BaseCommand
|
|
from django.db import transaction
|
|
|
|
from assets.models.sign import (
|
|
SignStreet as AssetsSignStreet,
|
|
SignStreetDocument as AssetsSignStreetDocument,
|
|
SignPoleModel as AssetsSignPoleModel,
|
|
SignPole as AssetsSignPole,
|
|
SignPanelModel as AssetsSignPanelModel,
|
|
SignPanelType as AssetsSignPanelType,
|
|
SignPanel as AssetsSignPanel,
|
|
ComplementaryPoliceRegulation as AssetsComplementaryPoliceRegulation,
|
|
)
|
|
|
|
from sign.models import (
|
|
SignStreet,
|
|
SignStreetDocument,
|
|
SignPoleModel,
|
|
SignPole,
|
|
SignPanelModel,
|
|
SignPanelType,
|
|
SignPanel,
|
|
ComplementaryPoliceRegulation,
|
|
)
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Copy sign-related data from assets app into sign app models."
|
|
|
|
@transaction.atomic
|
|
def handle(self, *args, **options):
|
|
self.stdout.write(self.style.WARNING("Starting copy of sign data from assets to sign..."))
|
|
|
|
# Maps for ID relationships
|
|
type_map = {}
|
|
pole_model_map = {}
|
|
panel_model_map = {}
|
|
street_map = {}
|
|
pole_map = {}
|
|
panel_map = {}
|
|
|
|
# 1) Panel Types
|
|
for src in AssetsSignPanelType.objects.all():
|
|
dst, _ = SignPanelType.objects.update_or_create(
|
|
code=src.code,
|
|
defaults={
|
|
"name_fr": src.name_fr,
|
|
"name_nl": src.name_nl,
|
|
"description": src.description,
|
|
"sgv_picture": src.sgv_picture,
|
|
},
|
|
)
|
|
type_map[src.id] = dst
|
|
self.stdout.write(self.style.SUCCESS(f"Copied {len(type_map)} panel types"))
|
|
|
|
# 2) Pole Models
|
|
for src in AssetsSignPoleModel.objects.all():
|
|
dst, _ = SignPoleModel.objects.update_or_create(
|
|
code=src.code,
|
|
defaults={
|
|
"name_fr": src.name_fr,
|
|
"name_nl": src.name_nl,
|
|
"description": src.description,
|
|
"brand": src.brand,
|
|
"color": src.color,
|
|
"ral": src.ral,
|
|
"height": src.height,
|
|
"diameter": src.diameter,
|
|
"material": src.material,
|
|
"category": src.category,
|
|
},
|
|
)
|
|
pole_model_map[src.id] = dst
|
|
self.stdout.write(self.style.SUCCESS(f"Copied {len(pole_model_map)} pole models"))
|
|
|
|
# 3) Panel Models
|
|
for src in AssetsSignPanelModel.objects.all():
|
|
dst, _ = SignPanelModel.objects.update_or_create(
|
|
code=src.code,
|
|
defaults={
|
|
"name_fr": src.name_fr,
|
|
"name_nl": src.name_nl,
|
|
"description": src.description,
|
|
"brand": src.brand,
|
|
"shape": src.shape,
|
|
"color": src.color,
|
|
"ral": src.ral,
|
|
"height": src.height,
|
|
"width": src.width,
|
|
"diameter": src.diameter,
|
|
"material": src.material,
|
|
"category": src.category,
|
|
},
|
|
)
|
|
panel_model_map[src.id] = dst
|
|
self.stdout.write(self.style.SUCCESS(f"Copied {len(panel_model_map)} panel models"))
|
|
|
|
# 4) Streets
|
|
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_map[src.id] = dst
|
|
self.stdout.write(self.style.SUCCESS(f"Copied {len(street_map)} streets"))
|
|
|
|
# 5) Poles
|
|
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"))
|
|
|
|
# 6) Panels
|
|
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
|
|
)
|
|
panel_map[src.id] = dst
|
|
self.stdout.write(self.style.SUCCESS(f"Copied {len(panel_map)} panels"))
|
|
|
|
# 7) Street Documents
|
|
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"))
|
|
|
|
# 8) Complementary Police Regulations
|
|
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"))
|
|
|
|
self.stdout.write(self.style.SUCCESS("Copy completed successfully."))
|