From 9d36ab13514e28b3e766b1d01d5db595653053a9 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Wed, 22 Oct 2025 18:52:17 +0300 Subject: [PATCH] Added generic rss feed parser for tasks. --- app/library/task_handlers/rss.py | 260 ++++++++++++++++++++++++++++++ app/tests/test_rss_handler.py | 261 +++++++++++++++++++++++++++++++ ui/app/components/TaskForm.vue | 43 +++-- ui/app/pages/tasks.vue | 9 +- 4 files changed, 552 insertions(+), 21 deletions(-) create mode 100644 app/library/task_handlers/rss.py create mode 100644 app/tests/test_rss_handler.py diff --git a/app/library/task_handlers/rss.py b/app/library/task_handlers/rss.py new file mode 100644 index 00000000..40bb62bc --- /dev/null +++ b/app/library/task_handlers/rss.py @@ -0,0 +1,260 @@ +import hashlib +import logging +import re +from typing import TYPE_CHECKING, Any +from xml.etree.ElementTree import Element + +from app.library.cache import Cache +from app.library.Tasks import Task, TaskFailure, TaskItem, TaskResult +from app.library.Utils import extract_info, get_archive_id + +from ._base_handler import BaseHandler + +if TYPE_CHECKING: + from xml.etree.ElementTree import Element + +LOG: logging.Logger = logging.getLogger(__name__) +CACHE: Cache = Cache() + + +class RssGenericHandler(BaseHandler): + FEED_PATTERN: re.Pattern[str] = re.compile( + r"\.(rss|atom)(\?.*)?$|handler=rss", + re.IGNORECASE, + ) + + @staticmethod + def can_handle(task: Task) -> bool: + LOG.debug(f"'{task.name}': Checking if task URL is parsable RSS feed: {task.url}") + return RssGenericHandler.parse(task.url) is not None + + @staticmethod + async def _get( + task: Task, + params: dict, + parsed: dict[str, str], + ) -> tuple[str, list[dict[str, str]], int]: + """ + Fetch the feed and return raw entries. + + Args: + task (Task): The task containing the feed URL. + params (dict): The ytdlp options. + parsed (dict): The parsed URL components (contains 'url' key). + + Returns: + tuple[str, list[dict[str, str]], int]: The feed URL, list of entry dictionaries, and entry count. + + """ + from defusedxml.ElementTree import fromstring + + feed_url: str = parsed["url"] + LOG.debug(f"'{task.name}': Fetching RSS/Atom feed from {feed_url}") + + response = await RssGenericHandler.request(url=feed_url, ytdlp_opts=params) + response.raise_for_status() + + root: Element = fromstring(response.text) + + # Define namespaces for different feed formats + ns: dict[str, str] = { + "atom": "http://www.w3.org/2005/Atom", + "rss": "http://www.rssboard.org/specification", + "content": "http://purl.org/rss/1.0/modules/content/", + "media": "http://search.yahoo.com/mrss/", + } + + items: list[dict[str, str]] = [] + real_count = 0 + + # Try to parse as Atom feed first + entries = root.findall("atom:entry", ns) + if entries: + LOG.debug(f"'{task.name}': Detected Atom feed format with {len(entries)} entries") + for entry in entries: + link_elem: Element | None = entry.find("atom:link[@rel='alternate']", ns) + if link_elem is None: + link_elem = entry.find("atom:link", ns) + + url: str = "" + if link_elem is not None and link_elem.get("href"): + url = link_elem.get("href", "") + + if not url: + LOG.warning(f"'{task.name}': Atom entry missing URL. Skipping.") + continue + + title_elem: Element | None = entry.find("atom:title", ns) + title: str = title_elem.text if title_elem is not None and title_elem.text else "" + + pub_elem: Element | None = entry.find("atom:published", ns) + published: str = pub_elem.text if pub_elem is not None and pub_elem.text else "" + + real_count += 1 + items.append({"url": url, "title": title, "published": published}) + else: + # Try to parse as RSS feed + rss_items = root.findall(".//item") + LOG.debug(f"'{task.name}': Detected RSS feed format with {len(rss_items)} items") + + for item in rss_items: + # Try different link element names (link, url, media:content) + url: str = "" + + link_elem = item.find("link") + if link_elem is not None and link_elem.text: + url = link_elem.text + else: + # Try media:content + media_elem = item.find("media:content", ns) + if media_elem is not None and media_elem.get("url"): + url = media_elem.get("url", "") + else: + # Try enclosure + enclosure_elem = item.find("enclosure") + if enclosure_elem is not None and enclosure_elem.get("url"): + url = enclosure_elem.get("url", "") + + if not url: + LOG.warning(f"'{task.name}': RSS item missing URL. Skipping.") + continue + + title_elem = item.find("title") + title: str = title_elem.text if title_elem is not None and title_elem.text else "" + + pub_elem = item.find("pubDate") + published: str = pub_elem.text if pub_elem is not None and pub_elem.text else "" + + real_count += 1 + items.append({"url": url, "title": title, "published": published}) + + return feed_url, items, real_count + + @staticmethod + async def extract(task: Task) -> TaskResult | TaskFailure: + """ + Extract items from an RSS/Atom feed. + + Args: + task (Task): The task containing the feed URL. + + Returns: + TaskResult | TaskFailure: Extraction result with parsed items or failure information. + + """ + parsed: dict[str, str] | None = RssGenericHandler.parse(task.url) + if not parsed: + return TaskFailure(message="Unrecognized RSS/Atom feed URL.") + + params: dict = task.get_ytdlp_opts().get_all() + + try: + feed_url, items, real_count = await RssGenericHandler._get(task, params, parsed) + except Exception as exc: + LOG.exception(exc) + return TaskFailure(message="Failed to fetch RSS/Atom feed.", error=str(exc)) + + task_items: list[TaskItem] = [] + + for entry in items: + if not (url := entry.get("url")): + continue + + # Try to get static archive ID first + id_dict: dict[str, str | None] = get_archive_id(url=url) + archive_id: str | None = id_dict.get("archive_id") + + # If static archive_id fails, try to fetch it via yt-dlp (like generic.py) + if not archive_id: + cache_key: str = hashlib.sha256(f"{task.name}-{url}".encode()).hexdigest() + + if CACHE.has(cache_key): + archive_id = CACHE.get(cache_key) + if not archive_id: + LOG.debug(f"'{task.name}': Cached failure for URL '{url}'. Skipping.") + continue + else: + LOG.warning( + f"'{task.name}': Unable to generate static archive ID for '{url}' in feed. " + "Doing real request to fetch yt-dlp archive ID." + ) + + info = extract_info( + config=params, + url=url, + no_archive=True, + no_log=True, + ) + + if not info: + LOG.error( + f"'{task.name}': Failed to extract info for URL '{url}' to generate archive ID. Skipping." + ) + CACHE.set(cache_key, None) + continue + + if not info.get("id") or not info.get("extractor_key"): + LOG.error( + f"'{task.name}': Incomplete info extracted for URL '{url}' to generate archive ID. Skipping." + ) + CACHE.set(cache_key, None) + continue + + archive_id = f"{str(info.get('extractor_key', '')).lower()} {info.get('id')}" + CACHE.set(cache_key, archive_id) + + metadata: dict[str, Any] = {k: v for k, v in entry.items() if k not in {"url", "title", "published"}} + + task_items.append( + TaskItem( + url=url, + title=entry.get("title"), + archive_id=archive_id, + metadata={"published": entry.get("published"), **metadata}, + ) + ) + + return TaskResult( + items=task_items, + metadata={"feed_url": feed_url, "entry_count": real_count}, + ) + + @staticmethod + def parse(url: str) -> dict[str, str] | None: + """ + Parse URL for valid RSS/Atom feed. + + Args: + url (str): The URL to parse. + + Returns: + dict[str, str] | None: A dictionary with 'url' key if valid RSS/Atom feed, None otherwise. + + """ + if not isinstance(url, str) or not url: + return None + + return {"url": url} if RssGenericHandler.FEED_PATTERN.search(url) 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.example.com/test.rss", True), + ("https://www.example.com/test.atom", True), + ("https://www.example.com/test.atom#handler=rss", True), + ("https://www.example.com/test.atom?handler=rss", True), + ("https://www.example.com/feed.rss?version=2.0", True), + ("https://www.example.com/test.xml", False), + ("https://www.example.com/channel/UC_x5XG1OV2P6uZZ5FSM9Ttw", False), + ("https://www.example.com/playlist?list=PLBCF2DAC6FFB574DE", False), + ("https://www.example.com/user/SomeUser", False), + ("https://example.com/feed.ATOM", True), + ("https://example.com/feed.RSS", True), + ] diff --git a/app/tests/test_rss_handler.py b/app/tests/test_rss_handler.py new file mode 100644 index 00000000..b8a5cb0a --- /dev/null +++ b/app/tests/test_rss_handler.py @@ -0,0 +1,261 @@ +import pytest + +from app.library.task_handlers.rss import RssGenericHandler +from app.library.Tasks import Task, TaskResult + + +class DummyResponse: + def __init__(self, text: str): + self.text = text + + def raise_for_status(self) -> None: + return None + + +class DummyOpts: + def __init__(self, data): + self._data = data + + def get_all(self): + return self._data + + +class TestRssHandlerParsing: + """Test URL parsing for RSS/Atom feeds using the tests() method.""" + + @pytest.mark.parametrize(("url", "expected"), RssGenericHandler.tests()) + def test_url_parsing(self, url: str, expected: bool): + """Test URL parsing against all defined test cases.""" + result = RssGenericHandler.parse(url) + is_matched = result is not None + assert is_matched == expected, f"URL '{url}' expected {expected}, got {is_matched}" + + @pytest.mark.parametrize(("url", "expected"), RssGenericHandler.tests()) + def test_returns_url_dict_on_match(self, url: str, expected: bool): + """Test that parse returns dict with 'url' key for valid feeds.""" + result = RssGenericHandler.parse(url) + if expected: + assert result is not None + assert "url" in result + assert result["url"] == url + else: + assert result is None + + +class TestRssHandlerExtraction: + """Test RSS feed extraction and parsing.""" + + @pytest.mark.asyncio + async def test_rss_atom_feed_extraction(self, monkeypatch): + """Test extraction from Atom feed.""" + atom_feed = """ + + Example Feed + + Video 1 + + 2024-01-01T00:00:00Z + + + Video 2 + + 2024-01-02T00:00:00Z + + + """.strip() + + async def fake_request(**kwargs): # noqa: ARG001 + return DummyResponse(atom_feed) + + monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request)) + monkeypatch.setattr(Task, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005 + + task = Task( + id="test_rss", + name="Test Atom Feed", + url="https://example.com/feed.atom", + preset="default", + ) + + result = await RssGenericHandler.extract(task) + + assert isinstance(result, TaskResult) + assert len(result.items) == 2 + assert result.items[0].title == "Video 1" + assert result.items[0].url == "https://www.youtube.com/watch?v=abc123" + assert result.items[1].title == "Video 2" + assert result.items[1].url == "https://www.youtube.com/watch?v=def456" + assert result.metadata["entry_count"] == 2 + + @pytest.mark.asyncio + async def test_rss_feed_extraction(self, monkeypatch): + """Test extraction from RSS feed.""" + rss_feed = """ + + + Example Channel + + Video 1 + https://www.youtube.com/watch?v=abc123 + Mon, 01 Jan 2024 00:00:00 +0000 + + + Video 2 + https://www.youtube.com/watch?v=def456 + Tue, 02 Jan 2024 00:00:00 +0000 + + + + """.strip() + + async def fake_request(**kwargs): # noqa: ARG001 + return DummyResponse(rss_feed) + + monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request)) + monkeypatch.setattr(Task, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005 + + task = Task( + id="test_rss", + name="Test RSS Feed", + url="https://example.com/feed.rss", + preset="default", + ) + + result = await RssGenericHandler.extract(task) + + assert isinstance(result, TaskResult) + assert len(result.items) == 2 + assert result.items[0].title == "Video 1" + assert result.items[0].url == "https://www.youtube.com/watch?v=abc123" + assert result.items[1].title == "Video 2" + assert result.items[1].url == "https://www.youtube.com/watch?v=def456" + assert result.metadata["entry_count"] == 2 + + @pytest.mark.asyncio + async def test_can_handle(self, monkeypatch): + """Test can_handle method.""" + atom_feed = """ + + + + + + """.strip() + + async def fake_request(**kwargs): # noqa: ARG001 + return DummyResponse(atom_feed) + + monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request)) + monkeypatch.setattr(Task, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005 + + task = Task( + id="test_rss", + name="Test Feed", + url="https://example.com/feed.atom", + preset="default", + ) + + assert RssGenericHandler.can_handle(task) is True + + non_feed_task = Task( + id="test_youtube", + name="YouTube Video", + url="https://www.youtube.com/watch?v=abc123", + preset="default", + ) + + assert RssGenericHandler.can_handle(non_feed_task) is False + + +class TestRssHandlerEdgeCases: + """Test edge cases in RSS handling.""" + + @pytest.mark.asyncio + async def test_empty_feed(self, monkeypatch): + """Test handling of empty feed.""" + empty_feed = """ + + + Empty Channel + + + """.strip() + + async def fake_request(**kwargs): # noqa: ARG001 + return DummyResponse(empty_feed) + + monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request)) + monkeypatch.setattr(Task, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005 + + task = Task( + id="test_empty", + name="Empty Feed", + url="https://example.com/feed.rss", + preset="default", + ) + + result = await RssGenericHandler.extract(task) + + assert isinstance(result, TaskResult) + assert len(result.items) == 0 + assert result.metadata["entry_count"] == 0 + + @pytest.mark.asyncio + async def test_invalid_feed_url(self, monkeypatch): + """Test handling of invalid feed URL.""" + from app.library.Tasks import TaskFailure + + async def fake_request(**kwargs): # noqa: ARG001 + msg = "Network error" + raise Exception(msg) + + monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request)) + monkeypatch.setattr(Task, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005 + + task = Task( + id="test_invalid", + name="Invalid Feed", + url="https://example.com/feed.rss", + preset="default", + ) + + result = await RssGenericHandler.extract(task) + + assert isinstance(result, TaskFailure) + + @pytest.mark.asyncio + async def test_missing_urls_in_feed(self, monkeypatch): + """Test handling of entries missing URLs.""" + feed = """ + + + + No URL Item + + + Valid Item + https://www.youtube.com/watch?v=abc123 + + + + """.strip() + + async def fake_request(**kwargs): # noqa: ARG001 + return DummyResponse(feed) + + monkeypatch.setattr(RssGenericHandler, "request", staticmethod(fake_request)) + monkeypatch.setattr(Task, "get_ytdlp_opts", lambda self: DummyOpts({"download_archive": "/tmp/archive"})) # noqa: ARG005 + + task = Task( + id="test_missing", + name="Feed with Missing URLs", + url="https://example.com/feed.rss", + preset="default", + ) + + result = await RssGenericHandler.extract(task) + + # Should only include the item with URL + assert isinstance(result, TaskResult) + assert len(result.items) == 1 + assert result.items[0].title == "Valid Item" diff --git a/ui/app/components/TaskForm.vue b/ui/app/components/TaskForm.vue index e05ac89a..d0089766 100644 --- a/ui/app/components/TaskForm.vue +++ b/ui/app/components/TaskForm.vue @@ -7,6 +7,12 @@ URL link. +
+ + You are using a generic RSS/Atom feed URL. The task handler will automatically download new items found + in this feed. + +
@@ -238,8 +244,8 @@ Mark all existing items as downloaded
- + @@ -299,21 +305,23 @@
    -
  • To enable YouTube RSS feed monitoring, The task URL must include a channel_id or - playlist_id. Other link types won’t work. +
  • YouTube RSS: Use channel_id or playlist_id URLs. Other link + types (custom names, handles, user profiles) are not supported.
    • -
  • RSS monitoring runs every hour alongside the actual task execution. Regardless if the task has timer set - or not. To opt out of RSS monitoring for a specific task, simply disable the Enable Handler - option. To have the task only monitor RSS feed, do not set timer.
    • -
  • RSS Feed monitoring will only work if you have --download-archive set in Command options - for yt-dlp via the task itself, or preset. command options +
  • Generic RSS/Atom: URL must end with .rss or .atom. If not + possible, append &handler=rss to existing query parameters, or add #handler=rss + as a fragment. +
    • +
  • RSS Monitoring Basics: Runs hourly independently. Timer controls scheduled downloads to + yt-dlp. Disable Enable Handler to disable RSS monitoring. +
    • +
  • Archive Requirement: RSS monitoring requires --download-archive in + Command options for yt-dlp (set in task or preset).
-
- - +
@@ -329,7 +337,7 @@ import { CronExpressionParser } from 'cron-parser' import TextareaAutocomplete from '~/components/TextareaAutocomplete.vue' import type { AutoCompleteOptions } from '~/types/autocomplete' import type { exported_task, task_item } from '~/types/tasks' -import {useConfirm} from '~/composables/useConfirm' +import { useConfirm } from '~/composables/useConfirm' const props = defineProps<{ reference?: string | null | undefined @@ -354,7 +362,7 @@ const ytDlpOpt = ref([]) const archiveAllAfterAdd = ref(false) const CHANNEL_REGEX = /^https?:\/\/(?:www\.)?youtube\.com\/(?:(?:channel\/(?UC[0-9A-Za-z_-]{22}))|(?:c\/(?[A-Za-z0-9_-]+))|(?:user\/(?[A-Za-z0-9_-]+))|(?:@(?[A-Za-z0-9_-]+)))(?\/.*)?\/?$/ - +const GENERIC_RSS_REGEX = /\.(rss|atom)(\?.*)?$|handler=rss/i const form = reactive({ ...props.task }) watch(() => config.ytdlp_options, newOptions => ytDlpOpt.value = newOptions @@ -497,6 +505,13 @@ const is_yt_handle = (url: string): boolean => { return false } +const is_generic_rss = (url: string): boolean => { + if (!url || '' === url) { + return false + } + return GENERIC_RSS_REGEX.test(url) +} + const convert_url = async (url: string): Promise => { if (!url || '' === url) { return url diff --git a/ui/app/pages/tasks.vue b/ui/app/pages/tasks.vue index 35ccad94..ea6c815e 100644 --- a/ui/app/pages/tasks.vue +++ b/ui/app/pages/tasks.vue @@ -389,14 +389,9 @@
    -
  • - If you don't wish to download ALL content from a channel or playlist, click on - Actions > Archive All to archive all existing content. +
  • Selective Downloads: To avoid downloading all existing content from a channel/playlist, use Actions > Archive All to mark existing items as already downloaded.
    • -
  • - For custom handler definitions, you shouldn't have a timer set, as yt-dlp wouldn't know what to do with - it. +
  • Custom Handlers: Leave timer empty for custom handler definitions. The handler runs hourly and doesn't require a scheduled timer.