86 lines
No EOL
2.7 KiB
Python
86 lines
No EOL
2.7 KiB
Python
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 ConsumerNTFY(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 NTFY
|
|
|
|
url = self.config["server_url"] + "/" + self.config["server_topic"]
|
|
body = ' '.join(tokens[1:])
|
|
if len(self.config["allowed_locations"]) != 1:
|
|
body = tokens[0] + ": " + body
|
|
|
|
# Consider authentication
|
|
authentication = None
|
|
if self.config["server_use_authentication"]:
|
|
authentication = aiohttp.BasicAuth(self.config["server_username"], self.config["server_password"])
|
|
|
|
# Send notification
|
|
for trash in range(3): # Try to send 3 times
|
|
async with self.http.post(url, data=body, auth=authentication) as response:
|
|
if response.status == 200:
|
|
await evt.react("👍")
|
|
return
|
|
|
|
|
|
|
|
# That's all, folks |