Jaroslav Beneš
8dd8a3c1ad
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 <noreply@anthropic.com>
471 lines
19 KiB
Python
471 lines
19 KiB
Python
from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
|
|
|
|
import json
|
|
import subprocess
|
|
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_localization import _
|
|
from ..utils.ytsage_logger import logger
|
|
|
|
if TYPE_CHECKING:
|
|
from .ytsage_gui_main import YTSageApp
|
|
|
|
|
|
class AnalysisThread(QThread):
|
|
"""
|
|
Thread-safe QThread for URL analysis.
|
|
All results are passed back via signals to ensure thread safety.
|
|
"""
|
|
# Signals for status updates
|
|
status_update = Signal(str)
|
|
progress_update = Signal(int) # Signal for progress bar updates
|
|
|
|
# Signals for playlist UI
|
|
playlist_info_visible = Signal(bool)
|
|
playlist_info_text = Signal(str)
|
|
playlist_select_btn_visible = Signal(bool)
|
|
playlist_select_btn_text = Signal(str)
|
|
|
|
# Signal for analysis results - passes all data at once
|
|
analysis_complete = Signal(dict)
|
|
|
|
# Signal for errors
|
|
analysis_error = Signal(str)
|
|
|
|
# Signal when thread finishes (success or failure)
|
|
analysis_finished = Signal()
|
|
|
|
def __init__(
|
|
self,
|
|
url: str,
|
|
cookie_file_path: Optional[str] = None,
|
|
browser_cookies_option: Optional[str] = None,
|
|
proxy_url: Optional[str] = None,
|
|
geo_proxy_url: Optional[str] = None,
|
|
parent=None
|
|
) -> None:
|
|
super().__init__(parent)
|
|
self.url = url
|
|
self.cookie_file_path = cookie_file_path
|
|
self.browser_cookies_option = browser_cookies_option
|
|
self.proxy_url = proxy_url
|
|
self.geo_proxy_url = geo_proxy_url
|
|
self._cancelled = False
|
|
|
|
def cancel(self) -> None:
|
|
"""Request cancellation of the analysis."""
|
|
self._cancelled = True
|
|
|
|
def run(self) -> None:
|
|
"""Main thread execution - performs URL analysis."""
|
|
try:
|
|
self.status_update.emit(_("main_ui.analyzing_extracting_basic"))
|
|
self.progress_update.emit(15)
|
|
|
|
url = self.url
|
|
# Clean up the URL to handle both playlist and video URLs
|
|
if "list=" in url and "watch?v=" in url:
|
|
playlist_id = url.split("list=")[1].split("&")[0]
|
|
url = f"https://www.youtube.com/playlist?list={playlist_id}"
|
|
|
|
self._analyze_url_with_subprocess(url)
|
|
|
|
except Exception as e:
|
|
logger.exception(f"Error in analysis: {e}")
|
|
self.analysis_error.emit(_("errors.generic_error", error=str(e)))
|
|
self.playlist_info_visible.emit(False)
|
|
self.playlist_select_btn_visible.emit(False)
|
|
finally:
|
|
self.analysis_finished.emit()
|
|
|
|
def _add_auth_options(self, cmd: List[str]) -> None:
|
|
"""Add authentication and proxy options to command."""
|
|
if self.cookie_file_path:
|
|
cmd.extend(["--cookies", str(self.cookie_file_path)])
|
|
elif self.browser_cookies_option:
|
|
cmd.extend(["--cookies-from-browser", self.browser_cookies_option])
|
|
|
|
if self.proxy_url:
|
|
cmd.extend(["--proxy", self.proxy_url])
|
|
|
|
if self.geo_proxy_url:
|
|
cmd.extend(["--geo-verification-proxy", self.geo_proxy_url])
|
|
|
|
@staticmethod
|
|
def _run_ytdlp(cmd: list, timeout: int) -> subprocess.CompletedProcess:
|
|
"""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."""
|
|
if self._cancelled:
|
|
return
|
|
|
|
yt_dlp_path = get_yt_dlp_path()
|
|
if not yt_dlp_path or str(yt_dlp_path) == "yt-dlp":
|
|
# Sentinel: no managed binary and no opted-in system binary.
|
|
# Never exec a bare command name from PATH.
|
|
logger.error("yt-dlp executable not found. Please install yt-dlp first.")
|
|
self.analysis_error.emit(_("errors.ytdlp_not_found"))
|
|
self.playlist_info_visible.emit(False)
|
|
self.playlist_select_btn_visible.emit(False)
|
|
return
|
|
|
|
self.status_update.emit(_("main_ui.analyzing_extracting_ytdlp"))
|
|
self.progress_update.emit(30) # This will trigger fake progress in UI
|
|
|
|
# Build command for basic info extraction
|
|
cmd = [yt_dlp_path, "--dump-single-json", "--flat-playlist", "--no-warnings", url]
|
|
self._add_auth_options(cmd)
|
|
|
|
logger.debug(f"Executing yt-dlp command: {cmd}")
|
|
|
|
try:
|
|
result = self._run_ytdlp(cmd, timeout=300)
|
|
except subprocess.TimeoutExpired as e:
|
|
stderr_tail = (e.stderr or "")[-500:] if isinstance(e.stderr, str) else ""
|
|
logger.error(f"Analysis timed out. Partial stderr: {stderr_tail}")
|
|
self.analysis_error.emit(_("errors.timeout"))
|
|
return
|
|
|
|
if self._cancelled:
|
|
return
|
|
|
|
if result.returncode != 0:
|
|
if "Private video" in result.stderr or "Sign in" in result.stderr:
|
|
logger.error(f"yt-dlp failed (private video): {result.stderr}")
|
|
self.analysis_error.emit(_("errors.private_video"))
|
|
else:
|
|
logger.error(f"yt-dlp failed: {result.stderr}")
|
|
self.analysis_error.emit(_("errors.ytdlp_failed", error=result.stderr))
|
|
|
|
self.playlist_info_visible.emit(False)
|
|
self.playlist_select_btn_visible.emit(False)
|
|
return
|
|
|
|
json_lines = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
|
|
|
|
if not json_lines:
|
|
logger.error("No data returned from yt-dlp")
|
|
self.analysis_error.emit(_("errors.no_data_returned"))
|
|
self.playlist_info_visible.emit(False)
|
|
self.playlist_select_btn_visible.emit(False)
|
|
return
|
|
|
|
try:
|
|
first_info = json.loads(json_lines[0])
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"Failed to parse yt-dlp output: {e}")
|
|
self.analysis_error.emit(_("errors.parse_failed", error=str(e)))
|
|
self.playlist_info_visible.emit(False)
|
|
self.playlist_select_btn_visible.emit(False)
|
|
return
|
|
|
|
if self._cancelled:
|
|
return
|
|
|
|
self.status_update.emit(_("main_ui.analyzing_processing_data"))
|
|
self.progress_update.emit(60)
|
|
|
|
# Prepare result data
|
|
result_data: Dict[str, Any] = {
|
|
"is_playlist": False,
|
|
"playlist_info": None,
|
|
"playlist_entries": [],
|
|
"video_info": None,
|
|
"all_formats": [],
|
|
"available_subtitles": {},
|
|
"available_automatic_subtitles": {},
|
|
"thumbnail_url": None,
|
|
}
|
|
|
|
if first_info.get("_type") == "playlist":
|
|
result_data["is_playlist"] = True
|
|
result_data["playlist_info"] = first_info
|
|
# Private/deleted videos can appear as None entries in flat playlists
|
|
playlist_entries = [e for e in first_info.get("entries", []) if e]
|
|
result_data["playlist_entries"] = playlist_entries
|
|
|
|
if not playlist_entries:
|
|
logger.error("Playlist contains no valid videos.")
|
|
self.analysis_error.emit(_("errors.playlist_no_videos"))
|
|
self.playlist_info_visible.emit(False)
|
|
self.playlist_select_btn_visible.emit(False)
|
|
return
|
|
|
|
# Fetch full info for the first video to get formats
|
|
self.status_update.emit(_("main_ui.analyzing_fetching_first_video"))
|
|
self.progress_update.emit(70)
|
|
first_video_entry = playlist_entries[0]
|
|
first_video_url = first_video_entry.get("url")
|
|
|
|
cmd_single = [yt_dlp_path, "--dump-single-json", "--no-warnings", first_video_url]
|
|
self._add_auth_options(cmd_single)
|
|
|
|
try:
|
|
result_single = self._run_ytdlp(cmd_single, timeout=60)
|
|
if result_single.returncode == 0:
|
|
result_data["video_info"] = json.loads(result_single.stdout)
|
|
else:
|
|
result_data["video_info"] = first_video_entry
|
|
except (subprocess.TimeoutExpired, json.JSONDecodeError):
|
|
result_data["video_info"] = first_video_entry
|
|
|
|
if self._cancelled:
|
|
return
|
|
|
|
# Update playlist UI via signals
|
|
playlist_text = _("playlist.display_format",
|
|
title=first_info.get('title', _('playlist.unknown')),
|
|
count=len(playlist_entries))
|
|
self.playlist_info_text.emit(playlist_text)
|
|
self.playlist_info_visible.emit(True)
|
|
self.playlist_select_btn_text.emit(_("main_ui.select_videos_all"))
|
|
self.playlist_select_btn_visible.emit(True)
|
|
|
|
else:
|
|
# Handle single video
|
|
result_data["is_playlist"] = False
|
|
result_data["video_info"] = first_info
|
|
result_data["playlist_entries"] = []
|
|
result_data["playlist_info"] = None
|
|
|
|
# Hide playlist UI
|
|
self.playlist_info_visible.emit(False)
|
|
self.playlist_select_btn_visible.emit(False)
|
|
|
|
# Verify we have format information
|
|
video_info = result_data["video_info"]
|
|
if not video_info or "formats" not in video_info:
|
|
logger.error("No format information available")
|
|
self.analysis_error.emit(_("errors.no_format_info"))
|
|
self.playlist_info_visible.emit(False)
|
|
self.playlist_select_btn_visible.emit(False)
|
|
return
|
|
|
|
self.status_update.emit(_("main_ui.analyzing_processing_formats_ytdlp"))
|
|
self.progress_update.emit(85)
|
|
result_data["all_formats"] = video_info.get("formats", [])
|
|
|
|
# Get thumbnail URL
|
|
self.status_update.emit(_("main_ui.analyzing_loading_thumbnail_ytdlp"))
|
|
self.progress_update.emit(90)
|
|
playlist_info = result_data.get("playlist_info") or {}
|
|
thumbnail_url = playlist_info.get("thumbnail") or video_info.get("thumbnail")
|
|
result_data["thumbnail_url"] = thumbnail_url
|
|
|
|
# Handle subtitles
|
|
self.status_update.emit(_("main_ui.analyzing_processing_subtitles_ytdlp"))
|
|
self.progress_update.emit(92)
|
|
result_data["available_subtitles"] = video_info.get("subtitles", {})
|
|
result_data["available_automatic_subtitles"] = video_info.get("automatic_captions", {})
|
|
|
|
self.status_update.emit(_("main_ui.analyzing_updating_table"))
|
|
self.progress_update.emit(95)
|
|
|
|
# Emit all results at once
|
|
self.analysis_complete.emit(result_data)
|
|
|
|
|
|
class AnalysisMixin:
|
|
"""Mixin class providing URL analysis functionality for YTSageApp."""
|
|
|
|
# Track the current analysis thread
|
|
_analysis_thread: Optional[AnalysisThread] = None
|
|
_analysis_timer: Optional[QTimer] = None
|
|
_fake_progress: int = 0
|
|
|
|
def analyze_url(self) -> None:
|
|
"""Start URL analysis in a background thread."""
|
|
self = cast("YTSageApp", self)
|
|
|
|
if self.is_updating_ytdlp:
|
|
QMessageBox.warning(self, _("update.update_in_progress_title"), _("update.update_in_progress_message"))
|
|
return
|
|
|
|
url = self.url_input.text().strip()
|
|
if not url:
|
|
self.signals.update_status.emit(_("main_ui.invalid_url_or_enter"))
|
|
if hasattr(self, "animate_widget_shake"):
|
|
self.animate_widget_shake(self.url_input)
|
|
return
|
|
|
|
# Validate URL before processing
|
|
is_valid, error_message = validate_video_url(url, generic_mode=self.generic_mode_enabled)
|
|
if not is_valid:
|
|
QMessageBox.warning(self, _("main_ui.error_title"), error_message)
|
|
if hasattr(self, "animate_widget_shake"):
|
|
self.animate_widget_shake(self.url_input)
|
|
return
|
|
|
|
# Cancel any existing analysis thread
|
|
if self._analysis_thread is not None and self._analysis_thread.isRunning():
|
|
self._analysis_thread.cancel()
|
|
self._analysis_thread.wait(1000) # Wait up to 1 second
|
|
|
|
# Reset analysis state and disable controls
|
|
self.analysis_completed = False
|
|
self.toggle_analysis_dependent_controls(enabled=False)
|
|
|
|
self.signals.update_status.emit(_("main_ui.analyzing_preparing"))
|
|
self.signals.update_progress.emit(0) # Reset progress
|
|
self.is_analyzing = True
|
|
|
|
# Stop any existing timer
|
|
if self._analysis_timer:
|
|
self._analysis_timer.stop()
|
|
self._analysis_timer = None
|
|
|
|
# Create and configure the analysis thread
|
|
self._analysis_thread = AnalysisThread(
|
|
url=url,
|
|
# ... arguments will be filled by ... usage below ...
|
|
cookie_file_path=self.cookie_file_path,
|
|
browser_cookies_option=self.browser_cookies_option,
|
|
proxy_url=self.proxy_url,
|
|
geo_proxy_url=self.geo_proxy_url,
|
|
parent=self
|
|
)
|
|
|
|
# Connect signals to handlers
|
|
self._analysis_thread.status_update.connect(self.signals.update_status.emit)
|
|
self._analysis_thread.progress_update.connect(self._handle_analysis_progress)
|
|
self._analysis_thread.playlist_info_visible.connect(self.signals.playlist_info_label_visible.emit)
|
|
self._analysis_thread.playlist_info_text.connect(self.signals.playlist_info_label_text.emit)
|
|
self._analysis_thread.playlist_select_btn_visible.connect(self.signals.playlist_select_btn_visible.emit)
|
|
self._analysis_thread.playlist_select_btn_text.connect(self.signals.playlist_select_btn_text.emit)
|
|
self._analysis_thread.analysis_complete.connect(self._on_analysis_complete)
|
|
self._analysis_thread.analysis_error.connect(self._on_analysis_error)
|
|
self._analysis_thread.analysis_finished.connect(self._on_analysis_finished)
|
|
|
|
# Start the thread
|
|
self._analysis_thread.start()
|
|
|
|
def _handle_analysis_progress(self, value: int) -> None:
|
|
"""Handle progress updates from analysis thread."""
|
|
self = cast("YTSageApp", self)
|
|
|
|
# Stop fake timer if running on any real update
|
|
if self._analysis_timer:
|
|
self._analysis_timer.stop()
|
|
self._analysis_timer = None
|
|
|
|
self.signals.update_progress.emit(value)
|
|
|
|
# If we hit the extraction phase (30%), start fake progress
|
|
if value == 30:
|
|
self._fake_progress = 30
|
|
self._analysis_timer = QTimer(self)
|
|
self._analysis_timer.timeout.connect(self._update_fake_progress)
|
|
self._analysis_timer.start(200) # Every 200ms
|
|
|
|
def _update_fake_progress(self) -> None:
|
|
"""Increment progress bar slowly during long operations."""
|
|
self = cast("YTSageApp", self)
|
|
|
|
# Asymptotically approach 85%
|
|
if self._fake_progress < 85:
|
|
# Slow down as we get higher
|
|
increment = 1
|
|
if self._fake_progress > 60:
|
|
if self._fake_progress % 3 == 0: # Slower
|
|
increment = 1
|
|
else:
|
|
increment = 0
|
|
|
|
if increment > 0:
|
|
self._fake_progress += increment
|
|
self.signals.update_progress.emit(self._fake_progress)
|
|
|
|
def _on_analysis_complete(self, result_data: Dict[str, Any]) -> None:
|
|
"""Handle successful analysis completion - runs in main thread."""
|
|
self = cast("YTSageApp", self)
|
|
|
|
# Stop fake timer
|
|
if self._analysis_timer:
|
|
self._analysis_timer.stop()
|
|
self._analysis_timer = None
|
|
|
|
self.signals.update_progress.emit(100)
|
|
|
|
# Update instance variables with results (safe - we're in main thread)
|
|
self.is_playlist = result_data["is_playlist"]
|
|
self.playlist_info = result_data["playlist_info"]
|
|
self.playlist_entries = result_data["playlist_entries"]
|
|
self.video_info = result_data["video_info"]
|
|
self.available_subtitles = result_data["available_subtitles"]
|
|
self.available_automatic_subtitles = result_data["available_automatic_subtitles"]
|
|
self.selected_playlist_items = None
|
|
self.selected_subtitles = []
|
|
|
|
from ..utils.ytsage_config_manager import ConfigManager
|
|
default_sub = ConfigManager.get("default_subtitle_language")
|
|
if default_sub:
|
|
if isinstance(default_sub, str):
|
|
default_sub_list = [s.strip() for s in default_sub.split(",")]
|
|
else:
|
|
default_sub_list = []
|
|
|
|
for lang in default_sub_list:
|
|
if lang in self.available_subtitles:
|
|
self.selected_subtitles.append(f"{lang} - Manual")
|
|
elif lang in self.available_automatic_subtitles:
|
|
self.selected_subtitles.append(f"{lang} - Auto-generated")
|
|
|
|
count = len(self.selected_subtitles)
|
|
try:
|
|
self.signals.selected_subs_label_text.emit(_("subtitle_selection.count_selected", count=count))
|
|
self.subtitle_select_btn.setProperty("subtitlesSelected", count > 0)
|
|
self.subtitle_select_btn.style().unpolish(self.subtitle_select_btn)
|
|
self.subtitle_select_btn.style().polish(self.subtitle_select_btn)
|
|
except Exception:
|
|
pass
|
|
|
|
# Update UI components (safe - we're in main thread)
|
|
self.update_video_info(self.video_info)
|
|
|
|
# Download thumbnail
|
|
thumbnail_url = result_data.get("thumbnail_url")
|
|
if thumbnail_url:
|
|
self.download_thumbnail(thumbnail_url)
|
|
|
|
# Save thumbnail if enabled
|
|
if self.save_thumbnail:
|
|
self.download_thumbnail_file(self.video_url, self.last_path)
|
|
|
|
# Update subtitle UI
|
|
count = len(self.selected_subtitles)
|
|
if count > 0:
|
|
self.signals.selected_subs_label_text.emit(_("subtitle_selection.count_selected", count=count))
|
|
else:
|
|
self.signals.selected_subs_label_text.emit(_("main_ui.zero_selected"))
|
|
|
|
# Update format table
|
|
self.video_button.setChecked(True)
|
|
self.audio_button.setChecked(False)
|
|
self.update_format_table(result_data["all_formats"])
|
|
|
|
self.signals.update_status.emit(_("main_ui.analysis_complete"))
|
|
|
|
# Mark analysis as complete and enable controls
|
|
self.analysis_completed = True
|
|
self.toggle_analysis_dependent_controls(enabled=True)
|
|
|
|
def _on_analysis_error(self, error_message: str) -> None:
|
|
"""Handle analysis error - runs in main thread."""
|
|
self = cast("YTSageApp", self)
|
|
if self._analysis_timer:
|
|
self._analysis_timer.stop()
|
|
self._analysis_timer = None
|
|
self.signals.update_progress.emit(0)
|
|
self.signals.update_status.emit(error_message)
|
|
|
|
def _on_analysis_finished(self) -> None:
|
|
"""Handle analysis thread completion - runs in main thread."""
|
|
self = cast("YTSageApp", self)
|
|
self.is_analyzing = False
|