Greg 95 cerebrum sponsor import

greg/management/commands/import_sponsors_from_cerebrum.py

+"""
+Fetch sponsors for all units in GREG.
+
+Uses the members of the adm-leder-<legacy_stedkode> groups to
+populate Sponsors and SponsorOrganizationalUnits.
+
+This script does only remove the SponsorOrganizationalUnit.
+The Sponsor objects are kept, even with no units
+"""
+
+from typing import Optional, Tuple
+import cerebrum_client
+
+import structlog
+
+from cerebrum_client import CerebrumClient
+from django.conf import settings
+from django.core.management.base import BaseCommand
+
+from greg.models import OrganizationalUnit, Sponsor, SponsorOrganizationalUnit
+
+logger = structlog.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    help = __doc__
+
+    CEREBRUM_SOURCE = "cerebrum"
+    CEREBRUM_FEIDE_INST = "uio.no"
+    CEREBRUM_NAME_SOURCE_PRIORITY = ["Cached", "Override", "DFO_SAP", "FS", "Manual"]
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.client = CerebrumClient(**settings.CEREBRUM_CLIENT)
+
+    def _has_active_dfo_aff(self, person_id: str):
+        """Check that a person has a valid employee affiliation from DFØ."""
+
+        dfo_employee_affs = [
+            x
+            for x in self.client.get_person_affiliations(person_id)
+            if x.source_system == "DFO_SAP" and x.affiliation == "ANSATT"
+        ]
+
+        return len(dfo_employee_affs) > 0
+
+    def _upsert_sponsor_unit_link(self, sponsor: Sponsor, unit: OrganizationalUnit):
+        """Ensure a link between sponsor and unit."""
+        sunit, created = SponsorOrganizationalUnit.objects.get_or_create(
+            sponsor=sponsor,
+            organizational_unit=unit,
+            source=self.CEREBRUM_SOURCE,
+            automatic=True,
+        )
+        if created:
+            logger.info("sponsor_ou_link_create", sponsor=sponsor.id, sunit=sunit.id)
+        else:
+            logger.info("sponsor_ou_link_found", sponsor=sponsor.id, sunit=sunit.id)
+            sunit.hierarchical_access = settings.CEREBRUM_HIERARCHICAL_ACCESS
+            sunit.save()
+
+        return SponsorOrganizationalUnit.objects.get(id=sunit.id)
+
+    def _remove_sponsor_unit_link(self, sunit: SponsorOrganizationalUnit):
+        logger.info("sponsor_ou_deleted", sunit=sunit.id)
+        sunit.delete()
+
+    def _get_person_name(
+        self, person: cerebrum_client.models.Person
+    ) -> Tuple[Optional[str], Optional[str]]:
+        """Get a persons chosen name."""
+        first_names = {x.source_system: x for x in person.names if x.variant == "FIRST"}
+        last_names = {x.source_system: x for x in person.names if x.variant == "LAST"}
+
+        for source_system in self.CEREBRUM_NAME_SOURCE_PRIORITY:
+            if source_system in first_names and source_system in last_names:
+                return first_names[source_system].name, last_names[source_system].name
+
+        return None, None
+
+    def _get_feide_id(self, person_id: str) -> Optional[str]:
+        """Infer the feide id from the primary user."""
+        primary_uname = self.client.get_person_primary_account_name(person_id)
+
+        if primary_uname:
+            return f"{primary_uname}@{self.CEREBRUM_FEIDE_INST}"
+        return None
+
+    def _upsert_sponsor_from_cerebrum(
+        self, person_id: str, unit: OrganizationalUnit
+    ) -> Optional[Sponsor]:
+        """Insert or update a sponsor from Cerebum data."""
+        person = self.client.get_person(person_id)
+        if not person:
+            logger.warning("cerebrum_person_missing", cerebrum_person_id=person_id)
+            return None
+
+        if not self._has_active_dfo_aff(person_id):
+            logger.warning("cerebrum_not_an_employee", cerebrum_person_id=person_id)
+            return None
+
+        feide_id = self._get_feide_id(person_id)
+        if not feide_id:
+            logger.warning("cerebrum_no_primary_account", cerebrum_person_id=person_id)
+            return None
+
+        first_name, last_name = self._get_person_name(person)
+        if not first_name or not last_name:
+            logger.warning("cerebrum_no_valid_name", cerebrum_person_id=person_id)
+            return None
+
+        try:
+            sponsor = Sponsor.objects.get(feide_id=feide_id)
+            sponsor.first_name = first_name
+            sponsor.last_name = last_name
+            sponsor.save()
+            logger.info(
+                "sponsor_updated", sponsor=sponsor.id, cerebrum_person_id=person_id
+            )
+
+        except Sponsor.DoesNotExist:
+            sponsor = Sponsor.objects.create(
+                first_name=first_name,
+                last_name=last_name,
+                feide_id=feide_id,
+            )
+            logger.info(
+                "sponsor_created", sponsor=sponsor.id, cerebrum_person_id=person_id
+            )
+        return Sponsor.objects.get(id=sponsor.id)
+
+    def handle(self, *args, **options):
+        """Import of Sponsors from Cerebrum."""
+        active_units = OrganizationalUnit.objects.filter(
+            active=True,
+            deleted=False,
+        )
+
+        logger.info("import_start", nr_of_units=len(active_units))
+        for unit in active_units:
+            logger.bind(unit=unit.id)
+            sko = unit.identifiers.filter(name="legacy_stedkode").first()
+            if not sko:
+                logger.warning("orgreg_unit_missing_legacy_stedkode")
+                continue
+            logger.bind(legacy_stedkode=sko.value)
+
+            current_sponsors = unit.link_unit.filter(
+                automatic=True, source=self.CEREBRUM_SOURCE
+            ).all()
+
+            group_name = f"adm-leder-{sko.value}"
+            group = self.client.get_group(group_name)
+            if not group:
+                # No group in cererbum, remove sponsors.
+                logger.info(
+                    "cerebrum_group_not_found",
+                    unit_id=unit.id,
+                    cerebrum_group=group_name,
+                )
+                for sponsor in current_sponsors:
+                    self._remove_sponsor_unit_link(sponsor)
+                continue
+
+            if group.expire_date:
+                # Group is expired, remove sponsors
+                logger.info(
+                    "cerebrum_group_expired",
+                    unit_id=unit.id,
+                    cerebrum_group=group_name,
+                )
+                for sponsor in current_sponsors:
+                    self._remove_sponsor_unit_link(sponsor)
+                continue
+
+            group_members = list(self.client.list_group_members(group_name))
+            if not group_members:
+                # No members in group, remove sponsors
+                logger.info(
+                    "cerebrum_group_empty", unit_id=unit.id, cerebrum_group=group_name
+                )
+                for sponsor in current_sponsors:
+                    self._remove_sponsor_unit_link(sponsor)
+                continue
+
+            cerebrum_sponsors = set()
+            for member in group_members:
+                if member.type == "person":
+                    sponsor = self._upsert_sponsor_from_cerebrum(member.id, unit)
+                    if sponsor:
+                        sponsor_link = self._upsert_sponsor_unit_link(
+                            sponsor=sponsor, unit=unit
+                        )
+                        cerebrum_sponsors.add(sponsor_link)
+
+            for sponsor in set(current_sponsors) - cerebrum_sponsors:
+                self._remove_sponsor_unit_link(sponsor)
+
+        logger.info("import_end")