RSS feed support for youtube channel/playlist.

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

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/(?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_-]+)|).*$")
+
+    @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': <channel-id>}
+            {'type': 'playlist', 'id': <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",
+        ]

ui/components/TaskForm.vue

@@ -1,5 +1,22 @@
 <template>
   <main class="columns mt-2 is-multiline">
+    <div class="column is-12" v-if="form?.url && is_yt_handle(form.url)">
+      <Message title="Warning" class="is-background-warning-80 has-text-dark" icon="fas fa-info-circle">
+        <span>
+          <ul>
+            <li>You are using a YouTube link with handle instead of channel_id. To activate RSS feed support for
+              channel, you need to use
+              the channel ID. For example, <code>https://www.youtube.com/channel/UCUi3_cffYenmMTuWEsLHzqg</code>
+            </li>
+            <li>
+              To get youtube channel_id simply visit the page, click on <b>more about this channel</b>, scroll down to
+              <b>share
+                channel</b>, click on <code>Copy channel id</code>.
+            </li>
+          </ul>
+        </span>
+      </Message>
+    </div>
     <div class="column is-12">
       <form autocomplete="off" id="taskForm" @submit.prevent="checkInfo()">
         <div class="card">
@@ -79,7 +96,10 @@
                   </div>
                   <span class="help">
                     <span class="icon"><i class="fa-solid fa-info" /></span>
-                    <span>The channel or playlist URL.</span>
+                    <span>The channel or playlist URL. For youtube there is rss feed support if you use URL with
+                      channel_id or playlist_id. For example, https://www.youtube.com/<span
+                        class="has-text-danger">channel/UCUi3_cffYenmMTuWEsLHzqg</span>
+                    </span>
                   </span>
                 </div>
               </div>
@@ -127,7 +147,7 @@
                   </label>
                   <div class="control">
                     <input type="text" class="input" id="timer" v-model="form.timer" :disabled="addInProgress"
-                    placeholder="0 12 * * 5">
+                      placeholder="0 12 * * 5">
                   </div>
                   <span class="help">
                     <span class="icon"><i class="fa-solid fa-info" /></span>
@@ -218,6 +238,28 @@
         </div>
       </form>
     </div>
+
+    <div class="column is-12">
+      <Message title="Tips" class="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>
+            <li>RSS monitoring runs every hour alongside the actual task execution. It checks each feed for new videos
+              and automatically queues any you haven’t downloaded yet.</li>
+            <li>To opt out of RSS monitoring for a specific task, append <code>[no_handler]</code> to that task’s name.
+            </li>
+            <li>RSS Feed monitoring will only work if you have <code>--download-archive</code> set in command options
+              for ytdlp.cli, preset or task.</li>
+            <li>If you don't have <code>--download-archive</code> set but <code>YTP_KEEP_ARCHIVE</code> environment
+              option is set to <code>true</code> which is the default, It will also work.
+            </li>
+          </ul>
+        </span>
+      </Message>
+    </div>
+
     <datalist id="folders" v-if="config?.folders">
       <option v-for="dir in config.folders" :key="dir" :value="dir" />
     </datalist>
@@ -228,13 +270,6 @@
 import { useStorage } from '@vueuse/core'
 import { CronExpressionParser } from 'cron-parser'
 
-const emitter = defineEmits(['cancel', 'submit'])
-const toast = useNotification()
-const config = useConfigStore()
-const showImport = useStorage('showImport', false)
-const import_string = ref('')
-const box = useConfirm()
-
 const props = defineProps({
   reference: {
     type: String,
@@ -252,6 +287,17 @@ const props = defineProps({
   },
 })
 
+const emitter = defineEmits(['cancel', 'submit'])
+
+const toast = useNotification()
+const config = useConfigStore()
+const box = useConfirm()
+const showImport = useStorage('showImport', false)
+
+const import_string = ref('')
+
+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_-]+)))\/?$/;
+
 const form = reactive(props.task)
 
 onMounted(() => {
@@ -420,4 +466,15 @@ const get_output_template = () => {
   }
   return config.app.output_template || '%(title)s.%(ext)s'
 }
+
+function is_yt_handle(url) {
+  let m = url.match(CHANNEL_REGEX);
+  if (m?.groups) {
+    if (m.groups?.channelId) {
+      return false
+    }
+    return true
+  }
+  return false
+}
 </script>