diff --git a/twspace_dl/Login.py b/twspace_dl/Login.py new file mode 100644 index 0000000..bd5c2b3 --- /dev/null +++ b/twspace_dl/Login.py @@ -0,0 +1,177 @@ +import requests +from typing import Optional + + +class Login: + def __init__(self, username, password, guest_token): + self.username = username + self.password = password + self.guest_token = guest_token + self.session = requests.Session() + self.task_url = "https://twitter.com/i/api/1.1/onboarding/task.json" + self.flow_token: str + + 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 = request_flow.json()["flow_token"] + except KeyError: + print("Error while intiial_params:", request_flow.json()) + return None + + # js instrumentation subtask + request_flow = self.session.post( + self.task_url, headers=self._headers, json=self._js_instrumentation_data + ) + try: + self.flow_token = request_flow.json()["flow_token"] + except KeyError: + print("Error while task0:", request_flow.json()) + return None + + # 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 = request_flow.json()["flow_token"] + except KeyError: + print("Error while task1:", request_flow.json()) + return None + + # account duplication check + request_flow = self.session.post( + self.task_url, headers=self._headers, json=self._account_dup_check_data + ) + try: + self.flow_token = request_flow.json()["flow_token"] + except KeyError: + print("Error while task2:", request_flow.json()) + return None + + # enter password + request_flow = self.session.post( + self.task_url, headers=self._headers, json=self._enter_password_data + ) + try: + auth_token = str(request_flow.cookies["auth_token"]) + except KeyError: + print("Error while task6:", request_flow.json()) + return None + return auth_token + + @property + def _headers(self): + return { + "authorization": ( + "Bearer " + "AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs" + "=1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" + ), + "x-guest-token": self.guest_token, + } + + @property + def _js_instrumentation_data(self) -> dict: + return { + "flow_token": self.flow_token, + "subtask_inputs": [ + { + "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} + }, + } + ], + "link": "next_link", + }, + } + ], + } + + @property + def _account_dup_check_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" + }, + } + ], + } + + @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"}, + } + ], + } + + @property + def _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, + }, + } 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)