diff --git a/main.py b/main.py index 7220ed0..9ee3b94 100644 --- a/main.py +++ b/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) + add_result = self.add_group_members(group_id, member_changes.to_add) + if add_result.is_err(): + return Err(add_result.unwrap_err()) - 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) - 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())