Make aardbei_sync actually work

piket_server/aardbei_sync.py

 from __future__ import annotations
 
-from typing import List, Dict, Any, Tuple, Optional
-from dataclasses import dataclass
+import datetime
+import json
 import logging
+from dataclasses import asdict, dataclass
+from enum import Enum
+from typing import Any, Dict, List, NewType, Optional, Tuple
 
 import requests
 
-from piket_server import Person, db
+from piket_server.models import Person
+from piket_server.flask import db
+
+# AARDBEI_ENDPOINT = "https://aardbei.app"
+AARDBEI_ENDPOINT = "http://localhost:3000"
+
+ActivityId = NewType("ActivityId", int)
+PersonId = NewType("PersonId", int)
+MemberId = NewType("MemberId", int)
+ParticipantId = NewType("ParticipantId", int)
 
 
 @dataclass(frozen=True)
-class SparseAardbeiPerson:
+class AardbeiPerson:
+    aardbei_id: PersonId
     full_name: str
-    display_name: str
-    aardbei_id: int
+
+    @classmethod
+    def from_aardbei_dict(cls, data: Dict[str, Any]) -> AardbeiPerson:
+        d = data["person"]
+        return cls(full_name=d["full_name"], aardbei_id=PersonId(d["id"]))
+
+
+@dataclass(frozen=True)
+class AardbeiMember:
+    person: AardbeiPerson
+    aardbei_id: MemberId
     is_leader: bool
+    display_name: str
+
+    @classmethod
+    def from_aardbei_dict(cls, data: Dict[str, Any]) -> AardbeiMember:
+        logging.debug("Init with data %s", json.dumps(data))
+        d = data["member"]
+        person = AardbeiPerson.from_aardbei_dict(d)
+        return cls(
+            person=person,
+            aardbei_id=MemberId(d["id"]),
+            is_leader=d["is_leader"],
+            display_name=d["display_name"],
+        )
+
+
+@dataclass(frozen=True)
+class AardbeiParticipant:
+    person: AardbeiPerson
+    member: Optional[AardbeiMember]
+    aardbei_id: ParticipantId
+    attending: bool
+    is_organizer: bool
+    notes: Optional[str]
+
+    @property
+    def name(self) -> str:
+        if self.member is not None:
+            return self.member.display_name
+
+        return self.person.full_name
+
+    @classmethod
+    def from_aardbei_dict(cls, data: Dict[str, Any]) -> AardbeiParticipant:
+        d = data["participant"]
+        person = AardbeiPerson.from_aardbei_dict(d)
+
+        member: Optional[AardbeiMember] = None
+        if d["member"] is not None:
+            member = AardbeiMember.from_aardbei_dict(d)
+
+        aardbei_id = ParticipantId(d["id"])
+
+        return cls(
+            person=person,
+            member=member,
+            aardbei_id=aardbei_id,
+            attending=d["attending"],
+            is_organizer=d["is_organizer"],
+            notes=d["notes"],
+        )
+
+
+class NoResponseAction(Enum):
+    Present = "present"
+    Absent = "absent"
+
+
+@dataclass(frozen=True)
+class ResponseCounts:
+    present: int
+    absent: int
+    unknown: int
 
     @classmethod
-    def from_aardbei_dict(cls, data: Dict[str, Any]) -> SparseAardbeiPerson:
+    def from_aardbei_dict(cls, data: Dict[str, int]) -> ResponseCounts:
         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"],
+            present=data["present"], absent=data["absent"], unknown=data["unknown"]
         )
 
 
 @dataclass(frozen=True)
+class SparseAardbeiActivity:
+    aardbei_id: ActivityId
+    name: str
+    description: str
+    location: str
+    start: datetime.datetime
+    end: Optional[datetime.datetime]
+    deadline: Optional[datetime.datetime]
+    reminder_at: Optional[datetime.datetime]
+    no_response_action: NoResponseAction
+    response_counts: ResponseCounts
+
+    def distance(self, reference: datetime.datetime) -> datetime.timedelta:
+        """Calculate how long ago this Activity ended / how much time until it starts."""
+        if self.end is not None:
+            if reference > self.start and reference < self.end:
+                return datetime.timedelta(seconds=0)
+
+            elif reference < self.start:
+                return self.start - reference
+
+            elif reference > self.end:
+                return reference - self.end
+
+        if reference > self.start:
+            return reference - self.start
+
+        return self.start - reference
+
+    @classmethod
+    def from_aardbei_dict(cls, data: Dict[str, Any]) -> SparseAardbeiActivity:
+        start: datetime.datetime = datetime.datetime.fromisoformat(
+            data["activity"]["start"]
+        )
+        end: Optional[datetime.datetime] = None
+
+        if data["activity"]["end"] is not None:
+            end = datetime.datetime.fromisoformat(data["activity"]["end"])
+
+        deadline: Optional[datetime.datetime] = None
+        if data["activity"]["deadline"] is not None:
+            deadline = datetime.datetime.fromisoformat(data["activity"]["deadline"])
+
+        reminder_at: Optional[datetime.datetime] = None
+        if data["activity"]["reminder_at"] is not None:
+            reminder_at = datetime.datetime.fromisoformat(
+                data["activity"]["reminder_at"]
+            )
+
+        no_response_action = NoResponseAction(data["activity"]["no_response_action"])
+
+        response_counts = ResponseCounts.from_aardbei_dict(
+            data["activity"]["response_counts"]
+        )
+
+        return cls(
+            aardbei_id=ActivityId(data["activity"]["id"]),
+            name=data["activity"]["name"],
+            description=data["activity"]["description"],
+            location=data["activity"]["location"],
+            start=start,
+            end=end,
+            deadline=deadline,
+            reminder_at=reminder_at,
+            no_response_action=no_response_action,
+            response_counts=response_counts,
+        )
+
+
+@dataclass(frozen=True)
+class AardbeiActivity(SparseAardbeiActivity):
+    participants: List[AardbeiParticipant]
+
+    @classmethod
+    def from_aardbei_dict(cls, data: Dict[str, Any]) -> AardbeiActivity:
+        participants: List[AardbeiParticipant] = [
+            AardbeiParticipant.from_aardbei_dict(x)
+            for x in data["activity"]["participants"]
+        ]
+
+        sparse = super().from_aardbei_dict(data)
+
+        return cls(participants=participants, **asdict(sparse))
+
+
+@dataclass(frozen=True)
 class AardbeiMatch:
     local: Person
-    remote: SparseAardbeiPerson
+    remote: AardbeiMember
 
 
 @dataclass(frozen=True)
     """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]
+    remote_only: List[AardbeiMember]
     """People that only exist on the remote."""
 
 
-def get_aardbei_people(token: str) -> List[SparseAardbeiPerson]:
+def get_aardbei_people(token: str) -> List[AardbeiMember]:
     resp = requests.get(
-        "https://aardbei.app/api/groups/0/", headers={"Authorization": f"Group {token}"}
+        f"{AARDBEI_ENDPOINT}/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]
+    return [AardbeiMember.from_aardbei_dict(x) for x in members]
 
 
-def match_local_aardbei(aardbei_people: List[SparseAardbeiPerson]) -> AardbeiLink:
+def match_local_aardbei(aardbei_members: List[AardbeiMember]) -> AardbeiLink:
     matches: List[AardbeiMatch] = []
     altered_name: List[AardbeiMatch] = []
-    remote_only: List[SparseAardbeiPerson] = []
+    remote_only: List[AardbeiMember] = []
 
-    for aardbei_person in aardbei_people:
+    for member in aardbei_members:
         p: Optional[Person] = Person.query.filter_by(
-            aardbei_id=aardbei_person.aardbei_id
+            aardbei_id=member.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
+                p.full_name != member.person.full_name
+                or p.display_name != member.display_name
             ):
-                altered_name.append(AardbeiMatch(p, aardbei_person))
+                altered_name.append(AardbeiMatch(p, member))
 
             else:
                 logging.info(
 
             continue
 
-        p = Person.query.filter_by(full_name=aardbei_person.full_name).one_or_none()
+        p = Person.query.filter_by(full_name=member.person.full_name).one_or_none()
 
         if p is not None:
-            matches.append(AardbeiMatch(p, aardbei_person))
+            matches.append(AardbeiMatch(p, member))
         else:
-            remote_only.append(aardbei_person)
+            remote_only.append(member)
 
     return AardbeiLink(matches, altered_name, remote_only)
 
         db.session.add(match.local)
 
 
-def create_missing(missing: List[SparseAardbeiPerson]) -> None:
-    for person in missing:
+def create_missing(missing: List[AardbeiMember]) -> None:
+    for member in missing:
         pnew = Person(
-            full_name=person.full_name,
-            display_name=person.display_name,
-            aardbei_id=person.aardbei_id,
+            full_name=member.person.full_name,
+            display_name=member.display_name,
+            aardbei_id=member.aardbei_id,
             active=False,
         )
         logging.info(
-            "Creating new person for %s (%s)", person.full_name, person.aardbei_id
+            "Creating new person for %s / %s (%s)",
+            member.person.full_name,
+            member.display_name,
+            member.aardbei_id,
         )
         db.session.add(pnew)
 
 def update_names(matches: List[AardbeiMatch]) -> None:
     for match in matches:
         p = match.local
-        aardbei_person = match.remote
+        member = match.remote
+        aardbei_person = member.person
 
         changed = False
 
             p.full_name = aardbei_person.full_name
             changed = True
 
-        if p.display_name != aardbei_person.display_name:
+        if p.display_name != member.display_name:
             logging.info(
                 "Updating %s (L%s/R%s) display name %s to %s",
-                aardbei_person.full_name,
+                p.full_name,
                 p.person_id,
                 aardbei_person.aardbei_id,
                 p.display_name,
-                aardbei_person.display_name,
+                member.display_name,
             )
-            p.display_name = aardbei_person.display_name
-
+            p.display_name = member.display_name
             changed = True
 
         assert changed, "got match but didn't update anything"
         db.session.add(p)
 
 
+def get_activities(token: str) -> List[SparseAardbeiActivity]:
+    result: List[SparseAardbeiActivity] = []
+
+    for category in ("upcoming", "current", "previous"):
+        resp = requests.get(
+            f"{AARDBEI_ENDPOINT}/api/groups/0/{category}_activities",
+            headers={"Authorization": f"Group {token}"},
+        )
+
+        resp.raise_for_status()
+
+        for item in resp.json():
+            result.append(SparseAardbeiActivity.from_aardbei_dict(item))
+
+    return result
+
+
 if __name__ == "__main__":
-    logging.basicConfig(level=logging.INFO)
+    logging.basicConfig(level=logging.DEBUG)
 
     token = input("Token: ")
     aardbei_people = get_aardbei_people(token)
+    activities = get_activities(token)
     link = match_local_aardbei(aardbei_people)
     link_matches(link.matches)
     create_missing(link.remote_only)