Fedi-Cleaner/main.py

import json
import logging
from datetime import datetime, timedelta
from mastodon import Mastodon
from config import *
from valid import valid_accounts
from clean import clean_accounts


def fetch_accounts(api_function):
    accounts = api_function()
    accounts = mastodon.fetch_remaining(accounts)
    accounts = [
        acc
        for acc in accounts
        if acc["last_status_at"] is None  # have no posts
        or acc["last_status_at"] < valid_date  # inactive accounts
    ]
    return accounts


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    logging.info("Starting Fedi Cleaner")
    mastodon = Mastodon(
        access_token=ACCESS_TOKEN, api_base_url=API_BASE_URL, ratelimit_method="pace"
    )

    current_user = mastodon.me()
    uid = current_user["id"]
    following, followers, lists_members, blocks, mutes = [], [], [], [], []
    valid_date = datetime.now() - timedelta(days=INACTIVE_DAYS)  # FIXME: UTC

    if CLEAN_FOLLOWING or not CLEAN_MUTUALS:
        logging.info("Fetching following")
        following = fetch_accounts(lambda: mastodon.account_following(uid))

    if CLEAN_FOLLOWERS or not CLEAN_MUTUALS:
        logging.info("Fetching followers")
        followers = fetch_accounts(lambda: mastodon.account_followers(uid))

    if CLEAN_LISTS:  # in akkoma/pleroma list members can be unfollowed
        logging.info("Fetching lists")
        lists = mastodon.lists()
        for list_ in lists:
            list_members = fetch_accounts(lambda: mastodon.list_accounts(list_["id"]))
            lists_members.append(
                {"id": list_["id"], "title": list_["title"], "members": list_members}
            )

    if CLEAN_BLOCKS:  # they can't follow us so there's no mutuals
        logging.info("Fetching blocks")
        blocks = fetch_accounts(mastodon.blocks)

    if CLEAN_MUTES:
        logging.info("Fetching mutes")
        mutes = fetch_accounts(mastodon.mutes)

    if not CLEAN_MUTUALS:  # just don't want to use relationship api
        logging.info("Excluding mutuals")
        following_id = [acc["id"] for acc in following]
        followers_id = [acc["id"] for acc in followers]
        mutuals = list(set(following_id) & set(followers_id))
        # mutuals are excluded
        following = [acc for acc in following if acc["id"] not in mutuals]
        followers = [acc for acc in followers if acc["id"] not in mutuals]
        mutes = [acc for acc in mutes if acc["id"] not in mutuals]
        for list_ in lists_members:
            list_["members"] = [
                acc for acc in list_["members"] if acc["id"] not in mutuals
            ]

    accounts = {
        "following": following if CLEAN_FOLLOWING else None,
        "followers": followers if CLEAN_FOLLOWERS else None,
        "lists_members": lists_members if CLEAN_LISTS else None,
        "blocks": blocks if CLEAN_BLOCKS else None,
        "mutes": mutes if CLEAN_MUTES else None,
    }

    # if DRY_RUN:
    #     with open("accounts.json", "w") as f:
    #         f.write(
    #             json.dumps(
    #                 accounts,
    #                 indent=4,
    #                 default=str,
    #             )
    #         )

    # now validate those accounts
    # criteria: 1. account migration 2. inactive account 3. dead instance or errors
    criteria = {
        1: CLEAN_MIGRATED_ACCOUNTS,
        2: CLEAN_INACTIVE_ACCOUNTS,
        3: CLEAN_DEAD_ACCOUNTS,
    }
    operations = {}
    for k, v in accounts.items():
        logging.info(f"Validating {k}")
        if v is None:
            continue

        if k == "lists_members":
            operation = {
                list_["id"]: [
                    id
                    for id, state in valid_accounts(list_["members"])
                    if criteria[state]
                ]
                for list_ in v
            }
        else:
            operation = [id for id, state in valid_accounts(v) if criteria[state]]

        operations[k] = operation

    # now clean those accounts
    # or only write operations to file
    if DRY_RUN:
        with open("operations.json", "w") as f:
            f.write(
                json.dumps(
                    operations,
                    indent=4,
                    default=str,
                )
            )
    else:
        clean_accounts(mastodon, operations)
    logging.info("Done!")