Merge pull request #27 from mirusu400/main

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,
+            },
+        }

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 = ""

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)