Skip to content

Commit

Permalink
change a lot of stuff
Browse files Browse the repository at this point in the history
  • Loading branch information
@Ryu1845
Ryu1845 committed Jan 2, 2022
1 parent 223f5a1 commit 04f1dfa
Show file tree
Hide file tree
Showing 3 changed files with 182 additions and 138 deletions.
233 changes: 116 additions & 117 deletions twspace_dl/Login.py
View file
Original file line number Diff line number Diff line change
@@ -1,81 +1,72 @@
import requests
from typing import Optional


class Login:
def __init__(self, username, password):
def __init__(self, username, password, guest_token):
self.username = username
self.password = password
self.guest_token = guest_token
self.session = requests.Session()
self.guest_token = None
self.flow_token = None
self.auth_token = None
self.task_url = "https://twitter.com/i/api/1.1/onboarding/task.json"

def set_token(self, guest_token):
self.guest_token = guest_token
self.flow_token: str

def get_token(self) -> str:
return self.auth_token

def login(self) -> str:
if self.auth_token:
return self.auth_token
# Just do simple post request to get the auth token
r = self.session.post(self.task_url + "?flow_name=login",
headers = self.get_headers(),
json = self.get_initial_params()
def login(self) -> Optional[str]:
request_flow = self.session.post(
self.task_url,
params={"flow_name": "login"},
headers=self._headers,
json=self._initial_params,
)
try:
self.flow_token = r.json()["flow_token"]
except:
print("Error while intiial_params:", r.json())
self.flow_token = request_flow.json()["flow_token"]
except KeyError:
print("Error while intiial_params:", request_flow.json())
return None

# Task 0
r = self.session.post(self.task_url,
headers = self.get_headers(),
json = self.get_task0_data()

# js instrumentation subtask
request_flow = self.session.post(
self.task_url, headers=self._headers, json=self._js_instrumentation_data
)
try:
self.flow_token = r.json()["flow_token"]
except:
print("Error while task0:", r.json())
self.flow_token = request_flow.json()["flow_token"]
except KeyError:
print("Error while task0:", request_flow.json())
return None

# Task 1
r = self.session.post(self.task_url,
headers = self.get_headers(),
json = self.get_task1_data()

# user identifier sso subtask
request_flow = self.session.post(
self.task_url, headers=self._headers, json=self._user_identifier_sso_data
)
try:
self.flow_token = r.json()["flow_token"]
except:
print("Error while task1:", r.json())
self.flow_token = request_flow.json()["flow_token"]
except KeyError:
print("Error while task1:", request_flow.json())
return None

# Task 2
r = self.session.post(self.task_url,
headers = self.get_headers(),
json = self.get_task2_data()
# account duplication check
request_flow = self.session.post(
self.task_url, headers=self._headers, json=self._account_dup_check_data
)
try:
self.flow_token = r.json()["flow_token"]
except:
print("Error while task2:", r.json())
self.flow_token = request_flow.json()["flow_token"]
except KeyError:
print("Error while task2:", request_flow.json())
return None

# Task 6
r = self.session.post(self.task_url,
headers = self.get_headers(),
json = self.get_task6_data()
# enter password
request_flow = self.session.post(
self.task_url, headers=self._headers, json=self._enter_password_data
)
try:
self.auth_token = str(r.cookies["auth_token"])
except:
print("Error while task6:", r.json())
auth_token = str(request_flow.cookies["auth_token"])
except KeyError:
print("Error while task6:", request_flow.json())
return None
return self.auth_token

def get_headers(self):
return auth_token

@property
def _headers(self):
return {
"authorization": (
"Bearer "
Expand All @@ -85,94 +76,102 @@ def get_headers(self):
"x-guest-token": self.guest_token,
}

def get_task0_data(self) -> dict:
@property
def _js_instrumentation_data(self) -> dict:
return {
"flow_token": self.flow_token,
"subtask_inputs":[
"subtask_inputs": [
{
"subtask_id":"LoginJsInstrumentationSubtask",
"js_instrumentation":{
"response":"{\"rf\":{\"a976808c0d7d2c9a6081e997a1114f952af0067dcb59267ac8d852ec0ef98fe6\":-65,\"a1442a8ec6ecb5804b8f60864cc54ded1bc140c4adaef6ac8642d181ba17e0fd\":1,\"a3d158c7a0003247a36ff4860f096c8eeefb23608d983162f94344db0b0a1f84\":-10,\"a8031adb0e4372374f15125f9a10e5b8017743895f495b43c1aa74839c9a5876\":15},\"s\":\"HrR8ThECGdeQ4UgaWLIMSqGx0fL_7FOf7KjX8tlWN6WBN6HVcojL3if3rNbYDtDhDmwK1jxMViInpjc1hc-kOO5w6Ej7WxoqdmI0eTVj-5iul5FMdGaGZUVuWtkq3A7A42Y5RAsgNwYtpVB44XifZ3W1fMscefI8HovjFtWUm0caZkF6_Y_1iFr0FSWHgM95gx0pXkK910VlKn0HqT8Dvo6ss7LMA5Cf-VS84q284Vsx6h3nqwTgzo4Nx3V4d86VL45GqIzqbwKT0OMlM6DHKk2Pi8WxKZ_QoHAMQI0AzBCJ6McdfjGf7lCjtLLRb4ClfZNTW0gIX3dMSEj03mvOkgAAAX4abfqW\"}",
"link":"next_link"
}
}
]}
def get_task1_data(self) -> dict:
"subtask_id": "LoginJsInstrumentationSubtask",
"js_instrumentation": {
"response": (
'{"rf":{"a976808c0d7d2c9a6081e997a1114f952a'
'f0067dcb59267ac8d852ec0ef98fe6":-65,"a1442a'
"8ec6ecb5804b8f60864cc54ded1bc140c4adaef6ac8"
'642d181ba17e0fd":1,"a3d158c7a0003247a36ff48'
'60f096c8eeefb23608d983162f94344db0b0a1f84":-10'
',"a8031adb0e4372374f15125f9a10e5b8017743895f49'
'5b43c1aa74839c9a5876":15},"s":"HrR8ThECGdeQ4Ug'
"aWLIMSqGx0fL_7FOf7KjX8tlWN6WBN6HVcojL3if3rN"
"bYDtDhDmwK1jxMViInpjc1hc-kOO5w6Ej7WxoqdmI0eTVj-"
"5iul5FMdGaGZUVuWtkq3A7A42Y5RAsgNwYtpVB44XifZ3W1f"
"MscefI8HovjFtWUm0caZkF6_Y_1iFr0FSWHgM95gx0pXkK"
"910VlKn0HqT8Dvo6ss7LMA5Cf-VS84q284Vsx6h3nqwT"
"gzo4Nx3V4d86VL45GqIzqbwKT0OMlM6DHKk2Pi8WxKZ"
"_QoHAMQI0AzBCJ6McdfjGf7lCjtLLRb4ClfZNTW0g"
'IX3dMSEj03mvOkgAAAX4abfqW"}'
),
"link": "next_link",
},
}
],
}

@property
def _user_identifier_sso_data(self) -> dict:
# assert self.flow_token[-1] == "1"
return {
"flow_token": self.flow_token,
"subtask_inputs":[{
"subtask_id":"LoginEnterUserIdentifierSSOSubtask",
"settings_list":{
"setting_responses":[{
"key":"user_identifier",
"response_data":{
"text_data":{
"result": self.username
"subtask_inputs": [
{
"subtask_id": "LoginEnterUserIdentifierSSOSubtask",
"settings_list": {
"setting_responses": [
{
"key": "user_identifier",
"response_data": {
"text_data": {"result": self.username}
},
}
}
}
],
"link":"next_link"
],
"link": "next_link",
},
}
}]
],
}

def get_task2_data(self) -> dict:
@property
def _account_dup_check_data(self) -> dict:
# assert self.flow_token[-1] == "2"
return {
"flow_token": self.flow_token,
"subtask_inputs":[
"subtask_inputs": [
{
"subtask_id":"AccountDuplicationCheck",
"check_logged_in_account":{
"link":"AccountDuplicationCheck_false"
}
"subtask_id": "AccountDuplicationCheck",
"check_logged_in_account": {
"link": "AccountDuplicationCheck_false"
},
}
]
],
}

def get_task6_data(self) -> dict:
@property
def _enter_password_data(self) -> dict:
# assert self.flow_token[-1] == "6"
return {
"flow_token": self.flow_token,
"subtask_inputs":[{
"subtask_id":"LoginEnterPassword",
"enter_password":{
"password": self.password,
"link":"next_link"
"subtask_inputs": [
{
"subtask_id": "LoginEnterPassword",
"enter_password": {"password": self.password, "link": "next_link"},
}
}]
],
}

def get_initial_params(self) -> dict:
@property
def _initial_params(self) -> dict:
return {
"input_flow_data":{
"flow_context":{
"debug_overrides":{

},
"start_location":{
"location":"splash_screen"
}
"input_flow_data": {
"flow_context": {
"debug_overrides": {},
"start_location": {"location": "splash_screen"},
}
},
"subtask_versions":{
"contacts_live_sync_permission_prompt":0,
"email_verification":1,
"topics_selector":1,
"wait_spinner":1,
"cta":4
}
"subtask_versions": {
"contacts_live_sync_permission_prompt": 0,
"email_verification": 1,
"topics_selector": 1,
"wait_spinner": 1,
"cta": 4,
},
}


if __name__ == "__main__":
from TwspaceDL import TwspaceDL
username = input("Username: ")
password = input("Password: ")

t = Login(username, password)
t.set_token(TwspaceDL.guest_token())
t.login()
print(t.get_token())
78 changes: 58 additions & 20 deletions twspace_dl/TwspaceDL.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
import requests

from .FormatInfo import FormatInfo
from .Login import Login


class TwspaceDL:
Expand All @@ -34,18 +35,8 @@ def from_space_url(cls, url: str, format_str: str):
return cls(space_id, format_str)

@classmethod
def from_user_url(cls, url: str, format_str: str):
screen_name = re.findall(r"(?<=twitter.com/)\w*", url)[0]
params = {
"variables": (
"{"
f'"screen_name":"{screen_name}",'
'"withSafetyModeUserFields":true,'
'"withSuperFollowsUserFields":true,'
'"withNftAvatar":false'
"}"
)
}
def from_user_tweets(cls, url: str, format_str: str):
user_id = TwspaceDL.user_id(url)
headers = {
"authorization": (
"Bearer "
Expand All @@ -54,14 +45,6 @@ def from_user_url(cls, url: str, format_str: str):
),
"x-guest-token": TwspaceDL.guest_token(),
}
response = requests.get(
"https://twitter.com/i/api/graphql/1CL-tn62bpc-zqeQrWm4Kw/UserByScreenName",
headers=headers,
params=params,
)
user_data = response.json()
user_id = user_data["data"]["user"]["result"]["rest_id"]

params = {
"variables": (
"{"
Expand Down Expand Up @@ -93,6 +76,61 @@ def from_user_url(cls, url: str, format_str: str):
raise RuntimeError("User is not live") from err
return cls(space_id, format_str)

@classmethod
def from_user_avatar(cls, user_url, format_str, username, password):
login = Login(username, password, TwspaceDL.guest_token())
auth_token = login.login()
headers = {
"authorization": (
"Bearer "
"AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs"
"=1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA"
),
"cookie": f"auth_token={auth_token};",
}
user_id = TwspaceDL.user_id(user_url)
r = requests.get(
f"https://twitter.com/i/api/fleets/v1/avatar_content?user_ids={user_id}&only_spaces=true",
headers=headers,
)

obj = r.json()
broadcast_id = obj["users"][user_id]["spaces"]["live_content"]["audiospace"][
"broadcast_id"
]
return cls(broadcast_id, format_str)

@staticmethod
def user_id(user_url: str) -> str:
screen_name = re.findall(r"(?<=twitter.com/)\w*", user_url)[0]

params = {
"variables": (
"{"
f'"screen_name":"{screen_name}",'
'"withSafetyModeUserFields":true,'
'"withSuperFollowsUserFields":true,'
'"withNftAvatar":false'
"}"
)
}
headers = {
"authorization": (
"Bearer "
"AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs"
"=1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA"
),
"x-guest-token": TwspaceDL.guest_token(),
}
response = requests.get(
"https://twitter.com/i/api/graphql/1CL-tn62bpc-zqeQrWm4Kw/UserByScreenName",
headers=headers,
params=params,
)
user_data = response.json()
user_id = user_data["data"]["user"]["result"]["rest_id"]
return user_id

@staticmethod
def guest_token() -> str:
guest_token = ""
Expand Down
Loading

0 comments on commit 04f1dfa

Please sign in to comment.