diff --git a/.vscode/settings.json b/.vscode/settings.json index 2d265457..289067c3 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -28,6 +28,7 @@ "daterange", "dotenv", "edgechromium", + "engineio", "euuo", "finaldir", "flac", diff --git a/app/library/Download.py b/app/library/Download.py index 92c0e0d5..958de882 100644 --- a/app/library/Download.py +++ b/app/library/Download.py @@ -469,6 +469,8 @@ class Download: ff = await ffprobe(status.get("filename")) self.info.extras["is_video"] = ff.has_video() self.info.extras["is_audio"] = ff.has_audio() + if (ff.has_video() or ff.has_audio()) and not self.info.extras.get("duration"): + self.info.extras["duration"] = int(float(ff.metadata.get("duration", 0.0))) except Exception as e: self.info.extras["is_video"] = True self.info.extras["is_audio"] = True diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index 03185baa..d567809f 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -5,12 +5,13 @@ import logging from collections.abc import Awaitable from datetime import UTC, datetime, timedelta from pathlib import Path -from typing import Any import anyio from aiohttp import web from aiohttp.web import Request, RequestHandler, Response +from app.library.Services import Services + from .cache import Cache from .config import Config from .DownloadQueue import DownloadQueue @@ -24,27 +25,32 @@ LOG: logging.Logger = logging.getLogger("http_api") class HttpAPI: - di_context: dict[str, Any] = {} - def __init__(self, root_path: Path, queue: DownloadQueue): self.queue: DownloadQueue = queue or DownloadQueue.get_instance() self.encoder: Encoder = Encoder() self.config: Config = Config.get_instance() self._notify: EventBus = EventBus.get_instance() - self.rootPath: Path = root_path self.cache = Cache() self.app: web.Application | None = None - self.di_context: dict[str, Any] = { - "queue": self.queue, - "encoder": self.encoder, - "config": self.config, - "notify": self._notify, - "cache": self.cache, - "app": self.app, - "http_api": self, - "root_path": self.rootPath, - } + + services = Services.get_instance() + services.add_all( + { + k: v + for k, v in { + "queue": self.queue, + "encoder": self.encoder, + "config": self.config, + "notify": self._notify, + "cache": self.cache, + "app": self.app, + "http_api": self, + "root_path": self.rootPath, + }.items() + if not services.has(k) + } + ) async def on_shutdown(self, _: web.Application): pass @@ -297,8 +303,6 @@ class HttpAPI: Response: The response object. """ - context = {**self.di_context.copy(), "request": request} - try: sig = inspect.signature(handler) expected_args = sig.parameters.keys() @@ -307,8 +311,7 @@ class HttpAPI: if 1 == len(expected_args) and "request" in expected_args: response = await handler(request) else: - filtered = {k: v for k, v in context.items() if k in expected_args} - response = await handler(**filtered) + response = await Services.get_instance().handle_async(handler, request=request) except TypeError as te: LOG.exception(te) if "missing 1 required positional argument" in str(te) and "request" in str(te): diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 7c00e212..190221ec 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -1,5 +1,4 @@ import functools -import inspect import logging from pathlib import Path from typing import Any @@ -8,6 +7,7 @@ import socketio from aiohttp import web from app.library.router import RouteType, get_routes +from app.library.Services import Services from app.library.Utils import load_modules from .config import Config @@ -41,7 +41,6 @@ class HttpSocket: self.queue = queue or DownloadQueue.get_instance() self._notify = EventBus.get_instance() - # logger=True, engineio_logger=True, self.sio = sio or socketio.AsyncServer( async_handlers=True, async_mode="aiohttp", @@ -58,14 +57,21 @@ class HttpSocket: def emit(e: Event, _, **kwargs): return self.sio.emit(event=e.event, data=encoder.encode(e.data), **kwargs) - self.di_context = { - "config": self.config, - "queue": self.queue, - "sio": self.sio, - "encoder": encoder, - "notify": self._notify, - "root_path": self.rootPath, - } + services = Services.get_instance() + services.add_all( + { + k: v + for k, v in { + "config": self.config, + "queue": self.queue, + "sio": self.sio, + "encoder": encoder, + "notify": self._notify, + "root_path": self.rootPath, + }.items() + if not services.has(k) + } + ) self._notify.subscribe("frontend", emit, f"{__class__.__name__}.emit") @@ -107,23 +113,11 @@ class HttpSocket: LOG.debug( f"Add ({route.name}) {route.method.value if isinstance(route.method,RouteType) else route.method}: {route.path}." ) - self.sio.on(route.path)(HttpSocket._injector(route.handler, route.path, self.di_context)) + self.sio.on(route.path)(HttpSocket._injector(route.handler, route.path)) @staticmethod - def _injector(func, event: str, container: dict): - sig: inspect.Signature = inspect.signature(func) - + def _injector(func, event: str): async def wrapper(sid, data, **kwargs): - args = {} - - merged = {**container, "sid": sid, "data": data, "event_name": event} - if isinstance(kwargs, dict): - merged.update(kwargs) - - for name in sig.parameters: - if name in merged: - args[name] = merged[name] - - return await func(**args) + return await Services.get_instance().handle_async(func, sid=sid, data=data, event=event, **kwargs) return wrapper diff --git a/app/library/Services.py b/app/library/Services.py new file mode 100644 index 00000000..dabfa640 --- /dev/null +++ b/app/library/Services.py @@ -0,0 +1,61 @@ +import inspect +from typing import Any, TypeVar + +from app.library.Singleton import Singleton + +T = TypeVar("T") + + +class Services(metaclass=Singleton): + _dct: dict[str, T] = {} + + _instance = None + """The instance of the class.""" + + @staticmethod + def get_instance() -> "Services": + if Services._instance is None: + Services._instance = Services() + + return Services._instance + + def add(self, name: str, service: T): + self._dct[name] = service + + def add_all(self, services: dict[str, T]): + for name, service in services.items(): + self.add(name, service) + + def get(self, name: str) -> T | None: + return self._dct.get(name) + + def has(self, name: str) -> bool: + return name in self._dct + + def remove(self, name: str): + if name not in self._dct: + return + + self._dct.pop(name, None) + + def clear(self): + self._dct.clear() + + def get_all(self) -> dict[str, T]: + return self._dct.copy() + + async def handle_async(self, handler: callable, **kwargs) -> Any: + context = {**self.get_all(), **kwargs} + + sig = inspect.signature(handler) + expected_args = sig.parameters.keys() + filtered = {k: v for k, v in context.items() if k in expected_args} + return await handler(**filtered) + + def handle_sync(self, handler: callable, **kwargs) -> Any: + context = {**self.get_all(), **kwargs} + + sig = inspect.signature(handler) + expected_args = sig.parameters.keys() + filtered = {k: v for k, v in context.items() if k in expected_args} + return handler(**filtered) diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 8981fc0b..a20dfdb7 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -1,15 +1,18 @@ import asyncio +import inspect import json import logging +import pkgutil import time from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from typing import Any -import httpx from aiohttp import web +from app.library.Services import Services + from .config import Config from .encoder import Encoder from .Events import EventBus, Events, error, info, success @@ -58,7 +61,6 @@ class Tasks(metaclass=Singleton): loop: asyncio.AbstractEventLoop | None = None, config: Config | None = None, encoder: Encoder | None = None, - client: httpx.AsyncClient | None = None, scheduler: Scheduler | None = None, ): Tasks._instance = self @@ -68,11 +70,11 @@ class Tasks(metaclass=Singleton): self._debug: bool = config.debug self._default_preset: str = config.default_preset self._file: Path = Path(file) if file else Path(config.config_path).joinpath("tasks.json") - self._client: httpx.AsyncClient = client or httpx.AsyncClient() self._encoder: Encoder = encoder or Encoder() self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_event_loop() self._scheduler: Scheduler = scheduler or Scheduler.get_instance() self._notify: EventBus = EventBus.get_instance() + self._task_handler = HandleTask(self._scheduler, self) if self._file.exists() and "600" != self._file.stat().st_mode: try: @@ -301,7 +303,9 @@ class Tasks(metaclass=Singleton): LOG.info(f"Dispatched '{task.id}: {task.name}' at '{timeNow}'.") tasks: list = [ - self._notify.emit(Events.LOG_INFO, data=info(f"Dispatched '{task.name}' at '{timeNow}'.")), + self._notify.emit( + Events.LOG_INFO, data=info(f"Dispatched '{task.name}' at '{timeNow}'.", data={"lowPriority": True}) + ), self._notify.emit( Events.ADD_URL, data={ @@ -324,10 +328,89 @@ class Tasks(metaclass=Singleton): await self._notify.emit( Events.LOG_SUCCESS, - data=success(f"Completed '{task.name}' in '{ended - started:.2f}' seconds."), + data=success( + f"Completed '{task.name}' in '{ended - started:.2f}' seconds.", data={"lowPriority": True} + ), ) except Exception as e: LOG.error(f"Failed to execute '{task.id}: {task.name}' at '{timeNow}'. '{e!s}'.") await self._notify.emit( Events.ERROR, data=error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'.") ) + + +class HandleTask: + _tasks: Tasks + + def __init__(self, scheduler: Scheduler, tasks: Tasks) -> None: + self._tasks = tasks + self._handlers: list[type] = self._discover() + + scheduler.add( + timer="15 */1 * * *", + func=self._dispatcher, + id=f"{__class__.__name__}._dispatcher", + ) + + def _dispatcher(self): + for task in self._tasks.get_all(): + if not task.timer or "[no_handler]" in task.name: + continue + + try: + handler = self._find_handler(task) + if handler is None: + continue + + asyncio.create_task(self.dispatch(task, handler=handler), name=f"task-{task.id}") + except Exception as e: + LOG.error(f"Failed to handle task '{task.id}: {task.name}'. '{e!s}'.") + + def _find_handler(self, task: Task) -> type | None: + for cls in self._handlers: + try: + if Services.get_instance().handle_sync(handler=cls.can_handle, task=task): + return cls + except Exception as e: + LOG.exception(e) + continue + + return None + + async def dispatch(self, task: Task, handler: type | None = None, **kwargs) -> Any | None: + """ + Dispatch a task to the appropriate handler. + + Args: + task (Task): The task to dispatch. + handler (type|None): Optional specific handler to use instead of finding one. + **kwargs: Additional context to pass to the handler. + + Returns: + Any|None: The result of the handler's execution, or None if no handler found. + + """ + if not handler: + handler = self._find_handler(task) + if handler is None: + return None + + return await Services.get_instance().handle_async(handler=handler.handle, task=task, **kwargs) + + def _discover(self) -> list[type]: + import importlib + + import app.library.task_handlers as handlers_pkg + + handlers: list[type] = [] + + for _, module_name, _ in pkgutil.iter_modules(handlers_pkg.__path__): + module = importlib.import_module(f"{handlers_pkg.__name__}.{module_name}") + for _, cls in inspect.getmembers(module, inspect.isclass): + if cls.__module__ != module.__name__: + continue + + if callable(getattr(cls, "can_handle", None)) and callable(getattr(cls, "handle", None)): + handlers.append(cls) + + return handlers diff --git a/app/library/ffprobe.py b/app/library/ffprobe.py index 3475aea2..cb7d18da 100644 --- a/app/library/ffprobe.py +++ b/app/library/ffprobe.py @@ -234,7 +234,7 @@ class FFProbeResult: def __repr__(self): return "".format(**vars(self)) - def unserialize(self, data: dict): + def deserialize(self, data: dict): self.metadata = data.get("metadata", {}) self.video = [FFStream(v) for v in data.get("video", [])] self.audio = [FFStream(a) for a in data.get("audio", [])] diff --git a/app/library/task_handlers/youtube.py b/app/library/task_handlers/youtube.py new file mode 100644 index 00000000..a4f5f0ff --- /dev/null +++ b/app/library/task_handlers/youtube.py @@ -0,0 +1,217 @@ +import asyncio +import logging +import re +from pathlib import Path + +from app.library.cache import Cache +from app.library.config import Config +from app.library.DownloadQueue import DownloadQueue +from app.library.Events import EventBus, Events +from app.library.Tasks import Task +from app.library.Utils import is_downloaded +from app.library.YTDLPOpts import YTDLPOpts + +LOG: logging.Logger = logging.getLogger(__name__) + + +class YoutubeHandler: + queued_ids: set[str] = set() + + FEED_CHANNEL = "https://www.youtube.com/feeds/videos.xml?channel_id={id}" + FEED_PLAYLIST = "https://www.youtube.com/feeds/videos.xml?playlist_id={id}" + + CHANNEL_REGEX = re.compile(r"^https?://(?:www\.)?youtube\.com/(?:channel/(?PUC[0-9A-Za-z_-]{22})|)/?$") + + PLAYLIST_REGEX = re.compile(r"^https?://(?:www\.)?youtube\.com/(?:playlist\?list=(?P[A-Za-z0-9_-]+)|).*$") + + @staticmethod + def can_handle(task: Task, config: Config) -> bool: + has, _ = YoutubeHandler.has_archive(task, config) + if not has: + LOG.debug(f"Task '{task.id}: {task.name}' does not have an archive file configured.") + return False + + LOG.debug(f"Checking if task '{task.id}: {task.name}' can handle YouTube URL: {task.url}") + return YoutubeHandler.parse(task.url) is not None + + @staticmethod + async def handle(task: Task, cache: Cache, notify: EventBus, config: Config, queue: DownloadQueue): + """ + Fetch the Atom feed for a YouTube channel or playlist, parse entries, + and return a list of videos with metadata. + + Args: + task (Task): The task containing the YouTube URL. + cache (Cache): The cache instance for storing feed data. + notify (EventBus): The event bus for notifications. + config (Config): The configuration instance. + queue (DownloadQueue): The download queue instance. + + """ + archive_file, params = YoutubeHandler.has_archive(task, config) + if not archive_file: + LOG.error(f"Task '{task.id}: {task.name}' does not have an archive file configured.") + return + + import httpx + from defusedxml.ElementTree import fromstring + + parsed = YoutubeHandler.parse(task.url) + if not parsed: + LOG.error(f"Cannot parse URL: {task.url}") + return + + feed_id = parsed["id"] + feed_type = parsed["type"] + + if "channel" == feed_type: + feed_url = YoutubeHandler.FEED_CHANNEL.format(id=feed_id) + else: + feed_url = YoutubeHandler.FEED_PLAYLIST.format(id=feed_id) + + cache_key = f"youtube_feed_{feed_id}" + data = cache.get(cache_key) + if not data: + LOG.info(f"Fetching '{task.id}: {task.name}' feed.") + opts = { + "proxy": params.get("proxy", None), + "headers": { + "User-Agent": params.get( + "user_agent", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36", + ) + }, + } + async with httpx.AsyncClient(**opts) as client: + response = await client.request(method="GET", url=feed_url, timeout=60) + response.raise_for_status() + data = response.text + cache.set(cache_key, data, ttl=3600) + else: + LOG.info(f"Using cached '{task.id}: {task.name}' feed.") + + root = fromstring(data) + ns = {"atom": "http://www.w3.org/2005/Atom", "yt": "http://www.youtube.com/xml/schemas/2015"} + items = [] + + for entry in root.findall("atom:entry", ns): + vid_elem = entry.find("yt:videoId", ns) + title_elem = entry.find("atom:title", ns) + pub_elem = entry.find("atom:published", ns) + vid = vid_elem.text if vid_elem is not None else "" + title = title_elem.text if title_elem is not None else "" + published = pub_elem.text if pub_elem is not None else "" + items.append( + { + "id": vid, + "url": f"https://www.youtube.com/watch?v={vid}", + "title": title, + "published": published, + } + ) + + if len(items) < 1: + LOG.warning(f"No entries found in '{task.id}: {task.name}' feed. URL: {feed_url}") + return + + filtered = [] + for item in items: + status, _ = is_downloaded(archive_file, url=item["url"]) + if status is True or item["id"] in YoutubeHandler.queued_ids: + continue + + if queue.done.exists(url=item["url"]) or queue.queue.exists(url=item["url"]): + LOG.debug(f"Item '{item['id']}' already exists in the queue or download history.") + YoutubeHandler.queued_ids.add(item["id"]) + continue + + YoutubeHandler.queued_ids.add(item["id"]) + filtered.append(item) + + if len(filtered) < 1: + LOG.info(f"No new items found in '{task.id}: {task.name}' feed.") + return + + LOG.info(f"Found '{len(filtered)}' new items from '{task.id}: {task.name}' feed.") + + preset: str = str(task.preset or config.default_preset) + folder: str = task.folder if task.folder else "" + template: str = task.template if task.template else "" + cli: str = task.cli if task.cli else "" + + queued = asyncio.gather( + *[ + notify.emit( + Events.ADD_URL, + data={"url": item["url"], "preset": preset, "folder": folder, "template": template, "cli": cli}, + ) + for item in filtered + ] + ) + + try: + await queued + except Exception as e: + LOG.error(f"Error while adding items to the queue: {e!s}") + return + + @staticmethod + def has_archive(task: Task, config: Config) -> tuple[Path | None, dict]: + archive_file: Path | None = Path(config.archive_file) if config.keep_archive else None + params: YTDLPOpts = YTDLPOpts.get_instance() + + if task.preset: + params.preset(name=task.preset) + + if task.cli: + params.add_cli(task.cli, from_user=True) + + params = params.get_all() + if user_archive_file := params.get("download_archive", None): + archive_file = Path(user_archive_file) + + if not archive_file: + return (None, params) + + if not archive_file.exists(): + archive_file.parent.mkdir(parents=True, exist_ok=True) + archive_file.touch(exist_ok=True) + + return (archive_file, params) + + @staticmethod + def parse(url: str) -> dict[str, str] | None: + """ + Parse YouTube channel or playlist URL. + + Args: + url (str): The YouTube URL to parse. + + Returns: + {'type': 'channel', 'id': } + {'type': 'playlist', 'id': } + None if the URL is neither. + + """ + if m := YoutubeHandler.CHANNEL_REGEX.match(url): + return {"type": "channel", "id": m.group("id")} + + if m := YoutubeHandler.PLAYLIST_REGEX.match(url): + return {"type": "playlist", "id": m.group("id")} + + return None + + @staticmethod + def tests() -> list[str]: + """ + Return a list of test URLs to validate the parsing logic. + """ + return [ + "https://www.youtube.com/channel/UCabc123ABCDEFGHIJKLMN", + "https://youtube.com/c/MyCustomName", + "https://youtube.com/user/SomeUser123", + "https://youtube.com/@SomeHandle", + "https://youtube.com/playlist?list=PLxyz789ABCDEFGHIJ", + "https://youtube.com/watch?v=foo&list=PLxyz789ABCDEFGHIJ", + "https://youtube.com/watch?v=foo", + ] diff --git a/pyproject.toml b/pyproject.toml index 0d9e1cd2..a10a39da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,7 @@ dependencies = [ "yt-dlp", "platformdirs", "dateparser>=1.2.1", + "defusedxml>=0.7.1", ] [tool.ruff] diff --git a/ui/components/History.vue b/ui/components/History.vue index 822650f9..88508196 100644 --- a/ui/components/History.vue +++ b/ui/components/History.vue @@ -649,10 +649,10 @@ const setStatus = item => { } if (item.extras?.is_premiere) { - return 'Premiere' + return 'Premiered' } - return item.is_live ? 'Live Ended' : 'Completed' + return item.is_live ? 'Streamed' : 'Completed' } if ('error' === item.status) { @@ -660,14 +660,14 @@ const setStatus = item => { } if ('cancelled' === item.status) { - return display_style.value === 'cards' ? 'User Cancelled' : 'Cancelled' + return 'Cancelled' } if ('not_live' === item.status) { if (item.extras?.is_premiere) { return 'Premiere' } - return display_style.value === 'cards' ? 'Live Stream' : 'Live' + return display_style.value === 'cards' ? 'Stream' : 'Live' } return item.status diff --git a/ui/components/PresetForm.vue b/ui/components/PresetForm.vue index 44116009..d4b56f73 100644 --- a/ui/components/PresetForm.vue +++ b/ui/components/PresetForm.vue @@ -28,7 +28,7 @@
diff --git a/ui/components/Queue.vue b/ui/components/Queue.vue index 95524980..8d0d3950 100644 --- a/ui/components/Queue.vue +++ b/ui/components/Queue.vue @@ -310,15 +310,15 @@ const setStatus = item => { } if ('downloading' === item.status && item.is_live) { - return 'Live Streaming' + return 'Streaming' } if ('preparing' === item.status) { - return ag(item, 'extras.external_downloader') ? 'External DL' : 'Preparing..'; + return ag(item, 'extras.external_downloader') ? 'External-DL' : 'Preparing..'; } if (!item.status) { - return 'Unknown..' + return 'Unknown...' } return ucFirst(item.status) diff --git a/ui/components/TaskForm.vue b/ui/components/TaskForm.vue index 3e5b5dc7..a7cace0e 100644 --- a/ui/components/TaskForm.vue +++ b/ui/components/TaskForm.vue @@ -1,5 +1,22 @@