118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
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()
|