Merge branch 'command-change-role-end-dates' into 'master'

Add a management command for batch updating role end dates from CSV files

See merge request !386

greg/management/commands/change_role_end_dates.py

+"""
+Command for changing role end dates in batch.
+"""
+import csv
+import logging
+
+from django.core.management.base import BaseCommand, CommandParser
+from django.db import transaction
+from django.utils.dateparse import parse_date
+
+from greg.models import Identity, Person, Role
+
+logger = logging.getLogger(__name__)
+
+
+def handle_line(line: dict):
+    feide_id = line["feide_id"]
+    stedkode = line["legacy_stedkode"]
+    role_type = line["role_type"]
+    if not line.get("end_date"):
+        logger.info("No end date set: %r", line)
+        return
+    end_date_str = line.get("end_date")
+    assert isinstance(end_date_str, str)
+    end_date = parse_date(end_date_str)
+    assert feide_id and stedkode and role_type and end_date
+    try:
+        person = Person.objects.get(
+            identities__type=Identity.IdentityType.FEIDE_ID, identities__value=feide_id
+        )
+    except Person.DoesNotExist:
+        logger.warning("No person with feide_id %r", feide_id)
+        return
+    logger.info("Found %s: %s", feide_id, person)
+    try:
+        role = Role.objects.get(
+            person=person,
+            type__identifier=role_type,
+            orgunit__identifiers__name="legacy_stedkode",
+            orgunit__identifiers__value=stedkode,
+        )
+    except Role.MultipleObjectsReturned:
+        logger.warning(
+            "Multiple roles matching type=%s stedkode=%s for feide_id=%s",
+            role_type,
+            stedkode,
+            feide_id,
+        )
+        return
+    except Role.DoesNotExist:
+        logger.warning(
+            "No role matching type=%s stedkode=%s for feide_id=%s",
+            role_type,
+            stedkode,
+            feide_id,
+        )
+        return
+    logger.info("Found matching role %s for %s", role, feide_id)
+    old_end_date = role.end_date
+    role.end_date = end_date
+    changes = role.get_dirty_fields(check_relationship=True, verbose=True)
+    logger.info(
+        "Changing end_date for role_id=%d from %s to %s %r",
+        role.pk,
+        old_end_date,
+        end_date,
+        changes,
+    )
+    role.save()
+
+
+class Command(BaseCommand):
+    help = __doc__
+
+    def add_arguments(self, parser: CommandParser) -> None:
+        parser.add_argument(
+            "--file",
+            type=str,
+            help="Path to CSV file with new end dates",
+        )
+        parser.add_argument(
+            "--commit",
+            default=False,
+            action="store_true",
+            help="Commit the changes done by running the import once",
+        )
+
+    def handle(self, *args, **options):
+        commit = options["commit"]
+        file_path = options["file"]
+
+        with open(file_path, encoding="utf-8", newline="") as csv_file:
+            reader = csv.DictReader(csv_file)
+            lines = list(reader)
+
+        with transaction.atomic():
+            for line in lines:
+                handle_line(line)
+
+            if not commit:
+                logger.info("Rolling back changes...")
+                transaction.set_rollback(True)
+            else:
+                logger.info("Changes committed")