import typing from datetime import datetime, timedelta from .Config import Config from misskey import Misskey as lib_Misskey from misskey.enum import NotificationsType, NoteVisibility NOTES_FETCH_LIMIT = 100 # It is recommended to keep the timeline fetch limit low as every note that a user WOULD reply to in their timeline will make an API call even if a user WON'T reply TIMELINE_FETCH_LIMIT = 5 # Throttle note timeline to prevent instant reactions on runtime TIMELINE_THROTTLE_SECONDS = 300 class Misskey(lib_Misskey): def __init__(self, key: str): super().__init__(Config().get_misskey_server(), i=key) # Reverse lists returned by Misskey.py because for some stupid reason they're ordered oldest-to-newest @staticmethod def _reverse(x: list) -> list: x.reverse() return x def resolve_user_id(self, username: str) -> str: return self.users_show(username=username)["id"] # Return filtered Misskey home timeline for user def get_timeline(self) -> list: return Misskey._reverse(self.notes_timeline( limit=TIMELINE_FETCH_LIMIT, with_files=False, include_my_renotes=False, include_renoted_my_notes=False, # Don't fetch notes older than 24 hours since_date=datetime.now() - timedelta(days=1), # Don't fetch notes newer than threshold until_date=datetime.now() - timedelta(seconds=TIMELINE_THROTTLE_SECONDS) )) def get_user_notes(self, username: str) -> list: return self.users_notes( # Resolve user id from username user_id=self.resolve_user_id(username), limit=NOTES_FETCH_LIMIT ) # Return replies for a given note def get_replies(self, note: dict) -> list: return self.notes_replies( note_id=note["id"], limit=NOTES_FETCH_LIMIT ) def post_note(self, text: str, username: str = None) -> dict: # Set note visibility to specified (DM) if a username is provided visibility = NoteVisibility.SPECIFIED if username else NoteVisibility.PUBLIC return self.notes_create( text=text, visibility=visibility.value, # Resolve user id from username if note is a DM visible_user_ids=[self.resolve_user_id(username)] if username else None ) # Send a reply to a given note def post_reply(self, text: str, note: dict) -> dict: visibility = NoteVisibility(note["visibility"]) # Tag the sender in our reply text = "@{} {}".format(note["user"]["username"], text) return self.notes_create( text=text, visibility=visibility.value, visible_user_ids=[note["user"]["id"]] if visibility == NoteVisibility.SPECIFIED else None, reply_id=note["id"] ) # Send a reaction to a given note def post_reaction(self, reaction: str, note: dict) -> bool: return self.notes_reactions_create( note_id=note["id"], reaction=reaction ) # Send a follow request to username def follow_user(self, username: str) -> dict: return self.following_create( user_id=self.resolve_user_id(username) )