add timeout and retry of session

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!")

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

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()

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