Add script to sync people from Aardbei

piket_server/aardbei_sync.py

@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+from typing import List, Dict, Any, Tuple, Optional
+from dataclasses import dataclass
+import logging
+
+import requests
+
+from piket_server import Person, db
+
+
+@dataclass(frozen=True)
+class SparseAardbeiPerson:
+    full_name: str
+    display_name: str
+    aardbei_id: int
+    is_leader: bool
+
+    @classmethod
+    def from_aardbei_dict(cls, data: Dict[str, Any]) -> SparseAardbeiPerson:
+        return cls(
+            full_name=data["member"]["person"]["full_name"],
+            display_name=data["member"]["display_name"],
+            aardbei_id=data["member"]["person"]["id"],
+            is_leader=data["member"]["is_leader"],
+        )
+
+
+@dataclass(frozen=True)
+class AardbeiMatch:
+    local: Person
+    remote: SparseAardbeiPerson
+
+
+@dataclass(frozen=True)
+class AardbeiLink:
+    matches: List[AardbeiMatch]
+    """People that exist on both sides, but aren't linked in the people table."""
+    altered_name: List[AardbeiMatch]
+    """People that are already linked but changed one of their names."""
+    remote_only: List[SparseAardbeiPerson]
+    """People that only exist on the remote."""
+
+
+def get_aardbei_people(token: str) -> List[SparseAardbeiPerson]:
+    resp = requests.get(
+        "https://aardbei.app/api/groups/0/", headers={"Authorization": f"Group {token}"}
+    )
+    resp.raise_for_status()
+
+    members = resp.json()["group"]["members"]
+
+    return [SparseAardbeiPerson.from_aardbei_dict(x) for x in members]
+
+
+def match_local_aardbei(aardbei_people: List[SparseAardbeiPerson]) -> AardbeiLink:
+    matches: List[AardbeiMatch] = []
+    altered_name: List[AardbeiMatch] = []
+    remote_only: List[SparseAardbeiPerson] = []
+
+    for aardbei_person in aardbei_people:
+        p: Optional[Person] = Person.query.filter_by(
+            aardbei_id=aardbei_person.aardbei_id
+        ).one_or_none()
+
+        if p is not None:
+            if (
+                p.full_name != aardbei_person.full_name
+                or p.display_name != aardbei_person.display_name
+            ):
+                altered_name.append(AardbeiMatch(p, aardbei_person))
+
+            else:
+                logging.info(
+                    "OK: %s / %s (L%s/R%s)",
+                    p.full_name,
+                    p.display_name,
+                    p.person_id,
+                    p.aardbei_id,
+                )
+
+            continue
+
+        p = Person.query.filter_by(full_name=aardbei_person.full_name).one_or_none()
+
+        if p is not None:
+            matches.append(AardbeiMatch(p, aardbei_person))
+        else:
+            remote_only.append(aardbei_person)
+
+    return AardbeiLink(matches, altered_name, remote_only)
+
+
+def link_matches(matches: List[AardbeiMatch]) -> None:
+    for match in matches:
+        match.local.aardbei_id = match.remote.aardbei_id
+        match.local.display_name = match.remote.display_name
+        logging.info(
+            "Linking local %s (%s) to remote %s (%s)",
+            match.local.full_name,
+            match.local.person_id,
+            match.remote.display_name,
+            match.remote.aardbei_id,
+        )
+
+        db.session.add(match.local)
+
+
+def create_missing(missing: List[SparseAardbeiPerson]) -> None:
+    for person in missing:
+        pnew = Person(
+            full_name=person.full_name,
+            display_name=person.display_name,
+            aardbei_id=person.aardbei_id,
+            active=False,
+        )
+        logging.info(
+            "Creating new person for %s (%s)", person.full_name, person.aardbei_id
+        )
+        db.session.add(pnew)
+
+
+def update_names(matches: List[AardbeiMatch]) -> None:
+    for match in matches:
+        p = match.local
+        aardbei_person = match.remote
+
+        changed = False
+
+        if p.full_name != aardbei_person.full_name:
+            logging.info(
+                "Updating %s (L%s/R%s) full name %s to %s",
+                aardbei_person.full_name,
+                p.person_id,
+                aardbei_person.aardbei_id,
+                p.full_name,
+                aardbei_person.full_name,
+            )
+            p.full_name = aardbei_person.full_name
+            changed = True
+
+        if p.display_name != aardbei_person.display_name:
+            logging.info(
+                "Updating %s (L%s/R%s) display name %s to %s",
+                aardbei_person.full_name,
+                p.person_id,
+                aardbei_person.aardbei_id,
+                p.display_name,
+                aardbei_person.display_name,
+            )
+            p.display_name = aardbei_person.display_name
+
+            changed = True
+
+        assert changed, "got match but didn't update anything"
+
+        db.session.add(p)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+
+    token = input("Token: ")
+    aardbei_people = get_aardbei_people(token)
+    link = match_local_aardbei(aardbei_people)
+    link_matches(link.matches)
+    create_missing(link.remote_only)
+    update_names(link.altered_name)
+
+    confirm = input("Commit? Y/N")
+    if confirm.lower() == "y":
+        print("Committing.")
+        db.session.commit()
+    else:
+        print("Not committing.")