From 8dd8a3c1adf38d8aa8a0b1b9b8c5e1d4292c3945 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jaroslav=20Bene=C5=A1?= Date: Sat, 25 Jul 2026 01:39:20 +0200 Subject: [PATCH] Add YtdlpClient: shared JSON invocation layer for yt-dlp Every metadata call site previously built its own command list. The new ytsage/core/ytsage_client.py centralizes binary resolution (managed, verified binary only), cookie/proxy args from ConfigManager (with session-state overrides), utf-8 output handling, process-group-safe timeouts, and a short-TTL cache for flat/search results. Provides fetch_video_info, fetch_flat_info, fetch_flat_entries (with -I range pagination), search (ytsearchN:), and fetch_account_feed (youtube.com/feed/subscriptions with cookies) plus a generic YtdlpWorker QThread. AnalysisThread now delegates its subprocess execution to the shared runner; its signal surface is unchanged. Groundwork for the SageTube watch/search/browse/feed features. Co-Authored-By: Claude Fable 5 --- ytsage/core/ytsage_client.py | 262 ++++++++++++++++++++++++++++++ ytsage/gui/ytsage_gui_analysis.py | 45 +---- 2 files changed, 265 insertions(+), 42 deletions(-) create mode 100644 ytsage/core/ytsage_client.py diff --git a/ytsage/core/ytsage_client.py b/ytsage/core/ytsage_client.py new file mode 100644 index 0000000..d3f27dd --- /dev/null +++ b/ytsage/core/ytsage_client.py @@ -0,0 +1,262 @@ +""" +YtdlpClient - shared yt-dlp invocation layer +============================================ + +Single place that knows how to build and run yt-dlp commands for metadata +purposes (analysis, search, channel/playlist browsing, subscription feeds). +Every caller previously built its own command list; new features should go +through this client so cookies, proxies, timeouts and process-group cleanup +behave identically everywhere. + +Downloads keep using DownloadThread (streaming progress parsing); this client +is for JSON-returning invocations. +""" + +import json +import os +import signal +import subprocess +import sys +import threading +import time +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Tuple + +from PySide6.QtCore import QThread, Signal + +from .ytsage_yt_dlp import get_yt_dlp_path +from ..utils.ytsage_config_manager import ConfigManager +from ..utils.ytsage_constants import SUBPROCESS_CREATIONFLAGS +from ..utils.ytsage_logger import logger + +# Flat-entry results are cached briefly so tab switches and pagination don't +# hammer YouTube with identical requests +_CACHE_TTL_SECONDS = 600 +_cache: Dict[Tuple[str, ...], Tuple[float, Any]] = {} +_cache_lock = threading.Lock() + + +class YtdlpNotInstalledError(FileNotFoundError): + """Raised when neither the managed binary nor an opted-in system yt-dlp exists.""" + + +def run_ytdlp_capture(cmd: List[str], timeout: int) -> subprocess.CompletedProcess: + """Run yt-dlp capturing output, killing the whole process group on timeout. + + subprocess.run() only kills the direct child on TimeoutExpired, leaking + grandchildren (deno); it also loses any stderr produced before the + timeout. Use Popen with a new session so the entire group can be reaped, + and surface the partial stderr on the raised exception. + """ + popen_kwargs: Dict[str, Any] = { + "stdout": subprocess.PIPE, + "stderr": subprocess.PIPE, + "text": True, + "encoding": "utf-8", + "errors": "replace", + } + if sys.platform == "win32": + popen_kwargs["creationflags"] = SUBPROCESS_CREATIONFLAGS + else: + popen_kwargs["start_new_session"] = True + + proc = subprocess.Popen(cmd, **popen_kwargs) + try: + stdout, stderr = proc.communicate(timeout=timeout) + except subprocess.TimeoutExpired as exc: + if sys.platform == "win32": + subprocess.run( + ["taskkill", "/F", "/T", "/PID", str(proc.pid)], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + creationflags=SUBPROCESS_CREATIONFLAGS, + ) + else: + try: + os.killpg(os.getpgid(proc.pid), signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + pass + stdout, stderr = proc.communicate() + exc.stdout, exc.stderr = stdout, stderr + raise + return subprocess.CompletedProcess(cmd, proc.returncode, stdout, stderr) + + +class YtdlpClient: + """Builds and runs JSON-returning yt-dlp commands with the app's + cookie/proxy configuration applied.""" + + def __init__( + self, + cookie_file_path: Optional[str] = None, + browser_cookies: Optional[str] = None, + use_config_auth: bool = True, + ) -> None: + """ + Args: + cookie_file_path / browser_cookies: explicit overrides for session + state that differs from the persisted config. + use_config_auth: when True (default) and no override is given, + cookie and proxy options are read from ConfigManager. + """ + self._cookie_file_override = cookie_file_path + self._browser_cookies_override = browser_cookies + self._use_config_auth = use_config_auth + + # ------------------------------------------------------------- commands + + def resolve_binary(self) -> str: + path = get_yt_dlp_path() + if str(path) == "yt-dlp": + raise YtdlpNotInstalledError("yt-dlp is not installed - run the yt-dlp setup first") + return str(path) + + def _auth_args(self) -> List[str]: + args: List[str] = [] + cookie_file = self._cookie_file_override + browser_cookies = self._browser_cookies_override + + if cookie_file is None and browser_cookies is None and self._use_config_auth: + if ConfigManager.get("cookie_active"): + if ConfigManager.get("cookie_source") == "file": + saved = ConfigManager.get("cookie_file_path") + if saved and Path(saved).exists(): + cookie_file = str(saved) + else: + browser = ConfigManager.get("cookie_browser") + profile = ConfigManager.get("cookie_browser_profile") + if browser: + browser_cookies = f"{browser}:{profile}" if profile else browser + + if cookie_file: + args.extend(["--cookies", str(cookie_file)]) + elif browser_cookies: + args.extend(["--cookies-from-browser", browser_cookies]) + + if self._use_config_auth: + proxy = ConfigManager.get("proxy_url") + geo_proxy = ConfigManager.get("geo_proxy_url") + if proxy: + args.extend(["--proxy", str(proxy)]) + if geo_proxy: + args.extend(["--geo-verification-proxy", str(geo_proxy)]) + return args + + def build_base_cmd(self) -> List[str]: + return [self.resolve_binary(), "--no-warnings", "--no-color"] + self._auth_args() + + def run_json(self, extra_args: List[str], timeout: int = 60) -> Any: + """Run yt-dlp with --dump-single-json semantics and parse stdout.""" + cmd = self.build_base_cmd() + extra_args + logger.debug(f"YtdlpClient executing: {cmd}") + result = run_ytdlp_capture(cmd, timeout=timeout) + if result.returncode != 0: + stderr_tail = (result.stderr or "").strip()[-1000:] + raise RuntimeError(stderr_tail or f"yt-dlp exited with code {result.returncode}") + return json.loads(result.stdout) + + # ------------------------------------------------------------- queries + + def fetch_video_info(self, url: str, timeout: int = 300) -> Dict[str, Any]: + """Full info dict for a single video, including formats[].url.""" + return self.run_json(["--dump-single-json", url], timeout=timeout) + + def fetch_flat_info(self, url: str, timeout: int = 300) -> Dict[str, Any]: + """Flat info dict (playlist/channel entries without per-video formats).""" + return self.run_json(["--dump-single-json", "--flat-playlist", url], timeout=timeout) + + def fetch_flat_entries( + self, url: str, start: int = 1, end: int = 30, timeout: int = 120, use_cache: bool = True + ) -> List[Dict[str, Any]]: + """Flat entries start..end (1-based, inclusive) of a playlist/channel URL.""" + cache_key = ("flat", url, str(start), str(end), *self._auth_args()) + if use_cache: + cached = _cache_get(cache_key) + if cached is not None: + return cached + info = self.run_json( + ["--dump-single-json", "--flat-playlist", "-I", f"{start}:{end}", url], + timeout=timeout, + ) + entries = [e for e in info.get("entries", []) if e] if isinstance(info, dict) else [] + _cache_put(cache_key, entries) + return entries + + def search( + self, query: str, n: int = 25, offset: int = 0, timeout: int = 120, use_cache: bool = True + ) -> List[Dict[str, Any]]: + """YouTube search returning flat entries n results at a time.""" + cache_key = ("search", query, str(n), str(offset), *self._auth_args()) + if use_cache: + cached = _cache_get(cache_key) + if cached is not None: + return cached + total = offset + n + info = self.run_json( + [ + "--dump-single-json", + "--flat-playlist", + "-I", + f"{offset + 1}:{total}", + f"ytsearch{total}:{query}", + ], + timeout=timeout, + ) + entries = [e for e in info.get("entries", []) if e] if isinstance(info, dict) else [] + _cache_put(cache_key, entries) + return entries + + def fetch_account_feed(self, n: int = 50, timeout: int = 180) -> List[Dict[str, Any]]: + """The logged-in account's subscription feed. Requires active cookies.""" + return self.fetch_flat_entries( + "https://www.youtube.com/feed/subscriptions", start=1, end=n, + timeout=timeout, use_cache=False, + ) + + +class YtdlpWorker(QThread): + """Generic worker running one YtdlpClient call off the GUI thread. + + Usage: + worker = YtdlpWorker(lambda c: c.search("query", 25)) + worker.result.connect(...); worker.error.connect(...) + worker.start() + """ + + result = Signal(object) + error = Signal(str) + + def __init__(self, fn: Callable[[YtdlpClient], Any], client: Optional[YtdlpClient] = None, parent=None) -> None: + super().__init__(parent) + self._fn = fn + self._client = client or YtdlpClient() + + def run(self) -> None: + try: + self.result.emit(self._fn(self._client)) + except Exception as e: + logger.exception(f"YtdlpWorker failed: {e}") + self.error.emit(str(e)) + + +# ------------------------------------------------------------------ cache + +def _cache_get(key: Tuple[str, ...]) -> Optional[Any]: + with _cache_lock: + hit = _cache.get(key) + if hit and time.time() - hit[0] < _CACHE_TTL_SECONDS: + return hit[1] + if hit: + del _cache[key] + return None + + +def _cache_put(key: Tuple[str, ...], value: Any) -> None: + with _cache_lock: + if len(_cache) > 256: + _cache.clear() + _cache[key] = (time.time(), value) + + +def clear_cache() -> None: + with _cache_lock: + _cache.clear() diff --git a/ytsage/gui/ytsage_gui_analysis.py b/ytsage/gui/ytsage_gui_analysis.py index edf87c6..0bd8ec9 100644 --- a/ytsage/gui/ytsage_gui_analysis.py +++ b/ytsage/gui/ytsage_gui_analysis.py @@ -1,16 +1,13 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast import json -import os -import signal import subprocess -import sys from PySide6.QtCore import QMetaObject, Qt, Q_ARG, QThread, Signal, QTimer from PySide6.QtWidgets import QMessageBox +from ..core.ytsage_client import run_ytdlp_capture from ..core.ytsage_utils import validate_video_url, parse_yt_dlp_error from ..core.ytsage_yt_dlp import get_yt_dlp_path -from ..utils.ytsage_constants import SUBPROCESS_CREATIONFLAGS from ..utils.ytsage_localization import _ from ..utils.ytsage_logger import logger @@ -100,44 +97,8 @@ class AnalysisThread(QThread): @staticmethod def _run_ytdlp(cmd: list, timeout: int) -> subprocess.CompletedProcess: - """Run yt-dlp capturing output, killing the whole process group on timeout. - - subprocess.run() only kills the direct child on TimeoutExpired, leaking - grandchildren (deno); it also loses any stderr produced before the - timeout. Use Popen with a new session so the entire group can be - reaped, and surface the partial stderr in the raised exception. - """ - popen_kwargs: Dict[str, Any] = { - "stdout": subprocess.PIPE, - "stderr": subprocess.PIPE, - "text": True, - "encoding": "utf-8", - "errors": "replace", - } - if sys.platform == "win32": - popen_kwargs["creationflags"] = SUBPROCESS_CREATIONFLAGS - else: - popen_kwargs["start_new_session"] = True - - proc = subprocess.Popen(cmd, **popen_kwargs) - try: - stdout, stderr = proc.communicate(timeout=timeout) - except subprocess.TimeoutExpired as exc: - if sys.platform == "win32": - subprocess.run( - ["taskkill", "/F", "/T", "/PID", str(proc.pid)], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, - creationflags=SUBPROCESS_CREATIONFLAGS, - ) - else: - try: - os.killpg(os.getpgid(proc.pid), signal.SIGKILL) - except (ProcessLookupError, PermissionError, OSError): - pass - stdout, stderr = proc.communicate() - exc.stdout, exc.stderr = stdout, stderr - raise - return subprocess.CompletedProcess(cmd, proc.returncode, stdout, stderr) + """Run yt-dlp via the shared client runner (process-group-safe).""" + return run_ytdlp_capture(cmd, timeout=timeout) def _analyze_url_with_subprocess(self, url: str) -> None: """Analyze URL using yt-dlp executable."""