Houmeres/SageTube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity
Files
39320db3a0fd4e1850cd76941ec7cf97c5875805
SageTube/ytsage/gui/ytsage_gui_dialogs/ytsage_dialogs_update.py
T
oop7 96dadd7e5b Refactor GUI package structure and update imports 
Moved all files from src/gui/ to ytsage/gui/ and updated import statements to use relative imports within the new package structure. This improves modularity and prepares the codebase for distribution as a proper Python package.
2026-01-25 14:07:51 +02:00

486 lines
19 KiB
Python
Raw Blame History

"""
Update-related dialogs and threads for YTSage application.
Contains dialogs and background threads for checking and performing yt-dlp binary updates.
Note: This module only handles binary updates. Python package updates have been removed.
"""
import os
import subprocess
import time
from pathlib import Path
import requests
from packaging import version
from PySide6.QtCore import Qt, QThread, QTimer, Signal
from PySide6.QtWidgets import QDialog, QHBoxLayout, QLabel, QProgressBar, QPushButton, QVBoxLayout
from ...core.ytsage_utils import get_ytdlp_version
from ...core.ytsage_yt_dlp import get_yt_dlp_path
from ...utils.ytsage_constants import OS_NAME, SUBPROCESS_CREATIONFLAGS, YTDLP_APP_BIN_PATH, YTDLP_DOWNLOAD_URL
from ...utils.ytsage_config_manager import ConfigManager
from ...utils.ytsage_localization import LocalizationManager
# Shorthand for localization
_ = LocalizationManager.get_text
from ...utils.ytsage_localization import _
from ...utils.ytsage_logger import logger
class VersionCheckThread(QThread):
finished = Signal(str, str, str) # current_version, latest_version, error_message
def run(self) -> None:
current_version = ""
latest_version = ""
error_message = ""
try:
# Get the yt-dlp executable path
yt_dlp_path = get_yt_dlp_path()
# Get current version with timeout
try:
result = subprocess.run(
[yt_dlp_path, "--version"],
capture_output=True,
text=True,
timeout=30, # 30 second timeout
creationflags=SUBPROCESS_CREATIONFLAGS,
)
if result.returncode == 0:
current_version = result.stdout.strip()
else:
error_message = "yt-dlp binary not accessible."
self.finished.emit(current_version, latest_version, error_message)
return
except subprocess.TimeoutExpired:
error_message = "yt-dlp version check timed out."
self.finished.emit(current_version, latest_version, error_message)
return
except Exception as e:
error_message = f"yt-dlp not found or accessible: {e}"
self.finished.emit(current_version, latest_version, error_message)
return
# Get latest version from PyPI (yt-dlp releases are also published to PyPI)
response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10)
response.raise_for_status()
latest_version = response.json()["info"]["version"]
# Clean up version strings
current_version = current_version.replace("_", ".")
latest_version = latest_version.replace("_", ".")
except requests.RequestException as e:
error_message = f"Network error checking PyPI: {e}"
except Exception as e:
error_message = f"Error checking version: {e}"
self.finished.emit(current_version, latest_version, error_message)
class UpdateThread(QThread):
update_status = Signal(str) # For status messages
update_progress = Signal(int) # For progress percentage (0-100)
update_finished = Signal(bool, str) # success (bool), message/error (str)
def run(self) -> None:
error_message = ""
success = False
try:
self.update_status.emit(_('update.checking_current'))
self.update_progress.emit(10)
# Get the yt-dlp path
try:
yt_dlp_path = get_yt_dlp_path()
self.update_status.emit(_('update.found_at', path=yt_dlp_path))
except Exception as e:
self.update_status.emit(_('update.error_getting_path', error=e))
self.update_finished.emit(False, _('update.error_getting_path', error=e))
return
self.update_progress.emit(20)
# Update the binary (no more pip-based updates)
self.update_status.emit(_('update.updating_binary'))
success = self._update_binary(yt_dlp_path)
if success:
self.update_progress.emit(100)
error_message = _('update.update_success')
else:
error_message = _('update.update_failed')
except requests.RequestException as e:
error_message = _('update.network_error', error=e)
self.update_status.emit(error_message)
success = False
except Exception as e:
error_message = _('update.general_error', error=e)
self.update_status.emit(error_message)
success = False
self.update_finished.emit(success, error_message)
def _update_binary(self, yt_dlp_path: Path) -> bool:
"""Update yt-dlp binary using its built-in updater (same logic as AutoUpdateThread)."""
try:
logger.info("UpdateThread: Checking for yt-dlp updates...")
result = subprocess.run(
[yt_dlp_path, "-U"],
capture_output=True,
text=True,
timeout=60,
creationflags=SUBPROCESS_CREATIONFLAGS,
)
if result.returncode == 0:
# Make executable on Unix systems
if OS_NAME != "Windows":
os.chmod(yt_dlp_path, 0o755)
logger.info("UpdateThread: yt-dlp update completed successfully.")
if result.stdout:
logger.debug(f"yt-dlp output: {result.stdout.strip()}")
self.update_status.emit(_('update.binary_updated'))
self.update_progress.emit(95)
return True
else:
logger.error(f"UpdateThread: yt-dlp update failed. {result.stderr.strip()}")
self.update_status.emit(_('update.update_failed_stderr', error=result.stderr.strip()))
return False
except subprocess.TimeoutExpired:
logger.error("UpdateThread: yt-dlp update timed out.")
self.update_status.emit(_('update.update_timeout'))
return False
except Exception as e:
logger.exception(f"UpdateThread: Unexpected error during update: {e}")
self.update_status.emit(_('update.unexpected_error', error=e))
return False
class YTDLPUpdateDialog(QDialog):
def __init__(self, parent=None) -> None:
super().__init__(parent)
self.setWindowTitle(_('update.title'))
self.setMinimumWidth(450)
self.setMinimumHeight(200)
self._closing = False # Flag to track if dialog is closing
layout = QVBoxLayout(self)
# Status label
self.status_label = QLabel(_('update.checking'))
self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.status_label.setWordWrap(True)
self.status_label.setMinimumHeight(60)
layout.addWidget(self.status_label)
# Progress bar
self.progress_bar = QProgressBar()
self.progress_bar.hide() # Hide initially
layout.addWidget(self.progress_bar)
# Buttons
button_layout = QHBoxLayout()
self.update_btn = QPushButton(_('buttons.update'))
self.update_btn.clicked.connect(self.perform_update)
self.update_btn.setEnabled(False)
self.close_btn = QPushButton(_('buttons.close'))
self.close_btn.clicked.connect(self.close)
button_layout.addWidget(self.update_btn)
button_layout.addWidget(self.close_btn)
layout.addLayout(button_layout)
# Style
self.setStyleSheet(
"""
QDialog {
background-color: #15181b;
}
QLabel {
color: #ffffff;
font-size: 12px;
padding: 10px;
}
QPushButton {
padding: 8px 15px;
background-color: #c90000;
border: none;
border-radius: 4px;
color: white;
font-weight: bold;
min-width: 100px;
}
QPushButton:disabled {
background-color: #666666;
}
QPushButton:hover {
background-color: #a50000;
}
QProgressBar {
border: 2px solid #1d1e22;
border-radius: 6px;
text-align: center;
color: white;
background-color: #1d1e22;
height: 30px;
font-weight: bold;
font-size: 12px;
}
QProgressBar::chunk {
background: qlineargradient(x1: 0, y1: 0, x2: 1, y2: 0,
stop: 0 #e60000, stop: 0.5 #ff3333, stop: 1 #c90000);
border-radius: 4px;
margin: 1px;
}
"""
)
# Start version check in background
self.check_version()
def check_version(self) -> None:
self.status_label.setText(_('update.checking'))
self.update_btn.setEnabled(False)
self.version_check_thread = VersionCheckThread()
self.version_check_thread.finished.connect(self.on_version_check_finished)
self.version_check_thread.start()
def on_version_check_finished(self, current_version, latest_version, error_message) -> None:
# Check if dialog is closing to avoid unnecessary updates
if hasattr(self, "_closing") and self._closing:
return
if error_message:
self.status_label.setText(error_message)
self.update_btn.setEnabled(False)
return
if not current_version or not latest_version:
self.status_label.setText(_('update.could_not_determine'))
self.update_btn.setEnabled(False)
return
try:
# Compare versions
current_ver = version.parse(current_version)
latest_ver = version.parse(latest_version)
if current_ver < latest_ver:
self.status_label.setText(
_('update.update_available', current=current_version, latest=latest_version)
)
self.update_btn.setEnabled(True)
else:
self.status_label.setText(_('update.already_latest', version=current_version))
self.update_btn.setEnabled(False)
except version.InvalidVersion:
# If version parsing fails, do a simple string comparison
if current_version != latest_version:
self.status_label.setText(
_('update.update_available_failed', current=current_version, latest=latest_version)
)
self.update_btn.setEnabled(True)
else:
self.status_label.setText(_('update.up_to_date', version=current_version))
self.update_btn.setEnabled(False)
except Exception as e:
self.status_label.setText(_('update.error_comparing', error=e))
self.update_btn.setEnabled(False)
def perform_update(self) -> None:
# Immediate visual feedback
self.update_btn.setEnabled(False)
self.close_btn.setEnabled(False)
self.update_btn.setText(_('update.updating'))
self.status_label.setText(_('update.initializing'))
# Show progress bar immediately
self.progress_bar.setRange(0, 100)
self.progress_bar.setValue(0)
self.progress_bar.show()
# Start the update thread
self._start_update_thread()
def _start_update_thread(self) -> None:
"""Start the actual update thread."""
# Create and start the update thread
self.update_thread = UpdateThread()
self.update_thread.update_status.connect(self.on_update_status)
self.update_thread.update_progress.connect(self.on_update_progress)
self.update_thread.update_finished.connect(self.on_update_finished)
self.update_thread.start()
def on_update_status(self, message) -> None:
"""Slot to receive status messages from UpdateThread."""
if not (hasattr(self, "_closing") and self._closing):
self.status_label.setText(message)
def on_update_progress(self, progress) -> None:
"""Slot to receive progress updates from UpdateThread."""
if not (hasattr(self, "_closing") and self._closing):
self.progress_bar.setValue(progress)
def on_update_finished(self, success, message) -> None:
"""Slot called when the UpdateThread finishes."""
# Check if dialog is closing to avoid unnecessary updates
if hasattr(self, "_closing") and self._closing:
return
self.progress_bar.setValue(100)
self.status_label.setText(message)
self.close_btn.setEnabled(True)
self.update_btn.setText(_('buttons.update')) # Reset button text
if success:
# Show success briefly then auto-check version
QTimer.singleShot(2000, self.check_version) # Wait 2 seconds then refresh
else:
# Re-enable update button on failure after a short delay
QTimer.singleShot(
3000,
lambda: (self.update_btn.setEnabled(True) if not (hasattr(self, "_closing") and self._closing) else None),
)
def closeEvent(self, event) -> None:
"""Ensure threads are terminated if the dialog is closed prematurely."""
# Set a flag to indicate dialog is closing
self._closing = True
if hasattr(self, "version_check_thread") and self.version_check_thread.isRunning():
self.version_check_thread.quit()
if not self.version_check_thread.wait(3000): # Wait up to 3 seconds
self.version_check_thread.terminate()
if hasattr(self, "update_thread") and self.update_thread.isRunning():
self.update_thread.quit()
if not self.update_thread.wait(5000): # Wait up to 5 seconds for update to finish
self.update_thread.terminate()
super().closeEvent(event)
class AutoUpdateThread(QThread):
"""Thread for performing automatic background updates without UI feedback."""
update_finished = Signal(bool, str) # success (bool), message (str)
def run(self) -> None:
"""Perform automatic yt-dlp update check and update if needed."""
try:
logger.info("AutoUpdateThread: Performing automatic yt-dlp update check...")
# Get current version
current_version = get_ytdlp_version()
if "Error" in current_version:
logger.warning("AutoUpdateThread: Could not determine current yt-dlp version, skipping auto-update")
self.update_finished.emit(False, "Could not determine current yt-dlp version")
return
# Get latest version from PyPI
try:
response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10)
response.raise_for_status()
latest_version = response.json()["info"]["version"]
# Clean up version strings
current_version = current_version.replace("_", ".")
latest_version = latest_version.replace("_", ".")
logger.info(f"AutoUpdateThread: Current yt-dlp version: {current_version}")
logger.info(f"AutoUpdateThread: Latest yt-dlp version: {latest_version}")
# Compare versions
if version.parse(latest_version) > version.parse(current_version):
logger.info(f"AutoUpdateThread: Auto-updating yt-dlp from {current_version} to {latest_version}...")
# Perform the update
success = self._perform_update()
if success:
logger.info("AutoUpdateThread: Auto-update completed successfully!")
# Update the last check timestamp
ConfigManager.set("last_update_check", time.time())
self.update_finished.emit(
True,
f"Successfully updated yt-dlp from {current_version} to {latest_version}",
)
else:
logger.warning("AutoUpdateThread: Auto-update failed")
self.update_finished.emit(False, "Auto-update failed")
else:
logger.info("AutoUpdateThread: yt-dlp is already up to date")
# Still update the timestamp even if no update was needed
ConfigManager.set("last_update_check", time.time())
self.update_finished.emit(
True,
f"yt-dlp is already up to date (version {current_version})",
)
except requests.RequestException as e:
logger.warning(f"AutoUpdateThread: Network error during auto-update check: {e}")
self.update_finished.emit(False, f"Network error: {e}")
except Exception as e:
logger.exception(f"AutoUpdateThread: Error during auto-update check: {e}", exc_info=True)
self.update_finished.emit(False, f"Update check error: {e}")
except Exception as e:
logger.critical(f"AutoUpdateThread: Critical error in auto-update: {e}", exc_info=True)
self.update_finished.emit(False, f"Critical error: {e}")
def _perform_update(self) -> bool:
"""Perform the actual binary update."""
try:
# Get the yt-dlp path
yt_dlp_path = get_yt_dlp_path()
# Always update the binary (no more pip-based updates)
logger.info("AutoUpdateThread: Updating yt-dlp binary...")
return self._update_binary(yt_dlp_path)
except Exception as e:
logger.exception(f"AutoUpdateThread: Error in _perform_update: {e}")
return False
def _update_binary(self, yt_dlp_path: Path) -> bool:
"""Update yt-dlp binary using its built-in updater."""
try:
logger.info("AutoUpdateThread: Checking for yt-dlp updates...")
result = subprocess.run(
[yt_dlp_path, "-U"],
capture_output=True,
text=True,
timeout=60,
creationflags=SUBPROCESS_CREATIONFLAGS,
)
if result.returncode == 0:
# Make executable on Unix systems
if OS_NAME != "Windows":
os.chmod(yt_dlp_path, 0o755)
logger.info("AutoUpdateThread: yt-dlp update completed successfully.")
if result.stdout:
logger.debug(f"yt-dlp output: {result.stdout.strip()}")
return True
else:
logger.error(f"AutoUpdateThread: yt-dlp update failed. {result.stderr.strip()}")
return False
except subprocess.TimeoutExpired:
logger.error("AutoUpdateThread: yt-dlp update timed out.")
return False
except Exception as e:
logger.exception(f"AutoUpdateThread: Unexpected error during update: {e}")
return False
Reference in New Issue View Git Blame Copy Permalink