From 223f5a1ce21a6eea2a47e514365a55a5cf3b0f40 Mon Sep 17 00:00:00 2001 From: mirusu400 Date: Sun, 2 Jan 2022 20:48:46 +0900 Subject: [PATCH 1/2] Add manual login class --- twspace_dl/Login.py | 178 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 178 insertions(+) create mode 100644 twspace_dl/Login.py diff --git a/twspace_dl/Login.py b/twspace_dl/Login.py new file mode 100644 index 0000000..b2109f7 --- /dev/null +++ b/twspace_dl/Login.py @@ -0,0 +1,178 @@ +import requests + +class Login: + def __init__(self, username, password): + self.username = username + self.password = password + 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 + + 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() + ) + try: + self.flow_token = r.json()["flow_token"] + except: + print("Error while intiial_params:", r.json()) + return None + + # Task 0 + r = self.session.post(self.task_url, + headers = self.get_headers(), + json = self.get_task0_data() + ) + try: + self.flow_token = r.json()["flow_token"] + except: + print("Error while task0:", r.json()) + return None + + # Task 1 + r = self.session.post(self.task_url, + headers = self.get_headers(), + json = self.get_task1_data() + ) + try: + self.flow_token = r.json()["flow_token"] + except: + print("Error while task1:", r.json()) + return None + + # Task 2 + r = self.session.post(self.task_url, + headers = self.get_headers(), + json = self.get_task2_data() + ) + try: + self.flow_token = r.json()["flow_token"] + except: + print("Error while task2:", r.json()) + return None + + # Task 6 + r = self.session.post(self.task_url, + headers = self.get_headers(), + json = self.get_task6_data() + ) + try: + self.auth_token = str(r.cookies["auth_token"]) + except: + print("Error while task6:", r.json()) + return None + return self.auth_token + + def get_headers(self): + return { + "authorization": ( + "Bearer " + "AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs" + "=1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" + ), + "x-guest-token": self.guest_token, + } + + def get_task0_data(self) -> dict: + return { + "flow_token": self.flow_token, + "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: + # 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 + } + } + } + ], + "link":"next_link" + } + }] + } + + def get_task2_data(self) -> dict: + # assert self.flow_token[-1] == "2" + return { + "flow_token": self.flow_token, + "subtask_inputs":[ + { + "subtask_id":"AccountDuplicationCheck", + "check_logged_in_account":{ + "link":"AccountDuplicationCheck_false" + } + } + ] + } + + def get_task6_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" + } + }] + } + + def get_initial_params(self) -> dict: + return { + "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 + } + } + + +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()) \ No newline at end of file From 04f1dfa6e7728476e8a76f082ddd607d06c9dd06 Mon Sep 17 00:00:00 2001 From: Ryu1845 Date: Sun, 2 Jan 2022 21:36:21 +0100 Subject: [PATCH 2/2] change a lot of stuff --- twspace_dl/Login.py | 233 ++++++++++++++++++++-------------------- twspace_dl/TwspaceDL.py | 78 ++++++++++---- twspace_dl/__main__.py | 9 +- 3 files changed, 182 insertions(+), 138 deletions(-) diff --git a/twspace_dl/Login.py b/twspace_dl/Login.py index b2109f7..bd5c2b3 100644 --- a/twspace_dl/Login.py +++ b/twspace_dl/Login.py @@ -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 " @@ -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()) \ No newline at end of file diff --git a/twspace_dl/TwspaceDL.py b/twspace_dl/TwspaceDL.py index fad298d..75e0bea 100644 --- a/twspace_dl/TwspaceDL.py +++ b/twspace_dl/TwspaceDL.py @@ -13,6 +13,7 @@ import requests from .FormatInfo import FormatInfo +from .Login import Login class TwspaceDL: @@ -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 " @@ -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": ( "{" @@ -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 = "" diff --git a/twspace_dl/__main__.py b/twspace_dl/__main__.py index 7c2fe8b..037dd79 100644 --- a/twspace_dl/__main__.py +++ b/twspace_dl/__main__.py @@ -19,6 +19,8 @@ def get_args() -> argparse.Namespace: parser.add_argument("-v", "--verbose", action="store_true") parser.add_argument("-s", "--skip-download", action="store_true") parser.add_argument("-k", "--keep-files", action="store_true") + parser.add_argument("--username", type=str, metavar="USERNAME") + parser.add_argument("--password", type=str, metavar="PASSWORD") input_group.add_argument("-i", "--input-url", type=str, metavar="SPACE_URL") input_group.add_argument("-U", "--user-url", type=str, metavar="USER_URL") @@ -110,7 +112,12 @@ def main() -> None: if args.input_url: twspace_dl = TwspaceDL.from_space_url(args.input_url, args.output) elif args.user_url: - twspace_dl = TwspaceDL.from_user_url(args.user_url, args.output) + if args.username and args.password: + twspace_dl = TwspaceDL.from_user_avatar( + args.user_url, args.output, args.username, args.password + ) + else: + twspace_dl = TwspaceDL.from_user_tweets(args.user_url, args.output) else: with open(args.input_metadata, "r", encoding="utf-8") as metadata_io: metadata = json.load(metadata_io)