main: add admin control

main.py

@@ -21,6 +21,25 @@ class Config(BaseModel):
 def signal_timestamp_to_datetime(timestamp: str) -> datetime.datetime:
     return datetime.datetime.fromtimestamp(int(timestamp)/1000)
 
+T = TypeVar("T")
+
+class Changes[T](BaseModel):
+    to_add: List[T] = []
+    to_remove: List[T] = []
+
+def necessary_changes(current: List[T], desired: List[T]) -> Changes[T]:
+    changes = Changes[T]()
+
+    for item in desired:
+        if item not in current:
+            changes.to_add.append(item)
+
+    for item in current:
+        if item not in desired:
+            changes.to_remove.append(item)
+
+    return changes
+
 class SignalAPI:
 
     def __init__(self, apiurl, number):
@@ -74,6 +93,9 @@ class SignalAPI:
         # {
         #    "members": ["+49123456789", "+49123456780"]
         # }
+        if len(numbers_to_add) == 0:
+            return Ok(None)
+
         r = requests.post(f"{self.apiurl}/v1/groups/{self.number}/{group_id}/members", json={"members": numbers_to_add})
 
         if r.status_code == 204:
@@ -87,6 +109,9 @@ class SignalAPI:
         # {
         #    "members": ["+49123456789", "+49123456780"]
         # }
+        if len(numbers_to_remove) == 0:
+            return Ok(None)
+
         r = requests.delete(f"{self.apiurl}/v1/groups/{self.number}/{group_id}/members", json={"members": numbers_to_remove})
 
         if r.status_code == 204:
@@ -94,38 +119,71 @@ class SignalAPI:
         else:
             return Err("Failed to remove group members.")
 
-    def update_group_members(self, group_id: str, other_members: List[str]) -> Result[UpdateGroupResult, str]:
+    def update_group_members(self, group_id: str, other_members: List[str], remove: bool = True) -> Result[UpdateGroupResult, str]:
+        # Add and remove members from the group, such that the only remaining
+        # members of the group are this bot and users specified by other_members.
         group = self.get_group(group_id)
         if is_err(group):
             return Err(group.unwrap_err())
 
         current_members = group.unwrap().members
 
-        members_to_add = []
-        members_to_remove = []
+        members_should = [self.number] + other_members
+        member_changes = necessary_changes(current_members, members_should)
 
-        for member in other_members:
-            if member not in current_members:
-                members_to_add.append(member)
-
-        for member in current_members:
-            if member not in other_members:
-                members_to_remove.append(member)
-
-        if self.number in members_to_remove:
-            members_to_remove.remove(self.number)
-
-        if len(members_to_add) > 0:
-            add_result = self.add_group_members(group_id, members_to_add)
+        add_result = self.add_group_members(group_id, member_changes.to_add)
         if add_result.is_err():
             return Err(add_result.unwrap_err())
 
-        if len(members_to_remove) > 0:
-            remove_result = self.remove_group_members(group_id, members_to_remove)
+        if remove:
+            remove_result = self.remove_group_members(group_id, member_changes.to_remove)
             if remove_result.is_err():
                 return Err(remove_result.unwrap_err())
 
-        return Ok(UpdateGroupResult(members_added=members_to_add, members_removed=members_to_remove))
+        return Ok(member_changes)
+
+    def add_group_admins(self, group_id: str, admins: List[str]) -> Result[None, str]:
+        if len(admins) == 0:
+            return Ok(None)
+
+        r = requests.post(f"{self.apiurl}/v1/groups/{self.number}/{group_id}/admins", json={"admins": admins})
+
+        if r.status_code == 204:
+            return Ok(None)
+        else:
+            return Err(r.json().error)
+
+    def remove_group_admins(self, group_id: str, admins: List[str]) -> Result[None, str]:
+        r = requests.delete(f"{self.apiurl}/v1/groups/{self.number}/{group_id}/admins", json={"admins": admins})
+
+        if r.status_code == 204:
+            return Ok(None)
+        else:
+            return Err(r.json().error)
+
+    def update_group_admins(self, group_id: str, other_admins: List[str], remove: bool = True) -> Result[Changes[str], str]:
+        # Add and remove admins from the group, such that the only remaining
+        # admins of the group are this bot and users specified by other_admins.
+        group = self.get_group(group_id)
+
+        if is_err(group):
+            return Err(group.unwrap_err())
+
+        admins_should = [self.number] + other_admins
+        current_admins = group.unwrap().admins
+
+        admin_changes = necessary_changes(current_admins, admins_should)
+
+        add_result = self.add_group_admins(group_id, admin_changes.to_add)
+        if add_result.is_err():
+            return Err(add_result.unwrap_err())
+
+        if remove:
+            remove_result = self.remove_group_admins(group_id, admin_changes.to_remove)
+            if remove_result.is_err():
+                return Err(remove_result.unwrap_err())
+
+        return Ok(admin_changes)
 
     def get_identities(self) -> Result[List[IdentityEntry], str]:
         r = requests.get(f"{self.apiurl}/v1/identities/{self.number}")
@@ -164,6 +222,8 @@ class SignalAPI:
 
 
 class LabCleaningBot:
+    api: SignalAPI
+    base_group: str
 
     def __init__(self, api, base_group):
         self.api = api
@@ -314,6 +374,7 @@ class LabCleaningBot:
             sync_result = self.sync_members_as_active_users(session)
 
             self.api.update_group_members(self.base_group, ["+4915773232355"])
+            self.api.update_group_admins(self.base_group, ["+4915773232355"], remove=False)
 
             if is_err(sync_result):
                 print(sync_result.unwrap_err())