Add source file

src/main.py

@@ -0,0 +1,88 @@
+import os
+import pprint
+import logging
+import sys
+import toml  # Used for the config file parsing
+from mastodon import Mastodon, StreamListener
+from mastodon.errors import MastodonError
+
+logger = logging.getLogger(__name__)
+
+class BotListener(StreamListener):
+    """A bot class that listens to incoming notifications"""
+    def __init__(self, mastodon, messages):
+        super().__init__()
+        self.mastodon = mastodon
+        self.messages = messages
+        logger.debug("Creating a new bot")
+
+    def on_notification(self, notification):
+        """Listen and react on a new notification"""
+        logger.debug("Got new notification")
+        try:
+            kind = notification["type"]
+            account = notification.get("account") or {}
+            acct = account.get("acct", "unknown")
+
+            if kind == "favourite":
+                logger.info("New fav ⭐ from %s", acct)
+
+            elif kind == "admin.sign_up":
+                logger.info("New signup: %s", acct)
+                welcome = self.mastodon.status_post(
+                    status=self.messages["welcome_message"].format(acct),
+                    visibility="private",
+                )
+
+                collections = self.mastodon.status_post(
+                    status=self.messages["collections"],
+                    in_reply_to_id=welcome,
+                    visibility="private",
+                )
+
+                logger.debug("Posted welcome %s and follow-up %s",
+                             welcome["id"], collections["id"])
+
+            elif kind == "follow":
+                logger.info("New follow from %s", acct)
+
+            else:
+                logger.debug("Notification type: %s", kind)
+
+        except Exception:
+            # Keep one failed response from interrupting notification handling.
+            logger.exception("Failed to handle notification")
+
+
+def main():
+    logging.basicConfig(
+        filename="botato.log",
+        level=logging.DEBUG,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+
+    try:
+        with open("config/botato.toml", "r", encoding="utf-8") as f:
+            configuration = toml.load(f)
+    except (OSError, toml.TomlDecodeError, TypeError):
+        logger.exception("Unable to load config/configuration file")
+        sys.exit(1)
+
+    mastodon = Mastodon(
+        api_base_url=configuration["mastodon"]["instance"],
+        access_token=configuration["mastodon"]["access_token"],
+    )
+
+    listener = BotListener(mastodon, configuration["messages"])
+
+    try:
+        # Blocking call: runs until interrupted or the connection fails.
+        logger.info("Starting bot...")
+        mastodon.stream_user(listener)
+    except KeyboardInterrupt:
+        logger.info("Stopping bot")
+    except MastodonError:
+        logger.exception("Mastodon streaming error")
+        raise
+if __name__ == "__main__":
+    main()