fediverse-admin-bot/bot.py

import time
import os
import sys
import logging
import random
from configparser import ConfigParser
from pathlib import Path
from misskey import Misskey
from misskey.exceptions import MisskeyAPIException

BASE = Path.cwd()
if (BASE / "data").exists():
    BASE = BASE / "data"
    print("Data/ directory found. Using it as base.")


logging.basicConfig(
    filename=BASE / "bot.log",
    filemode="a",
    level=getattr(logging, os.getenv("LOGLEVEL", "INFO")),
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger("fedibot")

config = ConfigParser()
config.read(str(BASE / "config.ini"))

server_domain = list(config.keys())[1]
api_key = config[server_domain]["api_key"]
sleep_time = int(config[server_domain].get("sleep_time", 60))
limit = int(config[server_domain].get("limit", 100))

bot = Misskey(server_domain, api_key)

with open(BASE / "copy.txt") as _fd:
    copy = _fd.read()


def main():
    print("Bot running.")
    while True:
        try:
            with open(BASE / ".lastfetch", "r") as fd:
                lastfetch = fd.read()
                logger.debug(f"Last fetch: {lastfetch}")
        except FileNotFoundError:
            lastfetch = os.getenv("EPOCH_ID")
            logger.warning(f"No epoch for last fetch found. Using $EPOCH_ID: {lastfetch}")

        for i in range(5):
            try:
                notifications = bot.i_notifications(
                    limit=100,
                    mark_as_read=True,
                    include_types=["reply", "mention"],
                    since_id=lastfetch,
                )
            except MisskeyAPIException as e:
                logger.error("Failed to fetch notifications: %r", e, exc_info=True)
                retry = random.uniform(0, 2**i)
                logger.warning("Retrying in %.2f seconds...", retry)
                time.sleep(retry)
            else:
                break
        else:
            logger.critical("Failed to fetch notifications after 5 retries. Exiting...")
            sys.exit(1)
        logger.debug("Fetched another batch of notifications: %r", notifications)
        for notification in notifications:
            if notification["note"]["user"].get("isBot", False) is True:
                logger.debug("Ignoring bot: %r", notification["note"]["user"]["id"])
                # Ignore bots
                continue

            logger.info("Replying to post %r", notification["note"]["id"])
            try:
                bot.notes_create(
                    text=copy.format(notification, bot),
                    reply_id=notification["note"]["id"],
                    visibility="specified",
                    visible_user_ids=[notification["user"]["id"]],
                )
                logger.info("Successfully replied to post %r", notification["note"]["id"])
                try:
                    logger.info(
                        "Reacting with the inbox emoji to post %r",
                        notification["note"]["id"],
                    )
                    bot.notes_reactions_create(note_id=notification["note"]["id"], reaction="📥")
                    logger.info(
                        "Successfully reacted with the inbox emoji to post %r",
                        notification["note"]["id"],
                    )
                except MisskeyAPIException:
                    logger.error(
                        "Failed to react with the inbox emoji to post %r",
                        notification["note"]["id"],
                        exc_info=True,
                    )

                # Save that we replied to this post
                with open(BASE / ".lastfetch", "w") as fd:
                    fd.write(notification["id"])
            except MisskeyAPIException as e:
                logger.error(
                    "Failed to reply to post %r: %r",
                    notification["note"]["id"],
                    e,
                    exc_info=True,
                )
        try:
            time.sleep(sleep_time)
        except KeyboardInterrupt:
            break


if __name__ == "__main__":
    main()