diff --git a/twspace_dl/login.py b/twspace_dl/login.py index c8ca06b..2476486 100644 --- a/twspace_dl/login.py +++ b/twspace_dl/login.py @@ -3,6 +3,7 @@ from datetime import datetime, timedelta import requests +from requests.adapters import HTTPAdapter, Retry def is_expired(filename: str) -> bool: @@ -49,6 +50,7 @@ def __init__(self, username, password, guest_token): self.password = password self.guest_token = guest_token self.session = requests.Session() + self.session.mount('https://', HTTPAdapter(max_retries=(Retry(total=5, backoff_factor=0.1)))) self.task_url = "https://twitter.com/i/api/1.1/onboarding/task.json" self.flow_token: str @@ -198,6 +200,7 @@ def login(self) -> str: params={"flow_name": "login"}, headers=self._headers, json=self._initial_params, + timeout=30, ) try: self.flow_token = request_flow.json()["flow_token"] @@ -208,7 +211,7 @@ def login(self) -> str: # js instrumentation subtask request_flow = self.session.post( - self.task_url, headers=self._headers, json=self._js_instrumentation_data + self.task_url, headers=self._headers, json=self._js_instrumentation_data, timeout=30 ) try: self.flow_token = request_flow.json()["flow_token"] @@ -219,7 +222,7 @@ def login(self) -> str: # user identifier sso subtask request_flow = self.session.post( - self.task_url, headers=self._headers, json=self._user_identifier_sso_data + self.task_url, headers=self._headers, json=self._user_identifier_sso_data, timeout=30 ) try: self.flow_token = request_flow.json()["flow_token"] @@ -228,7 +231,7 @@ def login(self) -> str: # alternate identifier request_flow = self.session.post( - self.task_url, headers=self._headers, json=self._login_alternate_identifier + self.task_url, headers=self._headers, json=self._login_alternate_identifier, timeout=30 ).json() if "flow_token" in request_flow.keys(): self.flow_token = request_flow["flow_token"] @@ -239,7 +242,7 @@ def login(self) -> str: # enter password request_flow = self.session.post( - self.task_url, headers=self._headers, json=self._enter_password_data + self.task_url, headers=self._headers, json=self._enter_password_data, timeout=30 ) if "auth_token" in request_flow.cookies.keys(): return str(request_flow.cookies["auth_token"]) @@ -255,7 +258,7 @@ def login(self) -> str: # account duplication check request_flow = self.session.post( - self.task_url, headers=self._headers, json=self._account_dup_check_data + self.task_url, headers=self._headers, json=self._account_dup_check_data, timeout=30 ).json() if "auth_token" in request_flow.cookies.keys(): return str(request_flow.cookies["auth_token"]) @@ -268,7 +271,7 @@ def login(self) -> str: # 2FA request_flow = self.session.post( - self.task_url, headers=self._headers, json=self._enter_2fa + self.task_url, headers=self._headers, json=self._enter_2fa, timeout=30 ) if "auth_token" in request_flow.cookies.keys(): print("Success!") diff --git a/twspace_dl/twitter.py b/twspace_dl/twitter.py index 5b721de..bca9303 100644 --- a/twspace_dl/twitter.py +++ b/twspace_dl/twitter.py @@ -13,7 +13,7 @@ def guest_token() -> str: ) } response = requests.post( - "https://api.twitter.com/1.1/guest/activate.json", headers=headers + "https://api.twitter.com/1.1/guest/activate.json", headers=headers, timeout=30 ).json() token = response["guest_token"] if not token: @@ -28,6 +28,7 @@ def user_id(user_url: str) -> str: response = requests.get( "https://cdn.syndication.twimg.com/widgets/followbutton/info.json", params=params, + timeout=30, ) user_data = response.json() usr_id = user_data[0]["id"] diff --git a/twspace_dl/twspace.py b/twspace_dl/twspace.py index 093b2ef..cc8bbfe 100644 --- a/twspace_dl/twspace.py +++ b/twspace_dl/twspace.py @@ -90,6 +90,7 @@ def _metadata(space_id) -> dict: "https://twitter.com/i/api/graphql/jyQ0_DEMZHeoluCgHJ-U5Q/AudioSpaceById", params=params, headers=headers, + timeout=30, ) metadata = response.json() try: @@ -216,6 +217,7 @@ def from_user_tweets(cls, url: str): "https://twitter.com/i/api/graphql/jpCmlX6UgnPEZJknGKbmZA/UserTweets", params=params, headers=headers, + timeout=30, ) tweets = response.text @@ -244,6 +246,7 @@ def from_user_avatar(cls, user_url, auth_token): "https://twitter.com/i/api/fleets/v1/avatar_content", params=params, headers=headers, + timeout=30, ) if avatar_content_res.ok: avatar_content = avatar_content_res.json() diff --git a/twspace_dl/twspace_dl.py b/twspace_dl/twspace_dl.py index f973d88..344668a 100644 --- a/twspace_dl/twspace_dl.py +++ b/twspace_dl/twspace_dl.py @@ -8,6 +8,7 @@ from urllib.parse import urlparse import requests +from requests.adapters import Retry, HTTPAdapter from .twspace import Twspace @@ -21,6 +22,7 @@ def __init__(self, space: Twspace, format_str: str) -> None: self.space = space self.format_str = format_str or DEFAULT_FNAME_FORMAT self.session = requests.Session() + self.session.mount('https://', HTTPAdapter(max_retries=(Retry(total=5, backoff_factor=0.1)))) self._tempdir = tempfile.TemporaryDirectory(dir=".") @cached_property @@ -50,9 +52,10 @@ def dyn_url(self) -> str: "cookie": "auth_token=", } media_key = space["media_key"] - response = requests.get( + response = self.session.get( "https://twitter.com/i/api/1.1/live_video_stream/status/" + media_key, headers=headers, + timeout=30, ) try: metadata = response.json() @@ -72,7 +75,7 @@ def master_url(self) -> str: @property def playlist_url(self) -> str: """Get the URL containing the chunks filenames""" - response = requests.get(self.master_url) + response = requests.get(self.master_url, timeout=30) playlist_suffix = response.text.splitlines()[3] domain = urlparse(self.master_url).netloc playlist_url = f"https://{domain}{playlist_suffix}" @@ -81,7 +84,7 @@ def playlist_url(self) -> str: @property def playlist_text(self) -> str: """Modify the chunks URL using the master one to be able to download""" - playlist_text = requests.get(self.playlist_url).text + playlist_text = requests.get(self.playlist_url, timeout=30).text master_url_wo_file = re.sub(r"master_playlist\.m3u8.*", "", self.master_url) playlist_text = re.sub(r"(?=chunk)", master_url_wo_file, playlist_text) return playlist_text