block_tunnelers.py

gistfile1.txt

import requests
import time

# https://github.com/twitterdev/enterprise-scripts-python/blob/main/Engagement-API/generate_user_access_tokens.py
from generate_user_access_tokens import request_token, get_user_authorization, get_user_access_tokens, CONSUMER_KEY, CONSUMER_SECRET

MY_ID = "PUT_YOUR_TWITTER_ID_NUMBER_HERE"


def create_url():
    id = "1354143047324299264"
    # You can adjust ids to include a single Tweets.
    # Or you can add to up to 100 comma-separated IDs
    url = "https://api.twitter.com/2/tweets/{}/liking_users".format(id)
    return url, {}


resource_owner_oauth_token, resource_owner_oauth_token_secret = request_token()
authorization_pin = get_user_authorization(resource_owner_oauth_token)
access_token, access_token_secret, user_id, screen_name = get_user_access_tokens(resource_owner_oauth_token,
                                                                                 resource_owner_oauth_token_secret,
                                                                                 authorization_pin)
user_context_auth = OAuth1(CONSUMER_KEY, CONSUMER_SECRET, access_token, access_token_secret)


def connect_to_endpoint(url, user_fields, verb="GET", json_body=None):
    response = requests.request(verb, url, auth=user_context_auth, params=user_fields, json=json_body)
    if response.status_code != 200:
        raise Exception(
            "Request returned an error: {} {}".format(
                response.status_code, response.text
            )
        )
    return response.json()


def block(res):
    if "data" not in res:
        return
    for username, uid in [(vals["username"], vals["id"]) for vals in res["data"]]:
        URL = f"https://api.twitter.com/2/users/{MY_ID}/blocking"
        connect_to_endpoint(URL, {}, json_body={"target_user_id": uid}, verb="POST")
        print(username, "blocked")
        time.sleep(20)  # 50 requests / 15 minutes


def main():
    url, tweet_fields = create_url()
    json_response = connect_to_endpoint(url, tweet_fields)
    print("blcoking")
    block(json_response)
    while "next_token" in json_response["meta"]:
        json_response = connect_to_endpoint(url, {"pagination_token": json_response['meta']['next_token']})
        block(json_response)


if __name__ == "__main__":
    main()