danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Merge pull request #263 from arabcoders/dev

Browse source 
More protection against stale downloads
This commit is contained in:
Abdulmohsen 2025-04-26 19:34:31 +03:00 committed by GitHub
commit 1bbaf461b2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 418 additions and 94 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
app/library/Download.py
View file

@ -218,6 +218,10 @@ class Download:
f"Live stream detected for '{self.info.title}', The following opts '{deletedOpts=}' have been deleted which are known to cause issues with live stream and post stream manifestless mode."
)
if isinstance(self.info_dict, dict) and len(self.info_dict.get("formats", [])) < 1:
msg = f"Failed to extract formats for '{self.info.url}'. The extracted info dict is empty."
raise ValueError(msg) # noqa: TRY301
self.logger.info(
f'Task id="{self.info.id}" PID="{os.getpid()}" title="{self.info.title}" preset="{self.preset}" cookies="{bool(params.get("cookiefile"))}" started.'
)

97
app/library/DownloadQueue.py
View file

@ -50,7 +50,8 @@ class DownloadQueue(metaclass=Singleton):
pool: AsyncPool | None = None
"""Pool of workers to download the files."""
_active_downloads: dict[str, Download] = {}
_active: dict[str, Download] = {}
"""Dictionary of active downloads."""
_instance = None
"""Instance of the DownloadQueue."""
@ -94,8 +95,8 @@ class DownloadQueue(metaclass=Singleton):
Scheduler.get_instance().add(
timer="* * * * *",
func=self.monitor_queue_for_stale_items,
id=f"{__class__.__name__}.{__class__.monitor_queue_for_stale_items.__name__}",
func=self.monitor_stale,
id=f"{__class__.__name__}.{__class__.monitor_stale.__name__}",
)
Scheduler.get_instance().add(
@ -176,10 +177,10 @@ class DownloadQueue(metaclass=Singleton):
async def on_shutdown(self, _: web.Application):
LOG.debug("Canceling all active downloads.")
if self._active_downloads:
if self._active:
self.pause()
try:
await self.cancel(list(self._active_downloads.keys()))
await self.cancel(list(self._active.keys()))
except Exception as e:
LOG.error(f"Failed to cancel downloads. {e!s}")
@ -603,14 +604,14 @@ class DownloadQueue(metaclass=Singleton):
items = {"queue": {}, "done": {}}
for k, v in self.queue.saved_items():
items["queue"][k] = v
items["queue"][k] = self._active[k].info if k in self._active else v
for k, v in self.done.saved_items():
items["done"][k] = v
for k, v in self.queue.items():
if k not in items["queue"]:
items["queue"][k] = v.info
items["queue"][k] = self._active[k].info if k in self._active else v
for k, v in self.done.items():
if k not in items["done"]:
@ -690,7 +691,7 @@ class DownloadQueue(metaclass=Singleton):
)
try:
self._active_downloads[entry.info._id] = entry
self._active[entry.info._id] = entry
await entry.start()
if "finished" != entry.info.status:
@ -703,8 +704,8 @@ class DownloadQueue(metaclass=Singleton):
entry.info.status = "error"
finally:
if entry.info._id in self._active_downloads:
self._active_downloads.pop(entry.info._id, None)
if entry.info._id in self._active:
self._active.pop(entry.info._id, None)
await entry.close()
@ -756,26 +757,73 @@ class DownloadQueue(metaclass=Singleton):
keys = ("playlist", "external_downloader", "live_in")
return any(key == k or key.startswith(k) for k in keys)
async def monitor_queue_for_stale_items(self):
async def monitor_stale(self):
"""
Monitor the queue for stale items and cancel them if needed.
Monitor the queue and pool for stale downloads and cancel them if needed.
"""
if self.is_paused() or self.queue.empty():
return
LOG.debug("Checking for stale items in the download queue.")
for id, item in list(self.queue.items()):
item_ref = f"{id=} {item.info.id=} {item.info.title=}"
if not item.is_stale():
LOG.debug(f"Item '{item_ref}' is not stale.")
continue
if not self.queue.empty():
LOG.debug("Checking for stale items in the download queue.")
for _id, item in list(self.queue.items()):
item_ref = f"{_id=} {item.info.id=} {item.info.title=}"
if not item.is_stale():
LOG.debug(f"Item '{item_ref}' is not stale.")
continue
LOG.warning(f"Cancelling staled item '{item_ref}' from download queue.")
try:
await self.cancel([id])
except Exception as e:
LOG.error(f"Failed to cancel staled item '{item_ref}'. {e!s}")
LOG.exception(e)
LOG.warning(f"Cancelling staled item '{item_ref}' from download queue.")
try:
await self.cancel([_id])
except Exception as e:
LOG.error(f"Failed to cancel staled item '{item_ref}'. {e!s}")
LOG.exception(e)
if self.pool:
time_now = datetime.now(tz=UTC)
workers = self.pool.get_workers_status()
if len(workers) > 0:
LOG.debug(f"Checking for stale workers. {len(workers)} workers found.")
for worker_id in workers:
worker = workers.get(worker_id, {})
if not worker:
continue
started = worker.get("started", None)
if not started:
LOG.debug(f"Worker '{worker_id}' not working yet.")
continue
started = datetime.fromisoformat(started)
if time_now - started < timedelta(minutes=5):
LOG.debug(f"Worker '{worker_id}' is not consider stale yet.")
continue
data = worker.get("data", {})
status = data.get("status", None)
if "preparing" != status:
LOG.debug(f"Worker '{worker_id}' not stuck. Status '{status}'.")
continue
_id = data.get("data._id", None)
if not _id:
LOG.debug(f"Worker '{worker_id}' has no id.")
continue
id = data.get("id", None)
title = data.get("title", None)
item_ref = f"{_id=} {id=} {title=}"
LOG.warning(f"Cancelling staled item '{item_ref}' from worker pool.")
try:
await self.cancel([_id])
except Exception as e:
LOG.error(f"Failed to cancel staled item '{item_ref}' from worker pool. {e!s}")
LOG.exception(e)
async def monitor_queue_live(self):
"""
@ -804,6 +852,7 @@ class DownloadQueue(metaclass=Singleton):
continue
starts_in = parsedate_to_datetime(item.info.live_in)
starts_in = starts_in.replace(tzinfo=UTC) if starts_in.tzinfo is None else starts_in.astimezone(UTC)
if time_now < (starts_in + timedelta(minutes=1)):
LOG.debug(f"Item '{item_ref}' is not yet live. will start in '{dt_delta(starts_in-time_now)}'.")

7
app/library/HttpAPI.py
View file

@ -1029,11 +1029,10 @@ class HttpAPI(Common):
"""
data: dict = {"queue": [], "history": []}
q = self.queue.get()
for _, v in self.queue.queue.saved_items():
data["queue"].append(v)
for _, v in self.queue.done.saved_items():
data["history"].append(v)
data["queue"].extend([q.get("queue", {}).get(k) for k in q.get("queue", {})])
data["history"].extend([q.get("done", {}).get(k) for k in q.get("done", {})])
return web.json_response(data=data, status=web.HTTPOk.status_code, dumps=self.encoder.encode)

3
app/library/Notifications.py
View file

@ -10,12 +10,13 @@ from typing import Any
import httpx
from aiohttp import web
from .ag_utils import ag
from .config import Config
from .encoder import Encoder
from .Events import Event, EventBus, Events
from .ItemDTO import ItemDTO
from .Singleton import Singleton
from .Utils import ag, validate_uuid
from .Utils import validate_uuid
from .version import APP_VERSION
LOG = logging.getLogger("notifications")

68
app/library/Utils.py
View file

@ -13,7 +13,6 @@ import uuid
from datetime import UTC, datetime, timedelta
from functools import lru_cache
from http.cookiejar import MozillaCookieJar
from typing import Any
import yt_dlp
from Crypto.Cipher import AES
@ -277,7 +276,7 @@ def load_file(file: str, check_type=None) -> tuple[dict | list, bool, str]:
opts = json.load(json_data)
if check_type:
assert isinstance(opts, check_type) # noqa: S101
assert isinstance(opts, check_type)
return (opts, True, "")
except Exception:
@ -288,7 +287,7 @@ def load_file(file: str, check_type=None) -> tuple[dict | list, bool, str]:
opts = json5_load(json_data)
if check_type:
assert isinstance(opts, check_type) # noqa: S101
assert isinstance(opts, check_type)
return (opts, True, "")
except AssertionError:
@ -327,63 +326,6 @@ def check_id(file: pathlib.Path) -> bool | str:
return False
def get_value(value):
return value() if callable(value) else value
def ag(array: dict | list, path: list[str | int] | str | int, default: Any = None, separator: str = ".") -> Any:
"""
dict/array getter: Retrieve a value from a nested dict or object using a path.
Args:
array (dict|list): dict-like or object from which to retrieve values.
path (list|str|int): Represents the path to retrieve:
- If None or empty string, returns the entire structure.
- If list, tries each path and returns the first found.
- If string, navigates through nested dict keys separated by `separator`.
default (Any): Value (or callable) returned if nothing is found.
separator (str): Separator for nested paths in strings.
Returns:
Any: The found value or the default if not found.
"""
if path is None or path == "":
return array
if not isinstance(array, dict):
try:
array = vars(array)
except TypeError:
array = dict(array)
if isinstance(path, list):
randomValue = str(uuid.uuid4())
for key in path:
val = ag(array, key, randomValue, separator)
if val != randomValue:
return val
return get_value(default)
if path in array and array[path] is not None:
return array[path]
if separator not in path:
return array.get(path, get_value(default))
keys = path.split(separator)
current = array
for segment in keys:
if isinstance(current, dict) and segment in current:
current = current[segment]
else:
return get_value(default)
return current
@lru_cache(maxsize=512)
def is_private_address(hostname: str) -> bool:
try:
@ -402,12 +344,12 @@ def validate_url(url: str) -> bool:
Args:
url (str): URL to validate.
Raises:
ValueError: If the URL is invalid or not allowed.
Returns:
bool: True if the URL is valid and allowed.
Raises:
ValueError: If the URL is invalid or not allowed.
"""
if not url:
msg = "URL is required."

329
app/library/ag_utils.py Normal file
View file

@ -0,0 +1,329 @@
from typing import Any
def get_value(value: Any) -> Any:
"""
Return the result of calling `value` if it is callable, else return `value`.
Args:
value: A value or a callable that returns a value.
Returns:
The original `value` if not callable, otherwise the result of calling it.
"""
return value() if callable(value) else value
def ag_set(data: dict[Any, Any], path: str, value: Any, separator: str = ".") -> dict[Any, Any]:
"""
Set a value into a nested dictionary at the specified path.
Args:
data: The dictionary to modify.
path: The key path (e.g., "a.b.c").
value: The value to assign at the specified path.
separator: The separator for splitting the path string.
Returns:
The modified dictionary.
Raises:
RuntimeError: If an intermediate value is not a dictionary.
"""
keys = path.split(separator)
at = data
while keys:
if len(keys) == 1:
if isinstance(at, dict):
at[keys.pop(0)] = value
else:
msg = f"Cannot set value at path '{path}' because '{at}' is not a dict."
raise RuntimeError(msg)
else:
key = keys.pop(0)
if key not in at or not isinstance(at[key], dict):
at[key] = {}
at = at[key]
return data
def ag(
array: dict[Any, Any] | list[Any] | object,
path: list[str | int] | str | int | None,
default: Any = None,
separator: str = ".",
) -> Any:
"""
Retrieve a value from a nested dict, list, or object using a path.
Args:
array: The structure to query (dict, list, or object).
path: The lookup path. If None or empty, returns the whole structure.
If list, tries each path in order and returns the first found.
If str or int, navigates nested structures using `separator`.
default: The value or callable to return if lookup fails.
separator: The separator for nested keys in `path`.
Returns:
The found value, or `default()` if `default` is callable, or `default` otherwise.
"""
if path is None or path == "":
return array
if not isinstance(array, dict | list):
try:
array = vars(array)
except TypeError:
return get_value(default)
if isinstance(array, list) and isinstance(path, int):
try:
val = array[path]
return val if val is not None else get_value(default)
except (IndexError, TypeError):
return get_value(default)
if isinstance(path, list):
_MISSING = object()
for key in path:
val = ag(array, key, default=_MISSING, separator=separator)
if val is not _MISSING:
return val
return get_value(default)
key = path
if isinstance(array, dict) and key in array and array[key] is not None:
return array[key]
if not isinstance(key, str) or separator not in key:
return get_value(default)
current = array
for segment in key.split(separator):
if isinstance(current, dict) and segment in current and current[segment] is not None:
current = current[segment]
elif isinstance(current, list):
try:
idx = int(segment)
except ValueError:
return get_value(default)
if 0 <= idx < len(current) and current[idx] is not None:
current = current[idx]
else:
return get_value(default)
else:
return get_value(default)
return current
def ag_sets(data: dict[Any, Any], path_values: dict[str, Any], separator: str = ".") -> dict[Any, Any]:
"""
Set multiple values into a nested dictionary.
Args:
data: The dictionary to modify.
path_values: A mapping of path strings to values.
separator: The separator for splitting each path.
Returns:
The modified dictionary.
"""
for path, value in path_values.items():
ag_set(data, path, value, separator)
return data
def ag_exists(data: dict[Any, Any] | list[Any] | object, path: str | int, separator: str = ".") -> bool:
"""
Check if a nested key or index exists and is not None.
Args:
data: The structure to check (dict, list, or object).
path: The key, index, or separator-delimited path string.
separator: The separator for splitting the path.
Returns:
True if the path exists and the value is not None, False otherwise.
"""
if not isinstance(data, dict | list):
try:
data = vars(data)
except TypeError:
return False
if isinstance(data, dict):
if path in data and data[path] is not None:
return True
elif isinstance(data, list) and isinstance(path, int):
return 0 <= path < len(data) and data[path] is not None
path_str = str(path)
segments = path_str.split(separator)
current = data
for seg in segments:
if isinstance(current, dict) and seg in current and current[seg] is not None:
current = current[seg]
elif isinstance(current, list):
try:
idx = int(seg)
except ValueError:
return False
if 0 <= idx < len(current) and current[idx] is not None:
current = current[idx]
else:
return False
else:
return False
return True
def ag_delete(
data: dict[Any, Any] | list[Any] | object, path: str | int | list[str | int], separator: str = "."
) -> dict[Any, Any] | list[Any]:
"""
Delete one or more keys or indices from a nested structure.
Args:
data: The structure to modify (dict, list, or object).
path: A key/index, a separator-delimited string path, or a list of such paths.
separator: The separator for splitting the path.
Returns:
The modified structure.
"""
if isinstance(path, list | tuple):
for p in path:
ag_delete(data, p, separator)
return data
if not isinstance(data, dict | list):
try:
data = vars(data)
except TypeError:
return data # type: ignore
if isinstance(data, dict) and path in data:
del data[path]
return data
if isinstance(data, list) and isinstance(path, int):
if 0 <= path < len(data):
data.pop(path)
return data
path_str = str(path)
segments = path_str.split(separator)
last = segments.pop()
current = data
for seg in segments:
if isinstance(current, dict) and seg in current:
current = current[seg]
elif isinstance(current, list):
try:
idx = int(seg)
except ValueError:
current = None
break
if 0 <= idx < len(current):
current = current[idx]
else:
current = None
break
else:
current = None
break
if current is None:
return data
if isinstance(current, dict) and last in current:
del current[last]
elif isinstance(current, list):
try:
idx = int(last)
if 0 <= idx < len(current):
current.pop(idx)
except (ValueError, IndexError):
pass
return data
def run_tests():
def test_ag_set_basic():
d = {"a": {"b": 1}}
assert ag_set(d, "a.c.d", 42) == {"a": {"b": 1, "c": {"d": 42}}}
def test_ag_set_overwrite_error():
try:
ag_set({"a": 1}, "a.b", 2)
except RuntimeError as e:
assert "Cannot set value at path" in str(e) # noqa: PT017
def test_ag_get_value_callable_and_value():
assert get_value(5) == 5
assert get_value(lambda: 7) == 7
def test_ag_simple_dict():
d = {"x": 10}
assert ag(d, "x") == 10
assert ag(d, "y", default=0) == 0
def test_ag_nested():
d = {"a": {"b": {"c": 3}}}
assert ag(d, "a.b.c") == 3
assert ag(d, "a.b.x", default="no") == "no"
def test_ag_list_index():
lst = [1, 2, None]
assert ag(lst, 1) == 2
assert ag(lst, 2, default=0) == 0
def test_ag_list_of_paths():
d = {"a": 1}
assert ag(d, ["x", "a"], default=9) == 1
def test_ag_sets():
d = {}
ag_sets(d, {"u.v": 5, "u.w": 6, "z": 7})
assert d == {"u": {"v": 5, "w": 6}, "z": 7}
def test_ag_exists():
d = {"a": {"b": None, "c": 2}, "lst": [0, None]}
assert not ag_exists(d, "a.b")
assert ag_exists(d, "a.c")
assert not ag_exists(d, "a.x")
assert ag_exists(d, "lst.0")
assert not ag_exists(d, "lst.1")
def test_ag_delete_basic():
d = {"a": {"b": 1, "c": 2}, "x": 5}
ag_delete(d, "a.b")
assert d == {"a": {"c": 2}, "x": 5}
def test_ag_delete_list_and_multiple():
lst = [10, 20, {"n": [1, 2, 3]}]
ag_delete(lst, 1)
assert lst == [10, {"n": [1, 2, 3]}]
ag_delete(lst, "1.n.2")
assert lst == [10, {"n": [1, 2]}]
def test_ag_delete_multiple_paths():
data = {"u": {"v": 100}, "w": 200}
ag_delete(data, ["u.v", "w"])
assert data == {"u": {}}
for test in [
test_ag_set_basic,
test_ag_set_overwrite_error,
test_ag_get_value_callable_and_value,
test_ag_simple_dict,
test_ag_nested,
test_ag_list_index,
test_ag_list_of_paths,
test_ag_sets,
test_ag_exists,
test_ag_delete_basic,
test_ag_delete_list_and_multiple,
test_ag_delete_multiple_paths,
]:
try:
test()
print(f"PASS: {test.__name__}") # noqa: T201
except AssertionError:
print(f"FAIL {test.__name__}") # noqa: T201
if __name__ == "__main__":
run_tests()

1
app/library/config.py
View file

@ -158,7 +158,6 @@ class Config:
"https://unsplash.it/1920/1080?random",
"https://picsum.photos/1920/1080",
"https://spaceholder.cc/i/1920x1080",
"https://imageipsum.com/1920x1080",
"https://placedog.net/1920/1080",
]
"The list of picture backends to use for the background."

3
pyproject.toml
View file

@ -107,7 +107,8 @@ ignore = [
"PLR0912",
"PLR0915",
"PLW2901",
"ERA001"
"ERA001",
"S101",
]
# Allow fix for all enabled rules (when `--fix`) is provided.