FEAT: new TwitchHandler

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__:

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))

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<id>[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<id>\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),
+        ]

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/(?P<id>UC[0-9A-Za-z_-]{22})|)/?$")
+    CHANNEL_REGEX: re.Pattern[str] = re.compile(
+        r"^https?://(?:www\.)?youtube\.com/(?:channel/(?P<id>UC[0-9A-Za-z_-]{22})|)/?$"
+    )
 
-    PLAYLIST_REGEX = re.compile(r"^https?://(?:www\.)?youtube\.com/(?:playlist\?list=(?P<id>[A-Za-z0-9_-]+)|).*$")
+    PLAYLIST_REGEX: re.Pattern[str] = re.compile(
+        r"^https?://(?:www\.)?youtube\.com/(?:playlist\?list=(?P<id>[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 = await YoutubeHandler.request(url=feed_url, ytdlp_opts=params)
-            response: httpx.Response = await client.request(method="GET", url=feed_url, timeout=120)
+        response.raise_for_status()
-            response.raise_for_status()
 
-            root: Element[str] = fromstring(response.text)
+        root: Element[str] = fromstring(response.text)
-            ns: dict[str, str] = {
+        ns: dict[str, str] = {
-                "atom": "http://www.w3.org/2005/Atom",
+            "atom": "http://www.w3.org/2005/Atom",
-                "yt": "http://www.youtube.com/xml/schemas/2015",
+            "yt": "http://www.youtube.com/xml/schemas/2015",
-            }
+        }
 
-            for entry in root.findall("atom:entry", ns):
+        for entry in root.findall("atom:entry", ns):
-                vid_elem: Element[str] | None = entry.find("yt:videoId", ns)
+            vid_elem: Element[str] | None = entry.find("yt:videoId", ns)
-                vid: str | None = vid_elem.text if vid_elem is not None else ""
+            vid: str | None = vid_elem.text if vid_elem is not None else ""
-                if not vid:
+            if not vid:
-                    LOG.warning(f"Entry in '{task.name}' feed is missing a video ID. Skipping entry.")
+                LOG.warning(f"Entry in '{task.name}' feed is missing a video ID. Skipping entry.")
-                    continue
+                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)
+            id_dict: dict[str, str | None] = get_archive_id(url)
-                title: str | None = title_elem.text if title_elem is not None else ""
+            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)
+            title_elem: Element[str] | None = entry.find("atom:title", ns)
-                published: str | None = pub_elem.text if pub_elem is not None else ""
+            title: str | None = title_elem.text if title_elem is not None else ""
-                has_items = True
 
-                if archive_id in YoutubeHandler.queued:
+            pub_elem: Element[str] | None = entry.find("atom:published", ns)
-                    continue
+            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]]:
         """