File size: 3,842 Bytes
9770efd |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
import requests
import json
import re
from urllib.parse import urlencode
class Threads:
API_URL = "https://www.threads.net/api/graphql"
TOKEN_URL = "https://www.threads.net/@instagram"
MAX_REDIRECTS = 20
DOC_IDS = {
"getUser": "23996318473300828",
"getUserThreads": "6232751443445612",
"getUserReplies": "6307072669391286",
"getPost": "5587632691339264",
"getPostLikers": "9360915773983802"
}
def __init__(self):
self.token = self.get_token()
self.default_headers = {
"Authority": "www.threads.net",
"Accept": "*/*",
"Accept-Language": "en-US,en;q=0.9",
"Cache-Control": "no-cache",
"Content-Type": "application/x-www-form-urlencoded",
"Connection": "keep-alive",
"Origin": "https://www.threads.net",
"Pragma": "no-cache",
"Sec-Fetch-Site": "same-origin",
"User-Agent": "python-requests",
"X-ASBD-ID": "129477",
"X-IG-App-ID": "238260118697367",
"X-FB-LSD": self.token
}
def get_token(self):
session = requests.Session()
session.max_redirects = self.MAX_REDIRECTS
response = session.get(self.TOKEN_URL, headers={"User-Agent": "python-requests"})
response.raise_for_status()
match = re.search(r'"token":"(\w+?)"', response.text)
if not match:
raise ValueError("Token not found in response")
return match.group(1)
def post_request(self, variables, doc_id, extra_headers=None):
headers = self.default_headers.copy()
if extra_headers:
headers.update(extra_headers)
data = {
"lsd": self.token,
"variables": json.dumps(variables),
"doc_id": doc_id
}
response = requests.post(self.API_URL, headers=headers, data=urlencode(data))
response.raise_for_status()
return response.json()
def get_post(self, post_id):
return self.post_request({"postID": post_id}, self.DOC_IDS["getPost"], {"X-FB-Friendly-Name": "BarcelonaPostPageQuery"})
def get_post_likers(self, media_id):
return self.post_request({"mediaID": media_id}, self.DOC_IDS["getPostLikers"])
def get_user(self, user_id):
return self.post_request({"userID": user_id}, self.DOC_IDS["getUser"], {"X-FB-Friendly-Name": "BarcelonaProfileRootQuery"})
def get_user_threads(self, user_id):
return self.post_request({"userID": user_id}, self.DOC_IDS["getUserThreads"], {"X-FB-Friendly-Name": "BarcelonaProfileThreadsTabQuery"})
def get_user_replies(self, user_id):
return self.post_request({"userID": user_id}, self.DOC_IDS["getUserReplies"], {"X-FB-Friendly-Name": "BarcelonaProfileRepliesTabQuery"})
def get_user_id(self, username):
url = f"https://www.threads.net/@{username}"
headers = self.default_headers.copy()
headers.update({
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
"Referer": url,
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "cross-site",
"Sec-Fetch-User": "?1",
"Upgrade-Insecure-Requests": "1"
})
headers.pop("X-ASBD-ID", None)
headers.pop("X-FB-LSD", None)
headers.pop("X-IG-App-ID", None)
response = requests.get(url, headers=headers)
response.raise_for_status()
match = re.search(r'"user_id":"(\d+)"', response.text)
if not match:
raise ValueError("User ID not found in response")
return int(match.group(1)) |