Initial commit

LICENSE

@@ -0,0 +1,28 @@
+Copyright (c) 2020, snow flurry. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+    1. Redistributions of source code must retain the above copyright
+       notice, this list of conditions and the following disclaimer.
+    2. Redistributions in binary form must reproduce the above copyright
+       notice, this list of conditions and the following disclaimer in
+       the documentation and/or other materials provided with the
+       distribution.
+    3. Neither the name of the copyright holder nor the names of its
+       contributors may be used to endorse or promote products derived
+       from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
+IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
+TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
+PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
+TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
+LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
+NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+

README.md

@@ -0,0 +1,48 @@
+# missclear
+
+Python script to quickly delete old messages on a Misskey instance.
+
+## Usage
+
+From the built-in help:
+```
+usage: missclear.py [-h] [-d DAYS] [-q] [-r RETRIES] [-A | -N] instance
+
+Deletes old Misskey posts.
+
+positional arguments:
+  instance              Misskey instance domain, in format domain.tld
+
+optional arguments:
+  -h, --help            show this help message and exit
+  -d DAYS, --days DAYS  How many days back to start deleting from
+  -q, --quiet           Suppress less urgent messages
+  -r RETRIES, --retries RETRIES
+                        Amount of times to retry a failed delete
+  -A, --no-auth         Disable interactive authentication
+  -N, --no-cache        Disable reading/saving cached tokens
+```
+
+## Cached API Tokens
+
+The cached API token is automatically stored in either `$XDG_CACHE_DIR`
+or `$HOME/.cache` if the former doesn't exist. Each instance is cached
+in a separate file, under `<cache-dir>/missclear/instance.tld` for a
+hypothetical instance.tld.
+
+The `--no-cache` flag will disable the cache functionality, but requires
+interactive authentication every time.
+
+## Automated runs (e.g. crontab)
+
+As long as you're comfortable with your API token being stored wherever
+your local cache folder is stored, something like the following command
+should work as an automatic function:
+
+```
+/path/to/missclear.py -Aqr 5 instance.tld
+```
+
+Usage of `--retries X` in this case is recommended, since otherwise an
+API issue such as rate limiting will cause the script to loop forever.
+

missclear.py

@@ -0,0 +1,191 @@
+#!/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):
+        """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)