Python script to ban a list of Twitch usernames/IDs

TwitchHammer.py

import requests
import time
import os
import argparse
import re

"""
Twitch API examples:

curl -X GET 'https://api.twitch.tv/helix/users?id=141981764' \
-H 'Authorization: Bearer X' \
-H 'Client-Id: X'

curl -X POST 'https://api.twitch.tv/helix/moderation/bans?broadcaster_id=1234&moderator_id=5678' \
-H 'Authorization: Bearer X' \
-H 'Client-Id: X' \
-H 'Content-Type: application/json' \
-d '{"data": {"user_id":"9876","reason":"no reason"}}'
"""

API_URL_USERS = "https://api.twitch.tv/helix/users?login={}"
API_URL_ID    = "https://api.twitch.tv/helix/users?id={}"
API_URL_BAN   = "https://api.twitch.tv/helix/moderation/bans?broadcaster_id={}&moderator_id={}"

class BanHammer:
    def __init__(self,
                 channel: str,
                 moderator: str,
                 access_token: str,
                 client_id: str,
                 ban_reason: str,
                 only_ids: bool,
                 rate_limit: float,
                 verbose_output: bool) -> None:
        """
        Init ban hammer
        """
        self.ban_reason         = ban_reason
        self.channel_username   = channel
        self.moderator_username = moderator
        self.rate_limit         = rate_limit
        self.verbose_output     = verbose_output
        self.already_banned     = 0

        # Two modes:
        # > True:  The list contains only user IDs
        # > False: The list contains only usernames
        self.only_ids = only_ids

        # Authorization headers
        # Use https://twitchtokengenerator.com/ to obtain them (requires moderator:manage:banned_users scope)
        self.headers = {
            "Authorization": f"Bearer {access_token}",
            "Client-Id": client_id,
            "Content-Type": "application/json"
        }

        self.broadcaster_id = self.get_user_id(channel)
        self.moderator_id   = self.get_user_id(moderator)

        if self.broadcaster_id is None or self.moderator_id is None:
            print("[!] Failed to obtain the channel or moderator ID. Leaving!")
            os.exit(1)

    def get_user_id(self, user: str):
        """
        Returns the ID of a given user
        """
        r = requests.get(API_URL_USERS.format(user), headers=self.headers)

        # Authorization headers are not valid
        if r.status_code == 401:
            print("[!] The provided authorization tokens are not valid. Leaving!")
            os.exit(1)

        try:
            return int(r.json()["data"][0]["id"])
        except:
            # User not found
            return None

    def convert_user_id(self, _id: int):
        """
        Returns the username of a given ID
        """
        # Perform request
        r = requests.get(API_URL_ID.format(_id), headers=self.headers)
        try:
            return r.json()["data"][0]["login"]
        except:
            # ID not found or authorization tokens invalid
            return None

    def ban_user(self, username: str) -> bool:
        """
        Bans user (using its username)
        """
        # Obtain user ID
        user_id = self.get_user_id(username)

        if user_id is None:
            # Failed to obtain user ID
            return False

        return self.ban_user_id(user_id)

    def ban_user_id(self, user_id: int) -> bool:
        """
        Bans user (using its ID)
        """
        if self.ban_reason:
            data = {
                "data": {
                    "user_id": user_id,
                    "reason": self.ban_reason
                }
            }
        else:
            data = {
                "data": {
                    "user_id": user_id
                }
            }
        # Perform request
        r = requests.post(API_URL_BAN.format(self.broadcaster_id,
                                             self.moderator_id),
                          headers=self.headers,
                          json=data)

        # User is not a moderator in the given channel
        if r.status_code == 403:
            print(f"[!] The user {self.moderator_username} is not a mod in {self.channel_username}. Leaving!")
            os.exit(1)

        # Authorization tokens do not include the moderator:manage:banned_users scope
        if r.status_code == 401:
            print("[!] The authorization tokens must include the moderator:manage:banned_users scope. Leaving!")
            os.exit(1)

        try:
            # If the user has been banned successfully there will be data in the response
            r.json()["data"][0]
            return True
        except:
            # Failed to ban user
            # Check if it was already banned
            error_message = r.json()["message"]
            if error_message == "The user specified in the user_id field is already banned.":
                self.already_banned += 1
                return True
            if self.verbose_output:
                print(f"[!] Failed to ban user {user_id}: {error_message}\n")
            return False

    def _is_valid_twitch_username(self, username):
        """
        Checks if a username is valid
        """
        pattern = r"^[a-zA-Z0-9][a-zA-Z0-9_]{3,24}$"
        return re.match(pattern, username) is not None

    def _check_list_content(self, data: list) -> None:
        """
        Iterates through the data to obtain valid data
        """
        # Valid IDs/usernames
        valid_data = []

        for idx, entry in enumerate(data):
            # Remove new lines
            entry = entry.rstrip()

            # Check username if needed
            if self.only_ids:
                if entry.isnumeric() and int(entry) > 0:
                    valid_data.append((None, int(entry)))
            else:
                print(f"[+] Converting usernames to Twitch IDs: {idx + 1}/{len(data)} " \
                      f"({round((idx + 1) / len(data) * 100, 2)}%)" + " " * 10, end="\r")
                if not self._is_valid_twitch_username(entry):
                    continue
                else:
                    entry_id = self.get_user_id(entry)
                    if entry is None:
                        continue
                    valid_data.append((entry, entry_id))

        if not self.only_ids: print()
        print(f"[+] Trying to ban {len(valid_data)} users")

        not_banned = []
        banned = []
        for idx, username_data in enumerate(valid_data):
            print(f"[+] Ban progress: {idx + 1}/{len(valid_data)} " \
                  f"({round((idx + 1) / len(valid_data) * 100, 2)}%)" + " " * 10, end="\r")
            
            # Try to ban user
            status = self.ban_user_id(username_data[1])
            if status:
                banned.append(username_data)
            else:
                not_banned.append(username_data)

            # Apply rate limit
            time.sleep(self.rate_limit)

        if len(not_banned):
            print(f"\n[+] The following accounts could not be banned (most probably they do not exist):")
            for account in not_banned:
                if account[0] is None:
                    print(f"- {account[1]}")
                else:
                    print(f"- {account[0]} ({account[1]})")

        if len(banned):
            print(f"\n[+] Successfully banned {len(banned)} users!")
        else:
            print("\n[+] No user was banned!")

        if self.already_banned > 0:
            if self.already_banned == 1:
                print("[+] Fact: 1 account was already banned")
            else:
                print(f"[+] Fact: {self.already_banned} accounts were already banned")

    def massban_file(self, filepath: str) -> None:
        """
        Reads a file with user IDs/usernames
        """
        if not os.path.isfile(filepath):
            print("[!] The specified filepath does not exist or is not a file. Leaving!")
            os.exit(1)

        # Check content and ban valid users
        self._check_list_content(open(filepath, "r").readlines())

    def massban_url(self, url: str) -> None:
        """
        Obtains a list of user IDs/usernames from a remote URL
        """
        try:
            r = requests.get(url)
            self._check_list_content(r.text.split("\r\n"))
        except:
            print("[!] Failed to fetch data from remote URL. Leaving!")
            os.exit(1)

if __name__ == "__main__":
    # Create argument parser
    parser = argparse.ArgumentParser()
    parser.add_argument("-c", "--channel", help="Channel where to ban", required=True)
    parser.add_argument("-m", "--mod", help="Moderator username", required=True)
    parser.add_argument("-a", "--access-token", help="Access token", required=True)
    parser.add_argument("-i", "--client-id", help="Client ID token", required=True)
    parser.add_argument("-r", "--reason", help="Ban reason", default="")
    parser.add_argument("-f", "--file-path", help="File containing usernames/IDs")
    parser.add_argument("-u", "--url", help="URL to paste containing usernames/IDs")
    parser.add_argument("--ids", help="Ban using IDs instead of usernames", action="store_true")
    parser.add_argument("--rate-limit", type=float, help="Rate limit applied (in seconds)", default=0.125)
    parser.add_argument("-v", "--verbose", help="Verbose output", action="store_true")
    args = parser.parse_args()

    if args.url is not None and args.file_path is not None:
        print("[!] You have specified both a path to a file and a URL. You must choose only one!")
        os.exit(1)
    if args.url is None and args.file_path is None:
        print("[!] You must provide a path to a file or a URL containing Twitch usernames/IDs!")
        os.exit(1)

    # Create hammer
    hammer = BanHammer(args.channel,
                       args.mod,
                       args.access_token,
                       args.client_id,
                       args.reason,
                       args.ids,
                       args.rate_limit,
                       args.verbose)
    if args.file_path:
        hammer.massban_file(args.file_path)
    if args.url:
        hammer.massban_url(args.url)