from maubot import Plugin, MessageEvent from maubot.handlers import command # Needed for configuration from typing import Type from mautrix.util.config import BaseProxyConfig, ConfigUpdateHelper import aiohttp # Ensures a running instance gets an updated config from the Maubot interface class Config(BaseProxyConfig): def do_update(self, helper: ConfigUpdateHelper) -> None: helper.copy("command_prefix") helper.copy("allowed_locations_enabled") helper.copy("allowed_locations") helper.copy("allowed_senders") helper.copy("server_url") helper.copy("server_topic") helper.copy("command_prefix") helper.copy("server_use_authentication") helper.copy("server_username") helper.copy("server_password") helper.copy("send_reaction") class ConsumerBriar(Plugin): # Get configuration at startup async def start(self) -> None: self.config.load_and_update() # Get config @classmethod def get_config_class(cls) -> Type[BaseProxyConfig]: return Config # Get !command_name setting from config to register it def get_command_name(self) -> str: return self.config["command_prefix"] # What gets called when !command_name message is sent @command.new(name=get_command_name, help="Report Something") @command.argument("message", pass_raw=True) async def report(self, evt: MessageEvent, message: str) -> None: # Split command (minus !command_name) into tokens tokens = message.split() st_desig = tokens[0].lower() # Each command must have a state/territory designation and a message if len(tokens) < 2: return # Check locations whitelist if self.config["allowed_locations_enabled"]: if not tokens[0].lower() in self.config["allowed_locations"]: return # Check allowed senders allowed_senders = self.config["allowed_senders"] print(evt.sender) if st_desig in self.config["allowed_senders"]: print(f"Allowed senders for {st_desig}: {self.config['allowed_senders'].get(st_desig, [])}") if not evt.sender in self.config['allowed_senders'].get(st_desig, []): return # Sending notification to Briar sent = None # Indicates after the following loop if something has been sent, # AND that it has been sent to all contacts # Loop through each user for userId in self.config["brair_sendto"]: # Reset sent status first time, as we want to detect if if sent is None: sent = True url = self.config["brair_url"] + "/v1/messages/" + str(userId) body = ' '.join(tokens[1:]) if len(self.config["allowed_locations"]) != 1: body = tokens[0] + ": " + body body = '{"text":"' + body + '"}' headers = {'Authorization': 'Bearer ' + self.config["brair_token"]} async with self.http.post(url, data=body, auth=authentication, headers=headers) as response: if not response.status == 200: sent = False if self.config.["send_reaction"]: if sent is True: await evt.react("👍") if sent is False: await evt.react("👎") # That's all, folks