oop7
49dde84481
Emit progress updates during URL analysis and simulate gradual progress for long-running yt-dlp operations. Added QTimer import and a progress_update Signal on AnalysisThread, with emits at multiple analysis milestones (e.g. 15, 30, 60, 70, 85, 90, 92, 95, 100). In AnalysisMixin introduced _analysis_timer and _fake_progress, connected thread progress to _handle_analysis_progress, and implemented a _update_fake_progress timer that slowly advances the progress bar between real updates. Timers are stopped on real progress, completion or error and the UI progress is reset appropriately to improve UX during long extraction steps.
437 lines
17 KiB
Python
437 lines
17 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_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)
|
|
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])
|
|
|
|
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"))
|
|
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 = 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"))
|
|
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
|
|
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"))
|
|
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 = 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"))
|
|
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)
|
|
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.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)
|
|
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
|