Created
March 19, 2024 17:17
-
-
Save cometothed4rkside/352af3b991c1bc0c508835e350f80388 to your computer and use it in GitHub Desktop.
Twitter'da troll engellemek için python kodları
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import requests | |
| import getpass | |
| import argparse | |
| import threading | |
| import json | |
| import requests | |
| from bs4 import BeautifulSoup | |
| import re | |
| import json | |
| import urllib3 | |
| import urllib.parse | |
| import sys | |
| import argparse | |
| import getpass | |
| import csv | |
| import time | |
| urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) | |
| import threading | |
| class BlockFailureException(Exception): | |
| pass | |
| def getTokens(): | |
| # guest_id | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Accept' : "*/*"} | |
| url_base = "https://twitter.com" | |
| r = requests.get(url_base, verify=False, headers=user_agent, allow_redirects=False) | |
| guest_id = r.cookies.get("guest_id") | |
| print("[*] Guest_id: %s" % guest_id) | |
| # Endpoints | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' } | |
| url_base = "https://twitter.com/home?precache=1" | |
| r = requests.get(url_base, verify=False, headers=user_agent,allow_redirects=True) | |
| soup = BeautifulSoup(r.text, "html.parser") | |
| js_with_bearer = "" | |
| for i in soup.find_all('link'): | |
| if i.get("href").find("/main") != -1: | |
| js_with_bearer = i.get("href") | |
| tweetActivity_endpoint = re.findall(r'"shared~bundle.TweetEditHistory~bundle.QuoteTweetActivity~bundle.TweetActivity":"(.*)",', r.text, re.IGNORECASE)[0].split("\"")[0] | |
| print("[*] endpoints.TweetActivity: %s" % tweetActivity_endpoint) | |
| print("[*] Js with Bearer token: %s" % js_with_bearer) | |
| # Guest token | |
| guest_token = re.findall(r'"gt=\d{19}', str(soup.find_all('script')[1]), re.IGNORECASE)[0].replace("\"gt=","") | |
| print("[*] Guest token: %s" % guest_token) | |
| # Get Bearer token | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Referer' : 'https://twitter.com/sw.js' } | |
| r = requests.get(js_with_bearer, verify=False, headers=user_agent) | |
| bearer = re.findall(r'"Bearer(.*)";', r.text, re.IGNORECASE)[0].split("\"")[0] | |
| authorization_bearer = "Bearer%s" % bearer | |
| print("[*] Bearer: %s" % bearer) | |
| # Guest token II | |
| url_guest_token = "https://api.twitter.com/1.1/guest/activate.json" | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', | |
| 'Accept' : "*/*", | |
| 'Authorization' : authorization_bearer, | |
| 'Cookie' : 'guest_id=%s;' % guest_id } | |
| r = requests.post(url_guest_token, verify=False, headers=user_agent, data="") | |
| guest_token = json.loads(r.text)['guest_token'] | |
| print("[*] Guest token: %s" % guest_token) | |
| # Retweeters path is in other JS now (endpoints.TweetActivity.xxxx) | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Origin' : 'https://twitter.com/' ,'Referer' : 'https://twitter.com/' } | |
| url_retweet_endpoint = "https://abs.twimg.com/responsive-web/client-web/shared~bundle.TweetEditHistory~bundle.QuoteTweetActivity~bundle.TweetActivity.%sa.js" % tweetActivity_endpoint | |
| print("[*] url_retweet_endpoint: %s" % url_retweet_endpoint) | |
| r = requests.get(url_retweet_endpoint, verify=False, headers=user_agent) | |
| rt_path = re.search(r'queryId:"(.+?)",operationName:"Retweeters"', r.text).group(1).split('"')[-1] | |
| # Viewer path is in other JS now (endpoints.UsersGraphQL.xxx) | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Origin' : 'https://twitter.com/' ,'Referer' : 'https://twitter.com/' } | |
| # endpoints.UsersGraphQL are now in main.js (js_with_bearer) | |
| url_viewer_endpoint = js_with_bearer | |
| r = requests.get(url_viewer_endpoint, verify=False, headers=user_agent) | |
| viewer_path = re.search(r'queryId:"(.+?)",operationName:"Viewer"', r.text).group(1).split('"')[-1] | |
| print("[*] rt_url: %s" % rt_path) | |
| return authorization_bearer,guest_token,rt_path,viewer_path | |
| def login(authorization_bearer, guest_token, username, password, email): | |
| # SSO login | |
| url_flow_1 = "https://api.twitter.com/1.1/onboarding/task.json?flow_name=login" | |
| url_flow_2 = "https://api.twitter.com/1.1/onboarding/task.json" | |
| # Flow 1 | |
| data = {'' : ''} | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', | |
| 'Referer' : 'https://twitter.com/', | |
| 'X-Guest-Token' : guest_token, 'Content-Type' : 'application/json', | |
| 'Authorization' : authorization_bearer, 'X-Twitter-Active-Use' : 'yes', | |
| 'X-Twitter-Client-Language' : 'en', | |
| 'Origin' : 'https://twitter.com' } | |
| r = requests.post(url_flow_1, verify=False, headers=user_agent, data=json.dumps(data)) | |
| flow_token = json.loads(r.text)['flow_token'] | |
| cookie = ';'.join(['%s=%s' % (name, value) for (name, value) in r.cookies.get_dict(domain=".twitter.com").items()]) | |
| print("[*] flow_token: %s" % flow_token) | |
| # Flow 2 | |
| data = {'flow_token' : flow_token, "subtask_inputs" : []} | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Referer' : 'https://twitter.com/sw.js', 'X-Guest-Token' : guest_token, 'Content-Type' : 'application/json', 'Authorization' : authorization_bearer, 'Cookie' : cookie } | |
| r = requests.post(url_flow_2, verify=False, headers=user_agent, data=json.dumps(data)) | |
| flow_token = json.loads(r.text)['flow_token'] | |
| print("[*] flow_token: %s" % flow_token) | |
| # Flow 3 | |
| data = {"flow_token": flow_token ,"subtask_inputs":[{"subtask_id":"LoginEnterUserIdentifierSSO","settings_list":{"setting_responses":[{"key":"user_identifier","response_data":{"text_data":{"result":username}}}],"link":"next_link"}}]} | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Referer' : 'https://twitter.com/sw.js', 'X-Guest-Token' : guest_token, 'Content-Type' : 'application/json', 'Authorization' : authorization_bearer, 'Cookie' : cookie } | |
| r = requests.post(url_flow_2, verify=False, headers=user_agent, data=json.dumps(data)) | |
| flow_token = json.loads(r.text)['flow_token'] | |
| print("[*] flow_token: %s" % flow_token) | |
| if (json.loads(r.text)['subtasks'][0]['subtask_id'] == "LoginEnterAlternateIdentifierSubtask"): | |
| # Sometimes login alternate because unusual LoginEnterUserIdentifierSSOSubtask | |
| data = {"flow_token": flow_token, "subtask_inputs":[{"subtask_id":"LoginEnterAlternateIdentifierSubtask","enter_text":{"text": email,"link":"next_link"}}]} | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Referer' : 'https://twitter.com/sw.js', 'X-Guest-Token' : guest_token, 'Content-Type' : 'application/json', 'Authorization' : authorization_bearer, 'Cookie' : cookie } | |
| r = requests.post(url_flow_2, verify=False, headers=user_agent, data=json.dumps(data)) | |
| flow_token = json.loads(r.text)['flow_token'] | |
| print("[*] flow_token: %s" % flow_token) | |
| # Flow 4 | |
| data = {"flow_token": flow_token ,"subtask_inputs":[{"subtask_id":"LoginEnterPassword","enter_password":{"password":password,"link":"next_link"}}]} | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Referer' : 'https://twitter.com/sw.js', 'X-Guest-Token' : guest_token, 'Content-Type' : 'application/json', 'Authorization' : authorization_bearer, 'Cookie' : cookie } | |
| r = requests.post(url_flow_2, verify=False, headers=user_agent, data=json.dumps(data)) | |
| flow_token = json.loads(r.text)['flow_token'] | |
| user_id = json.loads(r.text)['subtasks'][0]['check_logged_in_account']['user_id'] | |
| print("[*] flow_token: %s" % flow_token) | |
| print("[*] user_id: %s" % user_id) | |
| # Flow 5 (and get auth_token) | |
| data = {"flow_token":flow_token,"subtask_inputs":[{"subtask_id":"AccountDuplicationCheck","check_logged_in_account":{"link":"AccountDuplicationCheck_false"}}]} | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Referer' : 'https://twitter.com/sw.js', 'X-Guest-Token' : guest_token, 'Content-Type' : 'application/json', 'Authorization' : authorization_bearer, 'Cookie' : cookie } | |
| r = requests.post(url_flow_2, verify=False, headers=user_agent, data=json.dumps(data)) | |
| flow_token = json.loads(r.text)['flow_token'] | |
| auth_token = r.cookies['auth_token'] | |
| print("[*] flow_token: %s" % flow_token) | |
| print("[*] auth_token: %s" % auth_token) | |
| return auth_token | |
| def getCSRFToken(guest_token, auth_token, authorization_bearer): | |
| # Get CSRF Token | |
| payload = '{"withCommunitiesMemberships":true,"withCommunitiesCreation":true,"withSuperFollowsUserFields":true}' | |
| url_session_token = "https://twitter.com/i/api/graphql/%s/Viewer?variables=%s" % (viewer_path, urllib.parse.quote_plus(payload)) | |
| cookie = "ct0=%s; auth_token=%s" % (guest_token, auth_token) | |
| user_agent = { 'User-Agent' : 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', 'Referer' : 'https://twitter.com/sw.js', 'X-Guest-Token' : guest_token, 'Content-Type' : 'application/json', 'Authorization' : authorization_bearer, 'Cookie' : cookie } | |
| r = requests.get(url_session_token, verify=False, headers=user_agent) | |
| csrf_token = r.cookies['ct0'] | |
| print("[*] CSRF token: %s" % csrf_token) | |
| return csrf_token | |
| def block_users_worker(user_ids, auth_token, csrf_token, authorization_bearer, blocked_users): | |
| count = 0 | |
| for user_id in user_ids: | |
| try: | |
| engelle(user_id, auth_token, csrf_token, authorization_bearer, blocked_users) | |
| count += 1 | |
| if count % 60 == 0: | |
| print("Pausing for 15 seconds...") | |
| except BlockFailureException: | |
| print("Error") | |
| def remove_blocked_users(file_path, blocked_users): | |
| with open(file_path, "r") as file: | |
| lines = file.readlines() | |
| with open(file_path, "w") as file: | |
| for line in lines: | |
| if line.strip() not in blocked_users: | |
| file.write(line) | |
| def engelle(user_id, auth_token, csrf_token, authorization_bearer, blocked_users): | |
| url_block = "https://twitter.com/i/api/1.1/blocks/create.json" | |
| data = "screen_name=%s" % user_id | |
| if user_id == "mansuryavas06" or user_id == "ekrem_imamoglu": | |
| print("Başkanları atladık.") | |
| else: | |
| cookie = "ct0=%s; auth_token=%s" % (csrf_token, auth_token) | |
| user_agent = { | |
| 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:91.0) Gecko/20100101 Firefox/91.0', | |
| 'X-Csrf-Token': csrf_token, | |
| 'Content-Type': 'application/x-www-form-urlencoded', | |
| 'Authorization': authorization_bearer, | |
| 'Cookie': cookie | |
| } | |
| r = requests.post(url_block, verify=False, headers=user_agent, data=data) | |
| response_json = json.loads(r.text) | |
| if 'id_str' not in response_json or 'screen_name' not in response_json: | |
| raise BlockFailureException(f"Failed to block user: {user_id}") | |
| if 'id_str' in response_json and 'screen_name' in response_json: | |
| blocked_users.append(user_id) | |
| with open('twitter_users.csv', 'a', newline='', encoding='utf-8') as file: | |
| writer = csv.writer(file) | |
| writer.writerow([response_json['id_str'], response_json['screen_name']]) | |
| print("[+] Engellendi: %s" % response_json['screen_name']) | |
| else: | |
| print("[!] Hata. Yeniden giriş gerekiyor.") | |
| def write_to_csv(user_id, screen_name, file_path='blocked_users.csv'): | |
| with open(file_path, mode='a', newline='', encoding='utf-8') as file: | |
| writer = csv.writer(file) | |
| writer.writerow([user_id, screen_name]) | |
| def read_user_ids_from_file(file_path): | |
| with open(file_path, 'r') as file: | |
| user_ids = file.readlines() | |
| return [user_id.strip() for user_id in user_ids] | |
| def block_users_from_file(file_path, auth_token, csrf_token, authorization_bearer): | |
| user_ids = read_user_ids_from_file(file_path)[:60] | |
| blocked_users = [] | |
| threads = [] | |
| num_threads = 1 | |
| for i in range(num_threads): | |
| start_index = i * len(user_ids) // num_threads | |
| end_index = (i + 1) * len(user_ids) // num_threads | |
| thread_user_ids = user_ids[start_index:end_index] | |
| thread = threading.Thread(target=block_users_worker, args=(thread_user_ids, auth_token, csrf_token, authorization_bearer, blocked_users)) | |
| threads.append(thread) | |
| thread.start() | |
| for thread in threads: | |
| thread.join() | |
| remove_blocked_users(file_path, blocked_users) | |
| if __name__ == "__main__": | |
| while True: | |
| try: | |
| authorization_bearer, guest_token, rt_path, viewer_path = getTokens() | |
| auth_token = login(authorization_bearer, guest_token, "kullanıcı adı yazın", "şifre yazın", None) | |
| csrf_token = getCSRFToken(guest_token, auth_token, authorization_bearer) | |
| block_users_from_file("engelle.txt", auth_token, csrf_token, authorization_bearer) | |
| except BlockFailureException as e: | |
| print(f"Blocking failed, restarting: {e}") | |
| remove_blocked_users("engelle.txt", []) | |
| time.sleep(3) | |
| except Exception as e: | |
| print(f"An unexpected error occurred: {e}") | |
| break |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment