missclear/missclear.py

#!/usr/bin/env python3
#
# usage: ./missclear.py
#
# WARNING: This tool doesn't prompt for approval. Once you approve the
# app in Misskey, it'll start deleting right away!
# 

from Misskey import Misskey, Util, Exceptions # pip install Misskey.py
import webbrowser
import time
import datetime
import os
import sys
import argparse

SLEEP_TIME=2                # how many seconds to wait between each notes/delete call
BAD_SLEEP_TIME=15           # how many seconds to wait when things go bad

def get_cache_dir():
    cache_path = os.getenv('XDG_CACHE_DIR')
    if cache_path is None or len(cache_path) == 0:
        home_dir = os.getenv('HOME')
        if home_dir is None or len(home_dir) == 0:
            print("ERR: both $XDG_CACHE_DIR and $HOME are unset, not caching tokens...", file=sys.stderr)
            return None

        cache_path = os.path.join(home_dir, '.cache')

    return os.path.join(cache_path, 'missclear')


class Missclear(object):
    """ Main Missclear module """
    instance = None
    quiet = False
    cache_token=True
    retry_count=None
    interactive=True

    def __init__(self, args):
        self.instance = args.instance[0]
        self.quiet = args.quiet
        self.cache_token = args.cache
        self.retry_count = args.retries
        self.interactive = args.interactive

        self.print("instance=" + self.instance + ", quiet=" + str(self.quiet) +
                   ", caching=" + str(self.cache_token) + ", retries=" + str(self.retry_count) +
                   ", interactive=" + str(self.interactive))

        self.api = Misskey(self.instance)


    def use_cached_token(self):
        """ imports the cached token, if available and requested """
        if self.cache_token is False:
            return False

        token_file = os.path.join(get_cache_dir(), self.instance)
        if os.path.exists(token_file):
            with open(token_file) as f:
                token_data = f.read()

            try:
                self.api.apiToken = token_data
            except Exceptions.MisskeyAPITokenException as err:
                self.print("Cached token is invalid!")
                return False
            return True

        return False


    def set_token(self, token):
        self.api.apiToken = token

        if self.cache_token:
            cache_dir = get_cache_dir()
            token_file = os.path.join(cache_dir, self.instance)
            if cache_dir is None:
                return False

            os.makedirs(cache_dir, mode=0o700, exist_ok=True)

            with open(token_file, mode='w') as f:
                f.write(token)


        
    def interactive_get_token(self):
        """Gets the user's token using MiAuth"""
        if self.interactive is False:
            return False

        print("A new browser instance will be opened to request permissions for this app to function.")
        print("Please click 'Accept' to continue. If you don't want to continue, press Ctrl+C.")
        auth = Util.MiAuth(self.instance, permission=('read:account', 'write:notes'), name="missclear")
        webbrowser.open_new_tab(auth.getUrl())

        authed = False
        while not authed:
            try:
                authdata = auth.check()
                self.print("Authed as " + authdata['user']['username'])
                authed = True
                self.set_token(authdata['token'])
                return True
            except Exceptions.MisskeyMiAuthCheckException:
                time.sleep(5)

        return False

    def print(self, fmt, file=None):
        """Extension of print() to provide quiet support"""
        if not self.quiet:
            print(fmt)


    def auth(self):
        """ Wrapper for use_cached_token/interactive_get_token """
        if not self.use_cached_token():
            if not self.interactive:
                print("ERR: No cached token and interactive auth is disabled!", file=sys.stderr)
                return False
            else:
                if not self.interactive_get_token():
                    print("ERR: Couldn't auth", file=sys.stderr)
                    return False

        return True


    def clear_notes(self, since_days=14):
        until_stamp = int((datetime.datetime.now() - datetime.timedelta(days=since_days)).timestamp() * 1000)
        my_id = self.api.i()["id"]

        while True:
            old_notes = self.api.users_notes(my_id, untilDate=until_stamp)
            if len(old_notes) == 0:
                break

            for to_del in old_notes:
                deleted = False
                try_times = 0
                while not deleted:
                    try:
                        if self.api.notes_delete(to_del["id"]):
                            self.print("Successfully deleted " + to_del["id"] + " from " + to_del["createdAt"] + ".")
                            time.sleep(SLEEP_TIME)
                        else:
                            print("Was unable to delete " + to_del["id"] + " for some unknown reason.", file=sys.stderr)
                        # technically not necessarily deleted, but if we get False then there's nothing we can do
                        deleted = True
                    except Exception as err:
                        self.print("!!! An exception occurred of type {0}: {1}".format(type(err), err), file=sys.stderr)
                        time.sleep(BAD_SLEEP_TIME)
                        try_times += 1
                        if self.retry_count > 0 and try_times > self.retry_count:
                            print("Giving up on " + to_del["id"] + "...", file=sys.stderr)
                        else:
                            self.print("... trying again ...")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Deletes old Misskey posts.')
    parser.add_argument('instance', nargs=1,
                        help='Misskey instance domain, in format domain.tld')
    parser.add_argument('-d', '--days', type=int,
                        help='How many days back to start deleting from',
                        default=14)
    parser.add_argument('-q', '--quiet', action='store_true',
                        help='Suppress less urgent messages')
    parser.add_argument('-r', '--retries', type=int, nargs=1,
                        help='Amount of times to retry a failed delete',
                        default=0)
    authgrp = parser.add_mutually_exclusive_group()
    authgrp.add_argument('-A', '--no-auth', action='store_false',
                        help='Disable interactive authentication',
                        dest='interactive')
    authgrp.add_argument('-N', '--no-cache', action='store_false',
                        help='Disable reading/saving cached tokens',
                        dest='cache')

    args = parser.parse_args()

    mc = Missclear(args)
    if not mc.auth():
        sys.exit(1)
    if not mc.clear_notes(args.days):
        sys.exit(1)