From 0949df031ab618db984387d7693551173faf5a02 Mon Sep 17 00:00:00 2001 From: luciano Date: Tue, 14 Jul 2026 17:18:54 +0200 Subject: [PATCH] youtube: search_youtube via yt-dlp ytsearchN + parse JSON --- core/youtube_search.py | 72 +++++++++++++++++++++++++++++++ tests/test_youtube_search.py | 84 ++++++++++++++++++++++++++++++++++++ 2 files changed, 156 insertions(+) create mode 100644 core/youtube_search.py create mode 100644 tests/test_youtube_search.py diff --git a/core/youtube_search.py b/core/youtube_search.py new file mode 100644 index 0000000..30f2b10 --- /dev/null +++ b/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 diff --git a/tests/test_youtube_search.py b/tests/test_youtube_search.py new file mode 100644 index 0000000..00dda03 --- /dev/null +++ b/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)