RomanNum3ral
/
tubearchivist
mirror of https://github.com/tubearchivist/tubearchivist.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Reduce complexity in extract()

This commit is contained in:
Tyler Flowers 2026-03-22 17:14:18 -04:00 committed by GitHub
parent 95cdbae6c6
commit 778f6c2413
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 30 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
backend/download/src/yt_dlp_base.py
View File

@ -158,6 +158,28 @@ class YtWrap:
return True, True
def _handle_extract_dl_error(self, err, attempt, max_attempts) -> tuple[int, bool]:
"""
handle download errors in extract()
returns attempt count, continue, error
"""
if "This channel does not have a" in str(err):
return attempt, False, True
print(f"{url}: failed to get info from youtube: {err}")
if "Temporary failure in name resolution" in str(err):
raise ConnectionError("lost the internet, abort!") from err
if any(m in str(err) for m in self.BOT_MESSAGES):
print(self.BOT_ERROR_LOG)
if self.config["downloads"].get("gluetun_swap"):
attempt += 1
if (attempt == max_attempts):
raise ConnectionError("Reached maximum IP attempts!") from err
self._swap_ip()
return attempt, True, False
rand_sleep(self.config)
raise ConnectionError(self.BOT_ERROR_LOG) from err
def extract(self, url) -> tuple[dict | None, str | None]:
"""
make extract request
@ -181,23 +203,16 @@ class YtWrap:
print(f"{url}: failed to extract: {err}, continue...")
return None, str(err)
except yt_dlp.utils.DownloadError as err:
if "This channel does not have a" in str(err):
attempt, cont, error = self._handle_extract_dl_error(
err = err,
attempt = attempt,
max_attempts = max_attempts
)
if cont:
continue
elif error:
return None, None
print(f"{url}: failed to get info from youtube: {err}")
if "Temporary failure in name resolution" in str(err):
raise ConnectionError("lost the internet, abort!") from err
if any(m in str(err) for m in self.BOT_MESSAGES):
print(self.BOT_ERROR_LOG)
if self.config["downloads"].get("gluetun_swap"):
attempt += 1
if (attempt == max_attempts):
raise ConnectionError("Reached maximum IP attempts!") from err
self._swap_ip()
continue
rand_sleep(self.config)
raise ConnectionError(self.BOT_ERROR_LOG) from err
return None, str(err)
self._validate_cookie()