oop7
36eff02b18
Introduce a shake animation to visually indicate invalid inputs and wire it into validation/error flows. Changes: import QPoint; add animate_widget_shake(QWidget) implementation that uses QPropertyAnimation on widget.pos with keyframes; call the animation when URL/path/format validations fail and replace some direct status_label updates with set_status_message_animated. Updated files: ytsage/gui/ytsage_gui_main.py and ytsage/gui/ytsage_gui_analysis.py.
371 lines
14 KiB
Python
371 lines
14 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
|
|
from PySide6.QtWidgets import QMessageBox
|
|
|
|
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
|
|
|
|
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)
|
|
|
|
# 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"))
|
|
|
|
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])
|
|
|
|
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:
|
|
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"))
|
|
|
|
# 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 = subprocess.run(
|
|
cmd, capture_output=True, text=True, timeout=300,
|
|
creationflags=SUBPROCESS_CREATIONFLAGS
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
logger.error("Analysis timed out")
|
|
self.analysis_error.emit(_("errors.timeout"))
|
|
return
|
|
|
|
if self._cancelled:
|
|
return
|
|
|
|
if result.returncode != 0:
|
|
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"))
|
|
|
|
# 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
|
|
playlist_entries = first_info.get("entries", [])
|
|
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"))
|
|
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 = subprocess.run(
|
|
cmd_single, capture_output=True, text=True, timeout=60,
|
|
creationflags=SUBPROCESS_CREATIONFLAGS
|
|
)
|
|
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:
|
|
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"))
|
|
result_data["all_formats"] = video_info.get("formats", [])
|
|
|
|
# Get thumbnail URL
|
|
self.status_update.emit(_("main_ui.analyzing_loading_thumbnail_ytdlp"))
|
|
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"))
|
|
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"))
|
|
|
|
# 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
|
|
|
|
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)
|
|
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.is_analyzing = True
|
|
|
|
# Create and configure the analysis thread
|
|
self._analysis_thread = AnalysisThread(
|
|
url=url,
|
|
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.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 _on_analysis_complete(self, result_data: Dict[str, Any]) -> None:
|
|
"""Handle successful analysis completion - runs in main thread."""
|
|
self = cast("YTSageApp", self)
|
|
|
|
# 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.all_formats = result_data["all_formats"]
|
|
self.available_subtitles = result_data["available_subtitles"]
|
|
self.available_automatic_subtitles = result_data["available_automatic_subtitles"]
|
|
self.selected_playlist_items = None
|
|
self.selected_subtitles = []
|
|
|
|
# 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
|
|
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.filter_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)
|
|
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
|