[mastodon] First approach to a mastodon bot

backend/setup.cfg

@@ -34,6 +34,7 @@ install_requires =
     pytoml
     requests
     scrypt
+    Mastodon.py
 
 [options.packages.find]
 where = src

kibicara/platforms/mastodon/bot.py

@@ -0,0 +1,70 @@
+# Copyright (C) 2020 by Thomas Lindner <tom@dl6tom.de>
+# Copyright (C) 2020 by Cathy Hu <cathy.hu@fau.de>
+# Copyright (C) 2020 by Martin Rey <martin.rey@mailbox.org>
+#
+# SPDX-License-Identifier: 0BSD
+
+from kibicara.platformapi import Censor, Spawner, Message
+from kibicara.platforms.mastodon.model import MastodonAccount
+
+from mastodon import Mastodon, MastodonError
+from asyncio import gather
+import sys
+
+from logging import getLogger
+
+logger = getLogger(__name__)
+
+
+class MastodonBot(Censor):
+    def __init__(self, mastodon_account_model):
+        super().__init__(mastodon_account_model.hood)
+        self.model = mastodon_account_model
+        self.enabled = self.model.enabled
+        self.account = Mastodon(
+            client_id=self.model.instance_id.client_id,
+            client_secret=self.model.instance_id.client_secret,
+            access_token=self.model.access_token,
+        )
+
+    async def run(self):
+        await gather(self.poll(), self.push())
+
+    async def poll(self):
+        """Get new mentions and DMs from Mastodon"""
+        while True:
+            try:
+                notifications = self.account.notifications()
+            except MastodonError:
+                logger.warning(
+                    "%s in hood %s" % (sys.exc_info()[0], self.model.hood.name)
+                )
+                continue
+            last_seen = int(self.model.last_seen)
+            for status in notifications:
+                status_id = int(status['status']['id'])
+                if status_id <= last_seen:
+                    continue  # toot was already processed in the past
+                if status_id > self.model.last_seen:
+                    self.model.last_seen = status_id  # save last_seen in database
+                text = status['status']['content']
+                # sanitize toot content; see ticketfrei2 for regex magic
+                logger.debug(
+                    "Mastodon in %s received message: " % (self.model.hood.name,)
+                )
+                if status['status']['visibility'] == 'public':
+                    await self.publish(Message(text, toot_id=status_id))
+                else:
+                    await self.publish(Message(text))
+
+    async def push(self):
+        """Push new Ticketfrei reports to Mastodon; if source is mastodon, boost it."""
+        while True:
+            message = await self.receive()
+            if hasattr(message, "tood_id"):
+                await self.account.status_reblog(message.tood_id)
+            else:
+                await self.account.status_post(message.text)
+
+
+spawner = Spawner(MastodonAccount, MastodonBot)

kibicara/platforms/mastodon/model.py

@@ -0,0 +1,30 @@
+# Copyright (C) 2020 by Thomas Lindner <tom@dl6tom.de>
+# Copyright (C) 2020 by Martin Rey <martin.rey@mailbox.org>
+#
+# SPDX-License-Identifier: 0BSD
+
+from ormantic import ForeignKey, Integer, Text, Boolean, Model
+
+from kibicara.model import Hood, Mapping
+
+
+class MastodonInstance(Model):
+    id: Integer(primary_key=True) = None
+    name: Text()
+    client_id: Text()
+    client_secret: Text()
+
+    class Mapping(Mapping):
+        table_name = 'mastodoninstances'
+
+
+class MastodonAccount(Model):
+    id: Integer(primary_key=True) = None
+    hood: ForeignKey(Hood)
+    instance_id: ForeignKey(MastodonInstance)
+    access_token: Text()
+    enabled: Boolean() = False
+    last_seen: Text()
+
+    class Mapping(Mapping):
+        table_name = 'mastodonaccounts'