tweet_purge.py

import os
import time
import json
from typing import Union
from datetime import datetime
from argparse import ArgumentParser, Namespace

import tweepy
from tweepy.cursor import ItemIterator


class ThePurge:

    __slots__ = ('twitter_handle', 'access_credentials', 'date_range', 'exceptions')

    def __init__(
        self,
        access_credentials: dict[str, str],
        exceptions: set[int],
        # date_range: Union[tuple[str, str], None] = None,
    ) -> None:

        self.access_credentials: dict[str, str] = access_credentials
        self.exceptions: set[int] = exceptions
        self.date_range: Union[tuple[datetime, datetime], None] = None

        auth = tweepy.OAuthHandler(
            self.access_credentials['consumer_key'],
            self.access_credentials['consumer_secret']
        )
        auth.set_access_token(
            self.access_credentials['access_token'],
            self.access_credentials['access_token_secret']
        )

        self.twitter_handle: tweepy.API = tweepy.API(auth)

        # # Not implemented in the CLI
        # if date_range:

        #     if len(date_range) != 2:
        #         raise ValueError("Please enter both dates")
        #     elif all([type(i) == str for i in date_range]):
        #         raise ValueError("Both dates should be strings")

        #     try:

        #         start = datetime(*[int(i) for i in date_range[0].split('-')])  # type: ignore
        #         end = datetime(*[int(i) for i in date_range[1].split('-')])  # type: ignore

        #         if start > end:
        #             raise ValueError("Start date is less than the end date.")

        #         # Now its sanitized, let's save it
        #         self.date_range = start, end
        #         # We will do manual date filtering by retrieving everything then sorting by tweet.created_at

        #     except Exception as e:
        #         print(e)

    def purge(self, real: bool = False) -> None:
        ''' The Purge '''

        # Purge errrthaaang
        def purge_func(deletees: ItemIterator, destructor_function: str) -> None:

            to_delete: set[int] = set()

            for tweet in deletees:
                if self.date_range:
                    if tweet.created_at >= self.date_range[0] and tweet.tweeted_at <= self.date_range[1]:
                        to_delete.add(tweet.id)
                else:
                    to_delete.add(tweet.id)

            for tweet in to_delete - self.exceptions:
                print("Deleting ", tweet)
                try:
                    if real:
                        getattr(self.twitter_handle, destructor_function)(tweet)
                except tweepy.TweepError as e:
                    print(f"ID {tweet}: {e}")
                time.sleep(0.1)

        print("Deleting tweets ...")
        purge_func(tweepy.Cursor(self.twitter_handle.user_timeline).items(), 'destroy_status')

        print("Unliking things ...")
        purge_func(tweepy.Cursor(self.twitter_handle.favorites).items(), 'destroy_favorite')


if __name__ == '__main__':

    # credentials.json
    # {
    # "consumer_key": "ck",
    # "consumer_secret": "cs",
    # "access_token": "at",
    # "access_token_secret": "ats"
    # }

    # exceptions.json
    # {
    # "tweet_id_1",
    # "tweet_id_2",
    # "..."
    # }

    try:

        if not os.path.isfile("credentials.json"):
            raise ValueError("Credentials.json does not exist")
        elif not os.path.isfile("exceptions.json"):
            raise ValueError("exceptions.json does not exist. Please make one (even if empty).")

        with open("credentials.json") as creds, open("exceptions.json") as ex:

            parser = ArgumentParser("💥 TweetNuke 💥")
            parser.add_argument('--real', action='store_true')

            purger = ThePurge(
                access_credentials=json.load(creds),
                exceptions={int(tid) for tid in json.load(ex)}
            )

            args: Namespace = parser.parse_args()

            if args.real:
                purger.purge(real=True)
            else:
                purger.purge()

    except KeyboardInterrupt:
        exit()