Add script to sync engagement items from CSV

Author: trecouvr

docia/management/commands/sync_engagement_items.py

@@ -0,0 +1,27 @@
+import os
+
+from django.core.management.base import BaseCommand, CommandError
+
+from docia.sync.sync_engagement_items import read_csv, sync
+
+
+class Command(BaseCommand):
+    help = "Syncs engagement items from a CSV file to the database"
+
+    def add_arguments(self, parser):
+        parser.add_argument("filename", type=str, help="Path to the CSV file to import")
+
+    def handle(self, *args, **options):
+        filename = options["filename"]
+
+        # Check if file exists
+        if not os.path.exists(filename):
+            raise CommandError(f'File "{filename}" does not exist')
+
+        data = read_csv(filename)
+        self.stdout.write(self.style.SUCCESS(f"Found {len(data)} items in CSV file"))
+
+        # Perform the sync operation
+        created = sync(data)
+
+        self.stdout.write(self.style.SUCCESS(f"Successfully synced engagement items ({len(created)} created)"))

docia/migrations/0007_alter_dataengagementitems_num_ej.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2.7 on 2025-10-14 14:33
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('docia', '0006_trackingevent'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='dataengagementitems',
+            name='num_ej',
+            field=models.CharField(max_length=20),
+        ),
+    ]

docia/models.py

@@ -88,7 +88,7 @@ class DataEngagementItems(models.Model):
         db_default=RandomUUID(),
         editable=False,
     )
-    num_ej = models.ForeignKey(DataEngagement, on_delete=models.CASCADE, db_column="num_ej", to_field="num_ej")
+    num_ej = models.CharField(max_length=20)
     poste_ej = models.CharField()
     num_contrat = models.CharField(null=True, blank=True)  # noqa: DJ001
     groupe_marchandise = models.CharField(null=True, blank=True)  # noqa: DJ001
@@ -99,7 +99,7 @@ class Meta:
         unique_together = [("num_ej", "poste_ej")]
 
     def __str__(self):
-        return f"{self.num_ej_id} - {self.poste_ej}"
+        return f"{self.num_ej} - {self.poste_ej}"
 
 
 class DataProgrammesMinisteriels(models.Model):

docia/sync/__init__.py

@@ -0,0 +1,33 @@
+import csv
+
+from django.db.models import F, Value
+from django.db.models.functions import Concat
+
+from docia.models import DataEngagementItems
+
+
+def read_csv(filename):
+    with open(filename, "r") as file:
+        reader = csv.DictReader(file, delimiter=";")
+        data = []
+        for row in reader:
+            item = DataEngagementItems(**row)
+            data.append(item)
+    return data
+
+
+def sync(data):
+    to_create = []
+    num_ejs = [f"{item.num_ej}__{item.poste_ej}" for item in data]
+    existing_items = list(
+        DataEngagementItems.objects.annotate(
+            key=Concat(F("num_ej"), Value("__"), F("poste_ej")),
+        )
+        .filter(key__in=num_ejs)
+        .values_list("key", flat=True)
+    )
+    for item in data:
+        if f"{item.num_ej}__{item.poste_ej}" not in existing_items:
+            to_create.append(item)
+    created = DataEngagementItems.objects.bulk_create(to_create)
+    return created