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/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/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 @@