lab-signal-bot/main.py

#!/usr/bin/env python3

import requests
import json
from result import Result, Ok, Err
from typing import List
from websockets.asyncio.client import connect
import websockets
import asyncio
from apitypes import *
from pydantic import BaseModel

class Config(BaseModel):
    apiurl: str
    number: str

class SignalAPI:

    def __init__(self, apiurl, number):
        self.apiurl = apiurl
        self.number = number

    def set_username(self, username) -> Result[UsernameSetResponse, str]:
        # post request to set username
        # /v1/accounts/{number}/username
        # {
        # "username": "test"
        # }
        r = requests.post(f"{self.apiurl}/v1/accounts/{self.number}/username", json={"username": username})

        if r.status_code == 201:
            return parse_response(UsernameSetResponse, r.text)
        else:
            print("Failed to set username")
            print(r.json())

    def get_groups(self) -> Result[List[GroupEntry], str]:
        # get request to get groups
        # /v1/groups
        r = requests.get(f"{self.apiurl}/v1/groups/{self.number}")

        if r.status_code == 200:
            return parse_response(List[GroupEntry], r.text)
        else:
            print("Failed to get groups")
            print(r.json())

    def get_group(self, group_id: str) -> Result[GroupEntry, str]:
        # get request to get group
        # /v1/groups/{group_id}
        r = requests.get(f"{self.apiurl}/v1/groups/{self.number}/{group_id}")

        if r.status_code == 200:
            return parse_response(GroupEntry, r.text)
        else:
            print("Failed to get group")
            print(r.json())

    def get_identities(self) -> Result[List[IdentityEntry], str]:
        r = requests.get(f"{self.apiurl}/v1/identities/{self.number}")

        if r.status_code == 200:
            return parse_response(List[IdentityEntry], r.text)
        else:
            print("Failed to get identities")
            print(r.json())

    # /v1/identities/{number}/trust/{numberToTrust}
    def trust_identity(self, number_to_trust: str, trust_identity_request: TrustIdentityRequest) -> Result[None, str]:
        r = requests.put(f"{self.apiurl}/v1/identities/{self.number}/trust/{number_to_trust}", trust_identity_request.model_dump_json())

        if r.status_code == 204:
            return Ok(None)
        else:
            print("Failed to trust identity")
            print(r.text)

    async def receive_messages(self):
        ws_apiurl = self.apiurl.replace("http", "ws")

        async for websocket in connect(f"{ws_apiurl}/v1/receive/{config.number}"):
            try:
                print(parse_response(Message, await websocket.recv()))
                #print(await websocket.recv())
            except websockets.exceptions.ConnectionClosed:
                print("Websockets connection closed. Reestablishing connection.")

class User(BaseModel):
    number: str
    name: str
    pause_until: int | None = None

class UserList:
    members: List[str]

class DataStore(BaseModel):
    version: int = 1
    user_list: UserList



#def test_user_list():


class LabCleaningBot:

    def __init__(self, api, base_group):
        self.api = api
        self.base_group = base_group

    def get_other_members(self) -> Result[List[str], str]:
        group = self.api.get_group(self.base_group)
        if group.is_err():
            return Err(group.unwrap_err())

        members = group.unwrap().members
        other_members = []

        for member in members:
            if member != self.api.number:
                other_members.append(member)

        return Ok(other_members)



if __name__ == "__main__":

    with open("config.json", "r") as f:
        config = Config.model_validate(json.load(f))

    api = SignalAPI(config.apiurl, config.number)
    #print(api.set_username("leinelabbot"))

    #groups = api.get_groups()
    #print(groups)

    #group = api.get_group("group.bm5KT3NJUW5FdkpRNnR2ZGRFa01oOVZBeUYrVkdnd3NNTzFpNWdsR2pwUT0=")
    #print(group)

    #identities = api.get_identities()
    #print(identities)

    #trust = api.trust_identity("+4915773232355", TrustAllKnownKeys())
    #print(trust)

    #receive_messages = api.receive_messages(1)
    #print(receive_messages)

    bot = LabCleaningBot(api, "group.bm5KT3NJUW5FdkpRNnR2ZGRFa01oOVZBeUYrVkdnd3NNTzFpNWdsR2pwUT0=")
    print(bot.get_other_members())

    #asyncio.run(api.receive_test())