Add yt-dlp video download node

docs/index.md

@@ -11,6 +11,7 @@ permalink: /
 
 - **[lib.audio](nodetool_audio.md)** - Save audio files to the assets directory.
 - **[lib.html](nodetool_html.md)** - HTML utility nodes.
+- **[lib.yt_dlp](nodetool_yt_dlp.md)** - Download videos with yt-dlp.
 - **[nodetool.boolean](nodetool_boolean.md)** - Logical operators, comparisons and flow control helpers.
 - **[nodetool.base64](nodetool_base64.md)** - Base64 encoding and decoding utilities.
 - **[nodetool.code](nodetool_code.md)** - Evaluate expressions or run small Python snippets (development use).

docs/nodetool_yt_dlp.md

@@ -0,0 +1,20 @@
+---
+layout: default
+title: lib.yt_dlp
+parent: Nodes
+has_children: false
+nav_order: 2
+---
+
+# nodetool.nodes.lib.yt_dlp
+
+## VideoDownload
+
+Download a video from a URL using [yt-dlp](https://github.com/yt-dlp/yt-dlp).
+
+Use cases:
+- Save videos for offline processing
+- Fetch clips for video editing workflows
+- Archive online content
+
+**Tags:** youtube, download, video, ytdlp

src/nodetool/dsl/lib/yt_dlp.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel, Field
+import typing
+from typing import Any
+import nodetool.metadata.types
+import nodetool.metadata.types as types
+from nodetool.dsl.graph import GraphNode
+
+
+class VideoDownload(GraphNode):
+    """Download a video from a URL using ``yt_dlp``.
+    youtube, download, video, ytdlp
+
+    Use cases:
+    - Save videos for offline processing
+    - Fetch clips for video editing workflows
+    - Archive online video content
+    """
+
+    url: str | GraphNode | tuple[GraphNode, str] = Field(
+        default="", description="URL of the video to download"
+    )
+    format: str | GraphNode | tuple[GraphNode, str] = Field(
+        default="best", description="yt-dlp format string"
+    )
+
+    @classmethod
+    def get_node_type(cls):
+        return "lib.yt_dlp.VideoDownload"

src/nodetool/nodes/lib/yt_dlp.py

@@ -0,0 +1,45 @@
+import asyncio
+import os
+import tempfile
+from pydantic import Field
+from nodetool.workflows.base_node import BaseNode
+from nodetool.workflows.processing_context import ProcessingContext
+from nodetool.metadata.types import VideoRef
+
+
+class VideoDownload(BaseNode):
+    """Download a video from a URL using ``yt_dlp``.
+    youtube, download, video, ytdlp
+
+    Use cases:
+    - Save videos for offline processing
+    - Fetch clips for video editing workflows
+    - Archive online video content
+    """
+
+    url: str = Field(default="", description="URL of the video to download")
+    format: str = Field(default="best", description="yt-dlp format string")
+
+    @classmethod
+    def get_title(cls):
+        return "Download Video"
+
+    async def process(self, context: ProcessingContext) -> VideoRef:
+        if not self.url:
+            raise ValueError("url cannot be empty")
+
+        import yt_dlp
+
+        def _download() -> bytes:
+            with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp:
+                filename = tmp.name
+            ydl_opts = {"format": self.format, "outtmpl": filename, "quiet": True}
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                ydl.download([self.url])
+            with open(filename, "rb") as f:
+                data = f.read()
+            os.remove(filename)
+            return data
+
+        video_bytes = await asyncio.to_thread(_download)
+        return await context.video_from_bytes(video_bytes)

tests/nodetool/test_ytdlp.py

@@ -0,0 +1,39 @@
+import types
+import os
+import sys
+import pytest
+from nodetool.workflows.processing_context import ProcessingContext
+from nodetool.nodes.lib.yt_dlp import VideoDownload
+from nodetool.metadata.types import VideoRef
+
+TEST_MP4 = os.path.join(os.path.dirname(__file__), "../test.mp4")
+
+
+class DummyYDL:
+    def __init__(self, opts):
+        self.out = opts.get("outtmpl")
+
+    def download(self, urls):
+        with open(TEST_MP4, "rb") as src, open(self.out, "wb") as dst:
+            dst.write(src.read())
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        pass
+
+
+@pytest.fixture
+def context():
+    return ProcessingContext(user_id="test", auth_token="test")
+
+
+@pytest.mark.asyncio
+async def test_video_download(context, monkeypatch):
+    monkeypatch.setitem(
+        sys.modules, "yt_dlp", types.SimpleNamespace(YoutubeDL=DummyYDL)
+    )
+    node = VideoDownload(url="http://example.com/video")
+    result = await node.process(context)
+    assert isinstance(result, VideoRef)