From 9d5d6595fc5caf497a1a68d714128014525c4581 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Wed, 3 Sep 2025 22:45:33 +0300 Subject: [PATCH] FEAT: new TwitchHandler --- app/library/Tasks.py | 7 + app/library/task_handlers/_base_handler.py | 105 ++++++++++++ app/library/task_handlers/twitch.py | 189 +++++++++++++++++++++ app/library/task_handlers/youtube.py | 142 +++++----------- 4 files changed, 346 insertions(+), 97 deletions(-) create mode 100644 app/library/task_handlers/_base_handler.py create mode 100644 app/library/task_handlers/twitch.py diff --git a/app/library/Tasks.py b/app/library/Tasks.py index 7cc2af27..59c60a9c 100644 --- a/app/library/Tasks.py +++ b/app/library/Tasks.py @@ -504,6 +504,10 @@ class HandleTask: if not task.handler_enabled: continue + if not task.get_ytdlp_opts().get_all().get("download_archive"): + LOG.debug(f"Task '{task.name}' does not have an archive file configured.") + continue + try: handler = self._find_handler(task) if handler is None: @@ -565,6 +569,9 @@ class HandleTask: handlers: list[type] = [] for _, module_name, _ in pkgutil.iter_modules(handlers_pkg.__path__): + if module_name.startswith("_"): + continue + module = importlib.import_module(f"{handlers_pkg.__name__}.{module_name}") for _, cls in inspect.getmembers(module, inspect.isclass): if cls.__module__ != module.__name__: diff --git a/app/library/task_handlers/_base_handler.py b/app/library/task_handlers/_base_handler.py new file mode 100644 index 00000000..5c79a6be --- /dev/null +++ b/app/library/task_handlers/_base_handler.py @@ -0,0 +1,105 @@ +# flake8: noqa: ARG004 +import logging +from typing import Any + +import httpx +from yt_dlp.utils.networking import random_user_agent + +from app.library.config import Config +from app.library.DownloadQueue import DownloadQueue +from app.library.Events import EventBus, Events +from app.library.ItemDTO import ItemDTO +from app.library.Tasks import Task + +LOG: logging.Logger = logging.getLogger(__name__) + + +class BaseHandler: + queued: set[str] = set() + failure_count: dict[str, int] = {} + + def __init_subclass__(cls, **kwargs): + """Ensure each subclass has its own state containers.""" + super().__init_subclass__(**kwargs) + if "queued" not in cls.__dict__: + cls.queued = set() + if "failure_count" not in cls.__dict__: + cls.failure_count = {} + + EventBus.get_instance().subscribe( + Events.ITEM_ERROR, + lambda data, _, **__: cls.on_error(data.data), + f"{cls.__name__}.on_error", + ) + + @staticmethod + def can_handle(task: Task) -> bool: + return False + + @staticmethod + async def handle(task: Task, notify: EventBus, config: Config, queue: DownloadQueue): + pass + + @staticmethod + def parse(url: str) -> Any | None: + return None + + @classmethod + async def on_error(cls, item: ItemDTO) -> None: + """ + Handle errors by logging them and removing the queued ID if it exists. + + Args: + item (ItemDTO): The error data containing the URL and other information. + + """ + if not item or not isinstance(item, ItemDTO): + return + + if not item.archive_id or not cls.failure_count.get(item.archive_id, None): + LOG.debug(f"Item '{item.name()}' not queued by the handler.") + return + + failCount: int = int(cls.failure_count.get(item.archive_id, 0)) + + LOG.info(f"Removing '{item.name()}' from queued IDs due to error. Failure count: '{failCount + 1}'.") + if item.archive_id in cls.queued: + cls.queued.remove(item.archive_id) + + cls.failure_count[item.archive_id] = 1 + failCount + + @staticmethod + def tests() -> list[tuple[str, bool]]: + return [] + + @staticmethod + async def request(url: str, headers: dict | None = None, ytdlp_opts: dict | None = None) -> httpx.Response: + headers = {} if not isinstance(headers, dict) else headers + ytdlp_opts = {} if not isinstance(ytdlp_opts, dict) else ytdlp_opts + + opts: dict[str, Any] = { + "headers": { + "User-Agent": random_user_agent(), + }, + } + + try: + from httpx_curl_cffi import AsyncCurlTransport, CurlOpt + + opts["transport"] = AsyncCurlTransport( + impersonate="chrome", + default_headers=True, + curl_options={CurlOpt.FRESH_CONNECT: True}, + ) + opts["headers"].pop("User-Agent", None) + except Exception: + pass + + for k, v in headers.items(): + opts["headers"][k] = v + + if proxy := ytdlp_opts.get("proxy", None): + opts["proxy"] = proxy + + async with httpx.AsyncClient(**opts) as client: + return await client.request(method="GET", url=url, timeout=ytdlp_opts.get("socket_timeout", 120)) diff --git a/app/library/task_handlers/twitch.py b/app/library/task_handlers/twitch.py new file mode 100644 index 00000000..f764041a --- /dev/null +++ b/app/library/task_handlers/twitch.py @@ -0,0 +1,189 @@ +import asyncio +import logging +import re +from typing import TYPE_CHECKING +from xml.etree.ElementTree import Element + +from app.library.DownloadQueue import DownloadQueue +from app.library.Events import EventBus, Events +from app.library.ItemDTO import Item +from app.library.Tasks import Task +from app.library.Utils import archive_read, get_archive_id + +from ._base_handler import BaseHandler + +if TYPE_CHECKING: + from xml.etree.ElementTree import Element + + from app.library.Download import Download + +LOG: logging.Logger = logging.getLogger(__name__) + + +class TwitchHandler(BaseHandler): + FEED = "https://twitchrss.appspot.com/vodonly/{handle}" + + RX: re.Pattern[str] = re.compile(r"^https?:\/\/(?:www\.|m\.)?twitch\.tv\/(?P[a-z0-9_]{3,25})(?:\/.*)?$") + + @staticmethod + def can_handle(task: Task) -> bool: + LOG.debug(f"Checking if task '{task.name}' is using parsable Twitch URL: {task.url}") + return TwitchHandler.parse(task.url) is not None + + @staticmethod + async def handle(task: Task, notify: EventBus, queue: DownloadQueue): + """ + Fetch the RSS feed for a Twitch channel VODs, parse entries, + and enqueue new items that are not in the archive/queue already. + + Args: + task (Task): The task containing the Twitch channel URL. + notify (EventBus): The event bus for notifications. + queue (DownloadQueue): The download queue instance. + + """ + from defusedxml.ElementTree import fromstring + + handleName: str | None = TwitchHandler.parse(task.url) + if not handleName: + LOG.error(f"Cannot parse '{task.name}' URL: {task.url}") + return + + params: dict = task.get_ytdlp_opts().get_all() + archive_file: str | None = params.get("download_archive") + if not archive_file: + LOG.error(f"Task '{task.name}' does not have an archive file.") + return + + feed_url: str = TwitchHandler.FEED.format(handle=handleName) + + LOG.debug(f"Fetching '{task.name}' feed.") + response = await TwitchHandler.request(url=feed_url, ytdlp_opts=params) + response.raise_for_status() + + items: list = [] + has_items = False + + root: Element[str] = fromstring(response.text) + for entry in root.findall("channel/item"): + link_elem: Element[str] | None = entry.find("link") + url: str = link_elem.text.strip() if link_elem is not None and link_elem.text else "" + if not url: + LOG.warning(f"Entry in '{task.name}' feed is missing URL. Skipping entry.") + continue + + m: re.Match[str] | None = re.search(r"^https?://(?:www\.)?twitch\.tv/videos/(?P\d+)(?:[/?].*)?$", url) + if not m: + LOG.warning(f"URL in '{task.name}' feed does not look like a VOD link: {url}") + continue + + vid: str = m.group("id") + + title_elem: Element[str] | None = entry.find("title") + title: str = title_elem.text.strip() if title_elem is not None and title_elem.text else "" + + has_items = True + + id_dict = get_archive_id(url) + archive_id: str | None = id_dict.get("archive_id") + if not archive_id: + LOG.warning(f"Could not compute archive ID for video '{vid}' in '{task.name}' feed. Skipping entry.") + continue + + if archive_id in TwitchHandler.queued: + continue + + items.append({"id": vid, "url": url, "title": title, "archive_id": archive_id}) + + if len(items) < 1: + if not has_items: + LOG.warning(f"No entries found in '{task.name}' feed. URL: {feed_url}") + else: + LOG.debug(f"No new items found in '{task.name}' feed.") + return + + filtered: list = [] + + downloaded: list[str] = archive_read(archive_file, [item["archive_id"] for item in items]) + + for item in items: + TwitchHandler.queued.add(item["archive_id"]) + + if item["archive_id"] in downloaded: + continue + + if queue.queue.exists(url=item["url"]): + continue + + try: + done: Download = queue.done.get(url=item["url"]) + if "error" != done.info.status: + continue + except KeyError: + pass + + if item["archive_id"] not in TwitchHandler.failure_count: + TwitchHandler.failure_count[item["archive_id"]] = 0 + + filtered.append(item) + + if len(filtered) < 1: + LOG.debug(f"No new items found in '{task.name}' feed.") + return + + LOG.info(f"Found '{len(filtered)}' new items from '{task.name}' feed.") + + rItem: Item = Item.format( + { + "url": feed_url, + "preset": task.preset, + "folder": task.folder if task.folder else "", + "template": task.template if task.template else "", + "cli": task.cli if task.cli else "", + "auto_start": task.auto_start, + "extras": {"source_task": task.id}, + } + ) + + try: + await asyncio.gather( + *[notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) for item in filtered] + ) + except Exception as e: + LOG.exception(e) + LOG.error(f"Error while adding items from '{task.name}'. {e!s}") + return + + @staticmethod + def parse(url: str) -> str | None: + """ + Parse twitch URL to extract the channel. + + Args: + url (str): The url to check. + + Returns: + str | None: The parsed ID if successful, None otherwise. + + """ + match: re.Match[str] | None = TwitchHandler.RX.match(url) + return match.group("id") if match else None + + @staticmethod + def tests() -> list[tuple[str, bool]]: + """ + Test cases for the URL parser. + + Returns: + list[tuple[str, bool]]: A list of tuples containing the URL and expected result. + + """ + return [ + ("https://www.twitch.tv/test_username", True), + ("https://twitch.tv/test_username", True), + ("http://m.twitch.tv/test_username,", True), + ("https://www.twitch.tv/test_username/", True), + ("https://twitch.tv/test_username/", True), + ("http://m.twitch.tv/test_username/,", True), + ("twitch.tv/test_username/", False), + ] diff --git a/app/library/task_handlers/youtube.py b/app/library/task_handlers/youtube.py index eae2b340..834bf83d 100644 --- a/app/library/task_handlers/youtube.py +++ b/app/library/task_handlers/youtube.py @@ -1,15 +1,16 @@ import asyncio import logging import re -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING from xml.etree.ElementTree import Element -from app.library.config import Config from app.library.DownloadQueue import DownloadQueue from app.library.Events import EventBus, Events -from app.library.ItemDTO import Item, ItemDTO +from app.library.ItemDTO import Item from app.library.Tasks import Task -from app.library.Utils import archive_read +from app.library.Utils import archive_read, get_archive_id + +from ._base_handler import BaseHandler if TYPE_CHECKING: from xml.etree.ElementTree import Element @@ -18,22 +19,17 @@ if TYPE_CHECKING: LOG: logging.Logger = logging.getLogger(__name__) -EventBus.get_instance().subscribe( - Events.ITEM_ERROR, - lambda data, _, **__: YoutubeHandler.on_error(data.data), - f"{__name__}.on_error", -) - - -class YoutubeHandler: - queued: set[str] = set() - failure_count: dict[str, int] = {} +class YoutubeHandler(BaseHandler): FEED = "https://www.youtube.com/feeds/videos.xml?{type}={id}" - CHANNEL_REGEX = re.compile(r"^https?://(?:www\.)?youtube\.com/(?:channel/(?PUC[0-9A-Za-z_-]{22})|)/?$") + CHANNEL_REGEX: re.Pattern[str] = 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_-]+)|).*$") + PLAYLIST_REGEX: re.Pattern[str] = re.compile( + r"^https?://(?:www\.)?youtube\.com/(?:playlist\?list=(?P[A-Za-z0-9_-]+)|).*$" + ) @staticmethod def can_handle(task: Task) -> bool: @@ -45,7 +41,7 @@ class YoutubeHandler: return YoutubeHandler.parse(task.url) is not None @staticmethod - async def handle(task: Task, notify: EventBus, config: Config, queue: DownloadQueue): + async def handle(task: Task, notify: EventBus, queue: DownloadQueue): """ Fetch the Atom feed for a YouTube channel or playlist, parse entries, and return a list of videos with metadata. @@ -53,16 +49,9 @@ class YoutubeHandler: Args: task (Task): The task containing the YouTube URL. notify (EventBus): The event bus for notifications. - config (Config): The configuration instance. queue (DownloadQueue): The download queue instance. """ - params: dict = task.get_ytdlp_opts().get_all() - if not (archive_file := params.get("download_archive")): - LOG.error(f"Task '{task.name}' does not have an archive file.") - return - - import httpx from defusedxml.ElementTree import fromstring parsed: dict[str, str] | None = YoutubeHandler.parse(task.url) @@ -70,65 +59,49 @@ class YoutubeHandler: LOG.error(f"Cannot parse '{task.name}' URL: {task.url}") return + params: dict = task.get_ytdlp_opts().get_all() + feed_url: str = YoutubeHandler.FEED.format(type=parsed["type"], id=parsed["id"]) - LOG.debug(f"Fetching '{task.name}' feed.") - opts: dict[str, Any] = { - "proxy": params.get("proxy"), - "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", - ) - }, - } - - try: - from httpx_curl_cffi import AsyncCurlTransport, CurlOpt - - opts["transport"] = AsyncCurlTransport( - impersonate="chrome", - default_headers=True, - curl_options={CurlOpt.FRESH_CONNECT: True}, - ) - opts.pop("headers", None) - except Exception: - pass items: list = [] has_items = False - async with httpx.AsyncClient(**opts) as client: - response: httpx.Response = await client.request(method="GET", url=feed_url, timeout=120) - response.raise_for_status() + response = await YoutubeHandler.request(url=feed_url, ytdlp_opts=params) + response.raise_for_status() - root: Element[str] = fromstring(response.text) - ns: dict[str, str] = { - "atom": "http://www.w3.org/2005/Atom", - "yt": "http://www.youtube.com/xml/schemas/2015", - } + root: Element[str] = fromstring(response.text) + ns: dict[str, str] = { + "atom": "http://www.w3.org/2005/Atom", + "yt": "http://www.youtube.com/xml/schemas/2015", + } - for entry in root.findall("atom:entry", ns): - vid_elem: Element[str] | None = entry.find("yt:videoId", ns) - vid: str | None = vid_elem.text if vid_elem is not None else "" - if not vid: - LOG.warning(f"Entry in '{task.name}' feed is missing a video ID. Skipping entry.") - continue + for entry in root.findall("atom:entry", ns): + vid_elem: Element[str] | None = entry.find("yt:videoId", ns) + vid: str | None = vid_elem.text if vid_elem is not None else "" + if not vid: + LOG.warning(f"Entry in '{task.name}' feed is missing a video ID. Skipping entry.") + continue - archive_id: str = f"youtube {vid}" - url: str = f"https://www.youtube.com/watch?v={vid}" + url: str = f"https://www.youtube.com/watch?v={vid}" - title_elem: Element[str] | None = entry.find("atom:title", ns) - title: str | None = title_elem.text if title_elem is not None else "" + id_dict: dict[str, str | None] = get_archive_id(url) + archive_id: str | None = id_dict.get("archive_id") + if not archive_id: + LOG.warning(f"Could not compute archive ID for video '{vid}' in '{task.name}' feed. Skipping entry.") + continue - pub_elem: Element[str] | None = entry.find("atom:published", ns) - published: str | None = pub_elem.text if pub_elem is not None else "" - has_items = True + title_elem: Element[str] | None = entry.find("atom:title", ns) + title: str | None = title_elem.text if title_elem is not None else "" - if archive_id in YoutubeHandler.queued: - continue + pub_elem: Element[str] | None = entry.find("atom:published", ns) + published: str | None = pub_elem.text if pub_elem is not None else "" + has_items = True - items.append({"id": vid, "url": url, "title": title, "published": published, "archive_id": archive_id}) + if archive_id in YoutubeHandler.queued: + continue + + items.append({"id": vid, "url": url, "title": title, "published": published, "archive_id": archive_id}) if len(items) < 1: if not has_items: @@ -139,7 +112,7 @@ class YoutubeHandler: filtered: list = [] - downloaded: list[str] = archive_read(archive_file, [item["archive_id"] for item in items]) + downloaded: list[str] = archive_read(params.get("download_archive"), [item["archive_id"] for item in items]) for item in items: YoutubeHandler.queued.add(item["archive_id"]) @@ -170,7 +143,7 @@ class YoutubeHandler: rItem: Item = Item.format( { "url": feed_url, - "preset": str(task.preset or config.default_preset), + "preset": task.preset, "folder": task.folder if task.folder else "", "template": task.template if task.template else "", "cli": task.cli if task.cli else "", @@ -210,31 +183,6 @@ class YoutubeHandler: return None - @staticmethod - async def on_error(item: ItemDTO) -> None: - """ - Handle errors by logging them and removing the queued ID if it exists. - - Args: - item (ItemDTO): The error data containing the URL and other information. - - """ - cls = YoutubeHandler - if not item or not isinstance(item, ItemDTO): - return - - if not item.archive_id or not cls.failure_count.get(item.archive_id, None): - LOG.debug(f"Item '{item.name()}' not queued by the handler.") - return - - failCount: int = int(cls.failure_count.get(item.archive_id, 0)) - - LOG.info(f"Removing '{item.name()}' from queued IDs due to error. Failure count: '{failCount + 1}'.") - if item.archive_id in cls.queued: - cls.queued.remove(item.archive_id) - - cls.failure_count[item.archive_id] = 1 + failCount - @staticmethod def tests() -> list[tuple[str, bool]]: """