Initial commit

config.py

@@ -0,0 +1,26 @@
+import toml
+
+
+class Config:
+    command_prefix: str
+
+    matrix_homeserver: str
+    matrix_user: str
+    matrix_access_token: str
+
+    spaceping_token: str
+
+    def __init__(self, path=None):
+        if path is None:
+            path = "/etc/itsyndikat-bot.toml"
+
+        config = toml.load(path)
+
+        self.command_prefix = config['app']['command_prefix']
+
+        matrix = config['matrix']
+        self.matrix_homeserver = matrix['homeserver']
+        self.matrix_username = matrix['username']
+        self.matrix_access_token = matrix['access_token']
+
+        self.spaceping_token = config['spaceping']['api_token']

config.toml.sample

@@ -0,0 +1,10 @@
+[app]
+command_prefix = "!"
+
+[matrix]
+homeserver = ...
+username = ...
+access_token = ...
+
+[spaceping]
+api_token = ...

its_api.py

@@ -0,0 +1,24 @@
+import aiohttp
+
+from config import Config
+
+class ItSyndikatApi:
+    base_url: str
+    config: Config
+
+    def __init__(self, config: Config):
+        self.base_url = "https://it-syndikat.org/api/"
+        self.config = config
+
+    async def status(self):
+        async with aiohttp.ClientSession() as session:
+            async with session.get(self.base_url + "status.php") as response:
+                return await response.json()
+
+
+    async def ping(self):
+        params = {"apikey": self.config.spaceping_token}
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(self.base_url + "ping.php", params=params) as response:
+                await response.text()

main.py

@@ -0,0 +1,91 @@
+import argparse
+import datetime
+from nio.rooms import MatrixRoom
+from nio.events.room_events import RoomMessage
+import simplematrixbotlib as botlib
+
+from its_api import ItSyndikatApi
+from config import Config
+
+
+class ItSyndikatBot:
+    bot: botlib.Bot
+    its_api: ItSyndikatApi
+
+    config: Config
+
+    def __init__(self, config: Config):
+        self.config = config
+
+        creds = botlib.Creds(
+            config.matrix_homeserver,
+            config.matrix_username,
+            access_token=config.matrix_access_token,
+        )
+
+        self.its_api = ItSyndikatApi(config)
+
+        self.bot = botlib.Bot(creds)
+
+        self.bot.listener.on_message_event(self.on_message)
+        self.bot.run()
+
+    async def on_message(self, room, message):
+        m = botlib.MessageMatch(room, message, self.bot, self.config.command_prefix)
+
+        if m.is_not_from_this_bot() and m.prefix():
+            if m.command("echo"):
+                await self.echo(room, message, m.args())
+            elif m.command("isitopen"):
+                await self.isitopen(room, message)
+            elif m.command("spaceping"):
+                await self.spaceping(room, message)
+            else:
+                await self.bot.api.send_text_message(
+                    room.room_id, f"Unknown command: {m.command()}"
+                )
+
+    async def reply(self, room, message, reply):
+        await self.bot.api.async_client.room_send(
+            room_id=room.room_id,
+            message_type="m.room.message",
+            content={
+                "msgtype": "m.text",
+                "body": reply,
+                "m.relates_to": {"m.in_reply_to": {"event_id": message.event_id}},
+            },
+        )
+
+    async def echo(self, room, message, args):
+        await self.bot.api.send_text_message(
+            room.room_id, " ".join(arg for arg in args)
+        )
+
+    async def isitopen(self, room, message):
+        try:
+            status = await self.its_api.status()
+            if status["state"]["open"]:
+                date = datetime.datetime.fromtimestamp(status["state"]["lastchange"])
+                text = f"positive! space has been open since {date}"
+            else:
+                text = "negative!"
+        except Exception as e:
+            text = f"error checking space status: {e}"
+
+        await self.reply(room, message, text)
+
+    async def spaceping(self, room: MatrixRoom, message: RoomMessage):
+        await self.its_api.ping()
+        await self.reply(room, message, "Hello Space!")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="IT-Syndikat matrix bot")
+    parser.add_argument(
+        "-c", "--config",
+        help="path to the config file",
+    )
+
+    args = parser.parse_args()
+
+    ItSyndikatBot(Config(args.config))