youtube: search_youtube via yt-dlp ytsearchN + parse JSON

core/youtube_search.py

@@ -0,0 +1,72 @@
+"""Ricerca su YouTube via yt-dlp (subprocess, flat playlist metadata)."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+
+from core.paths import find_ytdlp, subprocess_flags
+
+
+_TIMEOUT_SEC = 30
+
+
+def search_youtube(query: str, limit: int = 50) -> list:
+    """Cerca su YouTube (ytsearchN:query) e ritorna metadata "flat" (senza scaricare).
+
+    Args:
+        query: testo di ricerca libero
+        limit: numero massimo di risultati (default 50)
+
+    Returns:
+        list[dict] con {id, url, title, channel, duration_sec}. Vuota se nessun match.
+
+    Raises:
+        RuntimeError: se yt-dlp non disponibile, timeout, o exit non-zero.
+    """
+    ytdlp = find_ytdlp()
+    if not ytdlp:
+        raise RuntimeError("yt-dlp non trovato nel bundle / PATH")
+
+    cmd = [
+        ytdlp,
+        "--flat-playlist",
+        "--dump-single-json",
+        "--no-warnings",
+        f"ytsearch{limit}:{query}",
+    ]
+    try:
+        proc = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=_TIMEOUT_SEC,
+            **subprocess_flags(),
+        )
+    except subprocess.TimeoutExpired as e:
+        raise RuntimeError(f"Timeout ricerca YouTube ({_TIMEOUT_SEC}s)") from e
+
+    if proc.returncode != 0:
+        tail = (proc.stderr or "").strip().splitlines()[-3:]
+        raise RuntimeError(f"yt-dlp error: {' | '.join(tail) or 'unknown'}")
+
+    try:
+        data = json.loads(proc.stdout)
+    except json.JSONDecodeError as e:
+        raise RuntimeError(f"yt-dlp output non-JSON: {e}") from e
+
+    entries = data.get("entries") or []
+    result: list = []
+    for e in entries:
+        if not isinstance(e, dict):
+            continue
+        video_id = e.get("id") or ""
+        url = e.get("url") or (f"https://www.youtube.com/watch?v={video_id}" if video_id else "")
+        result.append({
+            "id": video_id,
+            "url": url,
+            "title": e.get("title") or "",
+            "channel": e.get("uploader") or e.get("channel") or "",
+            "duration_sec": int(e.get("duration") or 0),
+        })
+    return result

tests/test_youtube_search.py

@@ -0,0 +1,84 @@
+"""Test per core.youtube_search."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from core import youtube_search
+
+
+def _mock_ytdlp_result(entries: list) -> MagicMock:
+    """Mock subprocess.CompletedProcess con JSON stub."""
+    result = MagicMock()
+    result.returncode = 0
+    result.stdout = json.dumps({"entries": entries})
+    result.stderr = ""
+    return result
+
+
+class TestSearchYoutube:
+    def test_parses_entries(self):
+        entries = [
+            {
+                "id": "abc123",
+                "url": "https://www.youtube.com/watch?v=abc123",
+                "title": "Kapuchon - Hot Sauce (Official Video)",
+                "uploader": "Kapuchon Official",
+                "duration": 336,
+            },
+            {
+                "id": "def456",
+                "url": "https://www.youtube.com/watch?v=def456",
+                "title": "Solomun @ Cocoricò 2024",
+                "uploader": "Cocoricò",
+                "duration": 3600,
+            },
+        ]
+        with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \
+             patch("core.youtube_search.subprocess.run", return_value=_mock_ytdlp_result(entries)):
+            result = youtube_search.search_youtube("kapuchon hot sauce", limit=50)
+        assert len(result) == 2
+        assert result[0]["title"] == "Kapuchon - Hot Sauce (Official Video)"
+        assert result[0]["channel"] == "Kapuchon Official"
+        assert result[0]["duration_sec"] == 336
+        assert result[0]["url"] == "https://www.youtube.com/watch?v=abc123"
+
+    def test_empty_entries_returns_empty(self):
+        with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \
+             patch("core.youtube_search.subprocess.run", return_value=_mock_ytdlp_result([])):
+            result = youtube_search.search_youtube("no-match", limit=50)
+        assert result == []
+
+    def test_raises_when_ytdlp_missing(self):
+        with patch("core.youtube_search.find_ytdlp", return_value=None):
+            with pytest.raises(RuntimeError, match="yt-dlp"):
+                youtube_search.search_youtube("q", limit=50)
+
+    def test_command_uses_ytsearch_with_limit(self):
+        with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \
+             patch("core.youtube_search.subprocess.run", return_value=_mock_ytdlp_result([])) as mock_run:
+            youtube_search.search_youtube("solomun", limit=30)
+        args = mock_run.call_args.args[0]
+        assert args[0] == "/fake/yt-dlp"
+        search_arg = [a for a in args if a.startswith("ytsearch")]
+        assert search_arg == ["ytsearch30:solomun"]
+
+    def test_timeout_raises_runtime_error(self):
+        import subprocess as sp
+        with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \
+             patch("core.youtube_search.subprocess.run", side_effect=sp.TimeoutExpired("yt-dlp", 30)):
+            with pytest.raises(RuntimeError, match="[Tt]imeout"):
+                youtube_search.search_youtube("q", limit=50)
+
+    def test_nonzero_exit_raises(self):
+        bad = MagicMock()
+        bad.returncode = 1
+        bad.stdout = ""
+        bad.stderr = "some yt-dlp error"
+        with patch("core.youtube_search.find_ytdlp", return_value="/fake/yt-dlp"), \
+             patch("core.youtube_search.subprocess.run", return_value=bad):
+            with pytest.raises(RuntimeError, match="yt-dlp"):
+                youtube_search.search_youtube("q", limit=50)