Added generic rss feed parser for tasks.

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

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 = """<?xml version="1.0" encoding="UTF-8"?>
+<feed xmlns="http://www.w3.org/2005/Atom">
+  <title>Example Feed</title>
+  <entry>
+    <title>Video 1</title>
+    <link href="https://www.youtube.com/watch?v=abc123" rel="alternate" />
+    <published>2024-01-01T00:00:00Z</published>
+  </entry>
+  <entry>
+    <title>Video 2</title>
+    <link href="https://www.youtube.com/watch?v=def456" rel="alternate" />
+    <published>2024-01-02T00:00:00Z</published>
+  </entry>
+</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 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 = """<?xml version="1.0" encoding="UTF-8"?>
+<rss version="2.0">
+  <channel>
+    <title>Example Channel</title>
+    <item>
+      <title>Video 1</title>
+      <link>https://www.youtube.com/watch?v=abc123</link>
+      <pubDate>Mon, 01 Jan 2024 00:00:00 +0000</pubDate>
+    </item>
+    <item>
+      <title>Video 2</title>
+      <link>https://www.youtube.com/watch?v=def456</link>
+      <pubDate>Tue, 02 Jan 2024 00:00:00 +0000</pubDate>
+    </item>
+  </channel>
+</rss>
+        """.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 = """<?xml version="1.0" encoding="UTF-8"?>
+<feed xmlns="http://www.w3.org/2005/Atom">
+  <entry>
+    <link href="https://www.youtube.com/watch?v=abc123" />
+  </entry>
+</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 = """<?xml version="1.0" encoding="UTF-8"?>
+<rss version="2.0">
+  <channel>
+    <title>Empty Channel</title>
+  </channel>
+</rss>
+        """.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 = """<?xml version="1.0" encoding="UTF-8"?>
+<rss version="2.0">
+  <channel>
+    <item>
+      <title>No URL Item</title>
+    </item>
+    <item>
+      <title>Valid Item</title>
+      <link>https://www.youtube.com/watch?v=abc123</link>
+    </item>
+  </channel>
+</rss>
+        """.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"

ui/app/components/TaskForm.vue

@@ -7,6 +7,12 @@
               URL</b></NuxtLink> link.</span>
       </Message>
     </div>
+    <div class="column is-12" v-if="form?.url && is_generic_rss(form.url)">
+      <Message title="Information" class="is-info is-background-info-80 has-text-dark" icon="fas fa-info-circle">
+        <span>You are using a generic RSS/Atom feed URL. The task handler will automatically download new items found
+          in this feed.</span>
+      </Message>
+    </div>
     <div class="column is-12">
       <form autocomplete="off" id="taskForm" @submit.prevent="checkInfo()">
         <div class="card">
@@ -238,8 +244,8 @@
                     Mark all existing items as downloaded
                   </label>
                   <div class="control is-unselectable">
-                    <input id="archive_all_after_add" type="checkbox" v-model="archiveAllAfterAdd" :disabled="addInProgress"
-                      class="switch is-danger" />
+                    <input id="archive_all_after_add" type="checkbox" v-model="archiveAllAfterAdd"
+                      :disabled="addInProgress" class="switch is-danger" />
                     <label for="archive_all_after_add" class="is-unselectable">
                       {{ archiveAllAfterAdd ? 'Yes' : 'No' }}
                     </label>
@@ -299,21 +305,23 @@
       <Message title="Tips" class="is-info is-background-info-80 has-text-dark" icon="fas fa-info-circle">
         <span>
           <ul>
-            <li>To enable YouTube RSS feed monitoring, The task URL must include a <code>channel_id</code> or
-              <code>playlist_id</code>. Other link types won’t work.
+            <li><strong>YouTube RSS:</strong> Use <code>channel_id</code> or <code>playlist_id</code> URLs. Other link
+              types (custom names, handles, user profiles) are not supported.
             </li>
-            <li>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 <code>Enable Handler</code>
-              option. To have the task only monitor RSS feed, <b>do not set timer</b>.</li>
-            <li>RSS Feed monitoring will only work if you have <code>--download-archive</code> set in <code>Command options
-            for yt-dlp</code> via the task itself, or preset. command options
+            <li><strong>Generic RSS/Atom:</strong> URL must end with <code>.rss</code> or <code>.atom</code>. If not
+              possible, append <code>&handler=rss</code> to existing query parameters, or add <code>#handler=rss</code>
+              as a fragment.
+            </li>
+            <li><strong>RSS Monitoring Basics:</strong> Runs hourly independently. Timer controls scheduled downloads to
+              yt-dlp. Disable <code>Enable Handler</code> to disable RSS monitoring.
+            </li>
+            <li><strong>Archive Requirement:</strong> RSS monitoring requires <code>--download-archive</code> in
+              <code>Command options for yt-dlp</code> (set in task or preset).
             </li>
           </ul>
         </span>
       </Message>
-    </div>
-
-    <datalist id="folders" v-if="config?.folders">
+    </div> <datalist id="folders" v-if="config?.folders">
       <option v-for="dir in config.folders" :key="dir" :value="dir" />
     </datalist>
     <Modal v-if="showOptions" @close="showOptions = false" :contentClass="'modal-content-max'">
@@ -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<AutoCompleteOptions>([])
 const archiveAllAfterAdd = ref<boolean>(false)
 
 const CHANNEL_REGEX = /^https?:\/\/(?:www\.)?youtube\.com\/(?:(?:channel\/(?<channelId>UC[0-9A-Za-z_-]{22}))|(?:c\/(?<customName>[A-Za-z0-9_-]+))|(?:user\/(?<userName>[A-Za-z0-9_-]+))|(?:@(?<handle>[A-Za-z0-9_-]+)))(?<suffix>\/.*)?\/?$/
-
+const GENERIC_RSS_REGEX = /\.(rss|atom)(\?.*)?$|handler=rss/i
 const form = reactive<task_item>({ ...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<string> => {
   if (!url || '' === url) {
     return url

ui/app/pages/tasks.vue

@@ -389,14 +389,9 @@
         <Message title="Tips" class="is-info is-background-info-80" icon="fas fa-info-circle">
           <span>
             <ul>
-              <li>
-                If you don't wish to download <strong>ALL</strong> content from a channel or playlist, click on
-                <code> <span class="icon"><i class="fa-solid fa-cogs" /></span> Actions > <span class="icon"><i
-                  class="fa-solid fa-box-archive" /></span> Archive All</code> to archive all existing content.
+              <li><strong>Selective Downloads:</strong> To avoid downloading all existing content from a channel/playlist, use <code><span class="icon"><i class="fa-solid fa-cogs" /></span> Actions > <span class="icon"><i class="fa-solid fa-box-archive" /></span> Archive All</code> to mark existing items as already downloaded.
               </li>
-              <li>
-                For custom handler definitions, you shouldn't have a timer set, as yt-dlp wouldn't know what to do with
-                it.
+              <li><strong>Custom Handlers:</strong> Leave timer empty for custom handler definitions. The handler runs hourly and doesn't require a scheduled timer.
               </li>
             </ul>
           </span>