Refactor core module imports and update paths
Renamed 'src/core' to 'ytsage/core' and updated all internal imports to use relative paths. This improves package structure and ensures correct module resolution after the directory move.
This commit is contained in:
oop7
@@ -0,0 +1,5 @@
|
||||
"""
|
||||
Core functionality modules for YTSage.
|
||||
|
||||
This package contains the core business logic and utility functions.
|
||||
"""
|
||||
@@ -0,0 +1,758 @@
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import requests
|
||||
from PySide6.QtCore import Qt, QThread, Signal
|
||||
from PySide6.QtGui import QIcon
|
||||
from PySide6.QtWidgets import (
|
||||
QDialog,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QMessageBox,
|
||||
QProgressBar,
|
||||
QPushButton,
|
||||
QVBoxLayout,
|
||||
)
|
||||
|
||||
from ..utils.ytsage_logger import logger
|
||||
from ..utils.ytsage_localization import _
|
||||
from ..utils.ytsage_constants import (
|
||||
APP_BIN_DIR,
|
||||
ICON_PATH,
|
||||
OS_FULL_NAME,
|
||||
OS_NAME,
|
||||
SUBPROCESS_CREATIONFLAGS,
|
||||
DENO_APP_BIN_PATH,
|
||||
DENO_DOWNLOAD_URL,
|
||||
DENO_SHA256_URL,
|
||||
)
|
||||
from .ytsage_ffmpeg import get_file_sha256
|
||||
|
||||
|
||||
def verify_deno_sha256(file_path: Path, sha256_url: str) -> bool:
|
||||
"""
|
||||
Verify Deno file SHA256 hash against official checksums.
|
||||
|
||||
Args:
|
||||
file_path: Path to the downloaded Deno zip file
|
||||
sha256_url: URL to download the SHA256 checksum file
|
||||
|
||||
Returns:
|
||||
bool: True if verification successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
# Download the SHA256 checksum file
|
||||
logger.info(f"Downloading SHA256 checksum from: {sha256_url}")
|
||||
response = requests.get(sha256_url, timeout=10)
|
||||
response.raise_for_status()
|
||||
checksum_content = response.text
|
||||
|
||||
# Parse the checksum file
|
||||
# Format can be either:
|
||||
# 1. Standard Unix format: "hash filename"
|
||||
# 2. Verbose format: "Hash : <HASH>"
|
||||
expected_hash = None
|
||||
for line in checksum_content.strip().split("\n"):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
|
||||
# Try standard Unix format first (hash followed by spaces and filename)
|
||||
if len(line) >= 64 and (" " in line or "\t" in line):
|
||||
# Extract first 64 characters as potential hash
|
||||
potential_hash = line.split()[0]
|
||||
if len(potential_hash) == 64 and all(c in "0123456789abcdefABCDEF" for c in potential_hash):
|
||||
expected_hash = potential_hash
|
||||
break
|
||||
|
||||
# Try verbose format
|
||||
if line.startswith("Hash"):
|
||||
parts = line.split(":", 1)
|
||||
if len(parts) == 2:
|
||||
expected_hash = parts[1].strip()
|
||||
break
|
||||
|
||||
if not expected_hash:
|
||||
logger.error("Could not find SHA256 hash in checksum file")
|
||||
logger.debug(f"Checksum file content: {checksum_content}")
|
||||
return False
|
||||
|
||||
# Calculate actual hash of downloaded file
|
||||
logger.info("Calculating SHA256 hash of downloaded file...")
|
||||
actual_hash = get_file_sha256(file_path)
|
||||
|
||||
# Compare hashes (case-insensitive)
|
||||
if actual_hash.lower() == expected_hash.lower():
|
||||
logger.info("✓ SHA256 verification successful!")
|
||||
logger.info(f" Expected: {expected_hash}")
|
||||
logger.info(f" Actual: {actual_hash}")
|
||||
return True
|
||||
else:
|
||||
logger.error("✗ SHA256 verification failed!")
|
||||
logger.error(f" Expected: {expected_hash}")
|
||||
logger.error(f" Actual: {actual_hash}")
|
||||
return False
|
||||
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"Failed to download SHA256 checksum: {e}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.exception(f"Error during SHA256 verification: {e}")
|
||||
return False
|
||||
|
||||
|
||||
class DownloadDenoThread(QThread):
|
||||
progress_signal = Signal(int)
|
||||
status_signal = Signal(str)
|
||||
finished_signal = Signal(bool, str)
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
def run(self) -> None:
|
||||
temp_zip_path = None
|
||||
try:
|
||||
# Create temporary file for zip download
|
||||
temp_zip_fd, temp_zip_path = tempfile.mkstemp(suffix=".zip")
|
||||
os.close(temp_zip_fd) # Close the file descriptor
|
||||
|
||||
# Download with progress reporting
|
||||
logger.info(f"Downloading Deno from: {DENO_DOWNLOAD_URL}")
|
||||
self.status_signal.emit(_("deno.downloading"))
|
||||
|
||||
response = requests.get(DENO_DOWNLOAD_URL, stream=True)
|
||||
response.raise_for_status()
|
||||
total_size = int(response.headers.get("content-length", 0))
|
||||
block_size = 8192 # 8KB blocks
|
||||
|
||||
if total_size == 0:
|
||||
self.progress_signal.emit(100)
|
||||
|
||||
with open(temp_zip_path, "wb") as f:
|
||||
downloaded = 0
|
||||
for data in response.iter_content(block_size):
|
||||
f.write(data)
|
||||
downloaded += len(data)
|
||||
if total_size > 0:
|
||||
progress = int(downloaded / total_size * 100)
|
||||
self.progress_signal.emit(progress)
|
||||
|
||||
logger.info("Download complete, verifying SHA256 hash...")
|
||||
self.status_signal.emit(_("deno.verifying"))
|
||||
|
||||
# Verify SHA256 hash
|
||||
if not verify_deno_sha256(Path(temp_zip_path), DENO_SHA256_URL):
|
||||
# Hash verification failed - delete the downloaded file
|
||||
logger.error("SHA256 verification failed! Removing downloaded file.")
|
||||
if Path(temp_zip_path).exists():
|
||||
Path(temp_zip_path).unlink()
|
||||
self.finished_signal.emit(
|
||||
False,
|
||||
_("deno.verification_failed")
|
||||
)
|
||||
return
|
||||
|
||||
# Extract deno executable from zip
|
||||
logger.info("Extracting Deno executable...")
|
||||
self.status_signal.emit(_("deno.extracting"))
|
||||
|
||||
with zipfile.ZipFile(temp_zip_path, 'r') as zip_ref:
|
||||
# Deno zip contains just the executable at root
|
||||
executable_name = "deno.exe" if OS_NAME == "Windows" else "deno"
|
||||
|
||||
# Find the executable in the zip
|
||||
if executable_name not in zip_ref.namelist():
|
||||
logger.error(f"Executable '{executable_name}' not found in zip file")
|
||||
self.finished_signal.emit(False, f"Executable '{executable_name}' not found in zip")
|
||||
return
|
||||
|
||||
# Extract to app bin directory (while zip_ref is open)
|
||||
target_dir = DENO_APP_BIN_PATH.parent
|
||||
zip_ref.extract(executable_name, target_dir)
|
||||
|
||||
# Verify the extracted file exists
|
||||
exe_path = DENO_APP_BIN_PATH
|
||||
if not exe_path.exists():
|
||||
logger.error(f"Extraction failed: {exe_path} does not exist")
|
||||
self.finished_signal.emit(False, "Extraction failed")
|
||||
return
|
||||
|
||||
# Make executable on macOS and Linux
|
||||
if OS_NAME != "Windows":
|
||||
os.chmod(exe_path, 0o755)
|
||||
logger.info("Set executable permissions on Unix system")
|
||||
|
||||
# Clean up the temporary zip file
|
||||
if temp_zip_path and Path(temp_zip_path).exists():
|
||||
Path(temp_zip_path).unlink()
|
||||
logger.info("Cleaned up temporary zip file")
|
||||
|
||||
logger.info("Deno downloaded, verified, and extracted successfully!")
|
||||
self.finished_signal.emit(True, str(exe_path))
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error downloading/extracting Deno: {e}")
|
||||
# Clean up temporary zip file on error
|
||||
if temp_zip_path and Path(temp_zip_path).exists():
|
||||
try:
|
||||
Path(temp_zip_path).unlink()
|
||||
except Exception:
|
||||
pass
|
||||
self.finished_signal.emit(False, str(e))
|
||||
|
||||
|
||||
class DenoSetupDialog(QDialog):
|
||||
setup_complete = Signal(str) # Signal emitting the path to Deno
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle(_("deno.setup_required"))
|
||||
self.setMinimumWidth(520)
|
||||
self.setMinimumHeight(300)
|
||||
self.resize(520, 320)
|
||||
|
||||
# Set the window icon to match the main app
|
||||
if parent and parent.windowIcon():
|
||||
self.setWindowIcon(parent.windowIcon())
|
||||
else:
|
||||
icon_path = ICON_PATH
|
||||
if Path.exists(icon_path):
|
||||
self.setWindowIcon(QIcon(str(icon_path)))
|
||||
|
||||
self.init_ui()
|
||||
|
||||
# Apply dark theme styling to match app
|
||||
self.setStyleSheet(
|
||||
"""
|
||||
QDialog {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel {
|
||||
color: #cccccc;
|
||||
line-height: 1.4;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 10px 20px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 6px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
font-size: 13px;
|
||||
min-width: 120px;
|
||||
min-height: 20px;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
QPushButton:pressed {
|
||||
background-color: #800000;
|
||||
}
|
||||
QPushButton:disabled {
|
||||
background-color: #666666;
|
||||
color: #999999;
|
||||
}
|
||||
QProgressBar {
|
||||
border: 2px solid #1d1e22;
|
||||
border-radius: 6px;
|
||||
text-align: center;
|
||||
color: white;
|
||||
background-color: #1d1e22;
|
||||
height: 25px;
|
||||
font-weight: bold;
|
||||
}
|
||||
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;
|
||||
}
|
||||
"""
|
||||
)
|
||||
|
||||
def init_ui(self) -> None:
|
||||
layout = QVBoxLayout()
|
||||
layout.setSpacing(15)
|
||||
layout.setContentsMargins(25, 25, 25, 25)
|
||||
|
||||
# Header title
|
||||
title_label = QLabel(_("deno.setup_required"))
|
||||
title_label.setStyleSheet("font-size: 16px; font-weight: bold; color: #ffffff; padding: 5px 0;")
|
||||
title_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
layout.addWidget(title_label)
|
||||
|
||||
# Information label with improved styling
|
||||
info_label = QLabel(
|
||||
_("deno.setup_description", os_name=OS_FULL_NAME)
|
||||
)
|
||||
info_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
info_label.setWordWrap(True)
|
||||
info_label.setStyleSheet("font-size: 13px; color: #cccccc; padding: 5px; line-height: 1.4;")
|
||||
layout.addWidget(info_label)
|
||||
|
||||
# Progress bar with proper sizing
|
||||
self.progress_bar = QProgressBar()
|
||||
self.progress_bar.setVisible(False)
|
||||
self.progress_bar.setFixedHeight(20)
|
||||
self.progress_bar.setStyleSheet(
|
||||
"""
|
||||
QProgressBar {
|
||||
border: 1px solid #3d3d3d;
|
||||
border-radius: 8px;
|
||||
background-color: #1d1e22;
|
||||
text-align: center;
|
||||
color: #ffffff;
|
||||
font-size: 12px;
|
||||
font-weight: bold;
|
||||
height: 20px;
|
||||
}
|
||||
QProgressBar::chunk {
|
||||
background-color: #c90000;
|
||||
border-radius: 6px;
|
||||
margin: 1px;
|
||||
}
|
||||
"""
|
||||
)
|
||||
layout.addWidget(self.progress_bar)
|
||||
|
||||
# Status label with better spacing
|
||||
self.status_label = QLabel("")
|
||||
self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
self.status_label.setStyleSheet("font-size: 12px; color: #cccccc; padding: 8px 0;")
|
||||
self.status_label.setWordWrap(True)
|
||||
layout.addWidget(self.status_label)
|
||||
|
||||
# Add stretch to push buttons to bottom
|
||||
layout.addStretch()
|
||||
|
||||
# Button layout with improved spacing
|
||||
button_layout = QHBoxLayout()
|
||||
button_layout.setSpacing(15)
|
||||
button_layout.setContentsMargins(0, 10, 0, 0)
|
||||
|
||||
self.setup_button = QPushButton(_("deno.setup_button"))
|
||||
self.setup_button.clicked.connect(self.download_deno)
|
||||
|
||||
self.cancel_button = QPushButton(_("buttons.cancel"))
|
||||
self.cancel_button.clicked.connect(self.reject)
|
||||
|
||||
button_layout.addWidget(self.setup_button)
|
||||
button_layout.addWidget(self.cancel_button)
|
||||
layout.addLayout(button_layout)
|
||||
|
||||
self.setLayout(layout)
|
||||
|
||||
def download_deno(self) -> None:
|
||||
self.progress_bar.setVisible(True)
|
||||
self.progress_bar.setValue(0)
|
||||
self.status_label.setText(_("deno.downloading"))
|
||||
self.setup_button.setEnabled(False)
|
||||
self.cancel_button.setEnabled(False)
|
||||
|
||||
self.download_thread = DownloadDenoThread()
|
||||
self.download_thread.progress_signal.connect(self.update_progress)
|
||||
self.download_thread.status_signal.connect(self.update_status)
|
||||
self.download_thread.finished_signal.connect(self.download_finished)
|
||||
self.download_thread.start()
|
||||
|
||||
def update_progress(self, value) -> None:
|
||||
self.progress_bar.setValue(value)
|
||||
|
||||
def update_status(self, status: str) -> None:
|
||||
self.status_label.setText(status)
|
||||
|
||||
def download_finished(self, success, result) -> None:
|
||||
self.setup_button.setEnabled(True)
|
||||
self.cancel_button.setEnabled(True)
|
||||
|
||||
if success:
|
||||
self.status_label.setText(_("deno.success"))
|
||||
self.setup_complete.emit(result)
|
||||
self.accept()
|
||||
else:
|
||||
self.status_label.setText(f"{_('deno.download_error', error=result)}")
|
||||
error_dialog = QMessageBox(self)
|
||||
error_dialog.setIcon(QMessageBox.Icon.Critical)
|
||||
error_dialog.setWindowTitle(_("deno.download_failed"))
|
||||
error_dialog.setText(_("deno.download_error", error=result))
|
||||
error_dialog.setWindowIcon(self.windowIcon())
|
||||
error_dialog.setStyleSheet(
|
||||
"""
|
||||
QMessageBox {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel {
|
||||
color: #ffffff;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 8px 15px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
"""
|
||||
)
|
||||
error_dialog.exec()
|
||||
|
||||
|
||||
def check_deno_binary() -> Optional[Path]:
|
||||
"""
|
||||
Check if Deno binary exists in the app's bin directory ONLY.
|
||||
We only use our managed binary, not system PATH.
|
||||
|
||||
Returns:
|
||||
Path or None: Path to Deno binary if found in app bin, None otherwise
|
||||
"""
|
||||
exe_path = DENO_APP_BIN_PATH
|
||||
if exe_path.exists():
|
||||
# Make sure it's executable on Unix systems
|
||||
if OS_NAME != "Windows" and not os.access(exe_path, os.X_OK):
|
||||
try:
|
||||
os.chmod(exe_path, 0o755)
|
||||
logger.info(f"Fixed permissions on Deno at {exe_path}")
|
||||
except Exception as e:
|
||||
logger.exception(f"Could not set executable permissions on {exe_path}: {e}")
|
||||
logger.info(f"Found Deno in app bin directory: {exe_path}")
|
||||
return exe_path
|
||||
|
||||
# Binary not found in app directory - return None to trigger setup
|
||||
logger.warning(f"Deno binary not found in app bin directory: {exe_path}")
|
||||
return None
|
||||
|
||||
|
||||
def check_deno_installed() -> bool:
|
||||
"""
|
||||
Check if Deno is installed and accessible.
|
||||
|
||||
Returns:
|
||||
bool: True if Deno is found and working, False otherwise
|
||||
"""
|
||||
try:
|
||||
deno_path = check_deno_binary()
|
||||
if deno_path:
|
||||
# Try to run deno --version to verify it's working
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[str(deno_path), "--version"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=5,
|
||||
creationflags=SUBPROCESS_CREATIONFLAGS
|
||||
)
|
||||
return result.returncode == 0
|
||||
except Exception:
|
||||
return False
|
||||
return False
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def get_deno_path() -> Path:
|
||||
"""
|
||||
Get the Deno path from the app's bin directory.
|
||||
|
||||
Returns:
|
||||
Path or str: Path to Deno binary, or "deno" as fallback command
|
||||
"""
|
||||
deno_path = check_deno_binary()
|
||||
if deno_path:
|
||||
logger.info(f"Using Deno from: {deno_path}")
|
||||
return deno_path
|
||||
|
||||
# If not found, fall back to the command name as a last resort
|
||||
logger.info("Deno not found in app directory, falling back to command name")
|
||||
return "deno" # type: ignore[return-value]
|
||||
|
||||
|
||||
def get_deno_version_direct(deno_path=None) -> str:
|
||||
"""
|
||||
Get Deno version directly without caching.
|
||||
|
||||
Args:
|
||||
deno_path: Optional path to Deno binary. If None, uses get_deno_path()
|
||||
|
||||
Returns:
|
||||
str: Version string or error message
|
||||
"""
|
||||
try:
|
||||
if deno_path is None:
|
||||
deno_path = get_deno_path()
|
||||
|
||||
if not deno_path or deno_path == "deno":
|
||||
return "Not found"
|
||||
|
||||
result = subprocess.run(
|
||||
[str(deno_path), "--version"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
creationflags=SUBPROCESS_CREATIONFLAGS
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
# Deno outputs: "deno 1.38.0 (release, x86_64-pc-windows-msvc)"
|
||||
# Extract version from first line
|
||||
lines = result.stdout.strip().split("\n")
|
||||
if lines:
|
||||
first_line = lines[0]
|
||||
# Extract version number (e.g., "1.38.0" from "deno 1.38.0 ...")
|
||||
parts = first_line.split()
|
||||
if len(parts) >= 2 and parts[0] == "deno":
|
||||
return parts[1]
|
||||
return first_line.strip()
|
||||
return "Unknown version"
|
||||
else:
|
||||
return "Error getting version"
|
||||
except Exception as e:
|
||||
logger.exception(f"Error getting Deno version: {e}")
|
||||
return "Error getting version"
|
||||
|
||||
|
||||
def setup_deno(parent_widget=None):
|
||||
"""
|
||||
Show the Deno setup dialog and handle the result.
|
||||
|
||||
Returns:
|
||||
str: Path to Deno binary
|
||||
"""
|
||||
logger.debug("Starting Deno setup dialog")
|
||||
dialog = DenoSetupDialog(parent_widget)
|
||||
|
||||
# Store the setup result from the signal
|
||||
setup_result = {"path": None}
|
||||
|
||||
def on_setup_complete(path) -> None:
|
||||
logger.debug(f"Received setup_complete signal with path: {path}")
|
||||
setup_result["path"] = path
|
||||
|
||||
# Connect to the setup_complete signal
|
||||
dialog.setup_complete.connect(on_setup_complete)
|
||||
|
||||
# Show the dialog
|
||||
result = dialog.exec()
|
||||
logger.debug(f"Dialog result: {result} (Accepted={QDialog.DialogCode.Accepted})")
|
||||
|
||||
if result == QDialog.DialogCode.Accepted:
|
||||
# First check if we received a path from the signal
|
||||
if setup_result["path"]:
|
||||
path_obj = Path(setup_result["path"]) if isinstance(setup_result["path"], str) else setup_result["path"]
|
||||
if path_obj.exists():
|
||||
logger.debug(f"Using path from signal: {setup_result['path']}")
|
||||
return str(setup_result["path"])
|
||||
|
||||
# Get the expected path for verification as fallback
|
||||
expected_path = DENO_APP_BIN_PATH
|
||||
logger.debug(f"Expected Deno path: {expected_path}")
|
||||
|
||||
# Verify the path exists after dialog is accepted
|
||||
if expected_path.exists():
|
||||
logger.debug(f"Deno successfully found at expected path: {expected_path}")
|
||||
return str(expected_path)
|
||||
else:
|
||||
logger.debug(f"Expected path does not exist, trying alternate detection")
|
||||
# Try to use the get_deno_path function to find Deno elsewhere
|
||||
deno_path = get_deno_path()
|
||||
logger.debug(f"Alternate detection result: {deno_path}")
|
||||
if deno_path != "deno":
|
||||
path_obj = Path(deno_path) if isinstance(deno_path, str) else deno_path
|
||||
if path_obj.exists():
|
||||
logger.debug(f"Deno found at alternate location: {deno_path}")
|
||||
return str(deno_path)
|
||||
|
||||
# Something went wrong, show an error message
|
||||
logger.debug(f"Setup failed, showing error dialog")
|
||||
if parent_widget:
|
||||
error_dialog = QMessageBox(parent_widget)
|
||||
error_dialog.setIcon(QMessageBox.Icon.Warning)
|
||||
error_dialog.setWindowTitle(_("deno.setup_error"))
|
||||
error_dialog.setText(_("deno.setup_failed"))
|
||||
error_dialog.setWindowIcon(parent_widget.windowIcon())
|
||||
error_dialog.setStyleSheet(
|
||||
"""
|
||||
QMessageBox {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel {
|
||||
color: #ffffff;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 8px 15px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
"""
|
||||
)
|
||||
error_dialog.exec()
|
||||
logger.warning(f"Deno setup failed, path does not exist: {expected_path}")
|
||||
else:
|
||||
logger.debug("User cancelled the setup dialog")
|
||||
|
||||
# User cancelled or setup failed, return the fallback command
|
||||
logger.debug("Returning fallback command 'deno'")
|
||||
return "deno"
|
||||
|
||||
|
||||
def get_latest_deno_version() -> Optional[str]:
|
||||
"""
|
||||
Fetch the latest Deno version from GitHub API.
|
||||
|
||||
Returns:
|
||||
str: Version string (e.g., "2.5.6") or None if fetch failed
|
||||
"""
|
||||
try:
|
||||
response = requests.get(
|
||||
"https://api.github.com/repos/denoland/deno/releases/latest",
|
||||
timeout=10
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
# Get tag_name (e.g., "v2.5.6") and remove 'v' prefix
|
||||
tag_name = data.get("tag_name", "")
|
||||
if tag_name.startswith("v"):
|
||||
version = tag_name[1:]
|
||||
else:
|
||||
version = tag_name
|
||||
|
||||
logger.info(f"Latest Deno version: {version}")
|
||||
return version
|
||||
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"Failed to fetch latest Deno version: {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.exception(f"Unexpected error fetching Deno version: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def compare_deno_versions(current: str, latest: str) -> bool:
|
||||
"""
|
||||
Compare two Deno version strings.
|
||||
|
||||
Args:
|
||||
current: Current version string (e.g., "2.5.6")
|
||||
latest: Latest version string (e.g., "2.5.7")
|
||||
|
||||
Returns:
|
||||
bool: True if update is needed (latest > current), False otherwise
|
||||
"""
|
||||
try:
|
||||
import re
|
||||
|
||||
def parse_version(version_str: str) -> tuple:
|
||||
"""Parse version string into tuple of integers."""
|
||||
# Remove 'v' prefix if present
|
||||
if version_str.startswith('v'):
|
||||
version_str = version_str[1:]
|
||||
|
||||
# Extract version numbers
|
||||
match = re.search(r'(\d+\.\d+\.\d+)', version_str)
|
||||
if match:
|
||||
version_str = match.group(1)
|
||||
|
||||
parts = version_str.split('.')
|
||||
return tuple(int(p) for p in parts if p.isdigit())
|
||||
|
||||
current_tuple = parse_version(current)
|
||||
latest_tuple = parse_version(latest)
|
||||
|
||||
logger.debug(f"Comparing Deno versions: {current_tuple} vs {latest_tuple}")
|
||||
return latest_tuple > current_tuple
|
||||
|
||||
except (ValueError, AttributeError) as e:
|
||||
logger.warning(f"Could not compare Deno versions: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def upgrade_deno() -> tuple[bool, str]:
|
||||
"""
|
||||
Upgrade Deno to the latest version using 'deno upgrade' command.
|
||||
|
||||
Returns:
|
||||
tuple: (success: bool, output: str) - Success status and command output
|
||||
"""
|
||||
try:
|
||||
deno_path = DENO_APP_BIN_PATH
|
||||
|
||||
if not deno_path.exists():
|
||||
error_msg = f"Deno binary not found at: {deno_path}"
|
||||
logger.error(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
logger.info(f"Upgrading Deno using: {deno_path}")
|
||||
|
||||
# Run deno upgrade command
|
||||
result = subprocess.run(
|
||||
[str(deno_path), "upgrade"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=300, # 5 minutes timeout
|
||||
creationflags=SUBPROCESS_CREATIONFLAGS
|
||||
)
|
||||
|
||||
output = result.stdout + result.stderr
|
||||
|
||||
if result.returncode == 0:
|
||||
logger.info("Deno upgrade successful")
|
||||
logger.debug(f"Upgrade output: {output}")
|
||||
return True, output
|
||||
else:
|
||||
logger.error(f"Deno upgrade failed with code {result.returncode}")
|
||||
logger.error(f"Output: {output}")
|
||||
return False, output
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
error_msg = "Deno upgrade timed out after 5 minutes"
|
||||
logger.error(error_msg)
|
||||
return False, error_msg
|
||||
except Exception as e:
|
||||
error_msg = f"Error upgrading Deno: {str(e)}"
|
||||
logger.exception(error_msg)
|
||||
return False, error_msg
|
||||
|
||||
|
||||
def check_deno_update() -> tuple[bool, str, str]:
|
||||
"""
|
||||
Check if a Deno update is available.
|
||||
|
||||
Returns:
|
||||
tuple: (update_needed: bool, current_version: str, latest_version: str)
|
||||
"""
|
||||
try:
|
||||
# Get current version
|
||||
current_version = get_deno_version_direct()
|
||||
if current_version in ["Not found", "Error getting version"]:
|
||||
return False, current_version, "Unknown"
|
||||
|
||||
# Get latest version
|
||||
latest_version = get_latest_deno_version()
|
||||
if not latest_version:
|
||||
return False, current_version, "Error"
|
||||
|
||||
# Compare versions
|
||||
update_needed = compare_deno_versions(current_version, latest_version)
|
||||
|
||||
return update_needed, current_version, latest_version
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error checking Deno update: {e}")
|
||||
return False, "Error", "Error"
|
||||
@@ -0,0 +1,761 @@
|
||||
import gc
|
||||
import os
|
||||
import re
|
||||
import shlex # For safely parsing command arguments
|
||||
import signal
|
||||
import subprocess # For direct CLI command execution
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Optional, List, Set
|
||||
|
||||
from PySide6.QtCore import QObject, QThread, Signal
|
||||
|
||||
from .ytsage_yt_dlp import get_yt_dlp_path
|
||||
from ..utils.ytsage_constants import (
|
||||
SUBPROCESS_CREATIONFLAGS,
|
||||
VIDEO_EXTENSIONS,
|
||||
AUDIO_EXTENSIONS,
|
||||
SUBTITLE_EXTENSIONS,
|
||||
MEDIA_EXTENSIONS,
|
||||
)
|
||||
from ..utils.ytsage_localization import LocalizationManager
|
||||
from ..utils.ytsage_logger import logger
|
||||
|
||||
# Shorthand for localization
|
||||
_ = LocalizationManager.get_text
|
||||
|
||||
|
||||
class SignalManager(QObject):
|
||||
update_formats = Signal(list)
|
||||
update_status = Signal(str)
|
||||
update_progress = Signal(float)
|
||||
playlist_info_label_visible = Signal(bool)
|
||||
playlist_info_label_text = Signal(str)
|
||||
selected_subs_label_text = Signal(str)
|
||||
playlist_select_btn_visible = Signal(bool)
|
||||
playlist_select_btn_text = Signal(str)
|
||||
|
||||
|
||||
class DownloadThread(QThread):
|
||||
progress_signal = Signal(float)
|
||||
status_signal = Signal(str)
|
||||
finished_signal = Signal()
|
||||
error_signal = Signal(str)
|
||||
file_exists_signal = Signal(str) # New signal for file existence
|
||||
update_details = Signal(str) # New signal for filename, speed, ETA
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
url,
|
||||
path,
|
||||
format_id,
|
||||
is_audio_only=False,
|
||||
format_has_audio=False,
|
||||
subtitle_langs=None,
|
||||
is_playlist=False,
|
||||
merge_subs=False,
|
||||
enable_sponsorblock=False,
|
||||
sponsorblock_categories=None,
|
||||
resolution="",
|
||||
playlist_items=None,
|
||||
save_description=False,
|
||||
embed_chapters=False,
|
||||
cookie_file=None,
|
||||
browser_cookies=None,
|
||||
rate_limit=None,
|
||||
download_section=None,
|
||||
force_keyframes=False,
|
||||
proxy_url=None,
|
||||
geo_proxy_url=None,
|
||||
force_output_format=False,
|
||||
preferred_output_format="mp4",
|
||||
force_audio_format=False,
|
||||
preferred_audio_format="best",
|
||||
) -> None:
|
||||
super().__init__()
|
||||
self.url = url
|
||||
self.path = Path(path)
|
||||
self.format_id = format_id
|
||||
self.is_audio_only = is_audio_only
|
||||
self.format_has_audio = format_has_audio
|
||||
self.subtitle_langs = subtitle_langs if subtitle_langs else []
|
||||
self.is_playlist = is_playlist
|
||||
self.merge_subs = merge_subs
|
||||
self.enable_sponsorblock = enable_sponsorblock
|
||||
self.sponsorblock_categories = sponsorblock_categories if sponsorblock_categories else ["sponsor"]
|
||||
self.resolution = resolution
|
||||
self.playlist_items = playlist_items
|
||||
self.save_description = save_description
|
||||
self.embed_chapters = embed_chapters
|
||||
self.cookie_file = cookie_file
|
||||
self.browser_cookies = browser_cookies
|
||||
self.rate_limit = rate_limit
|
||||
self.download_section = download_section
|
||||
self.force_keyframes = force_keyframes
|
||||
self.proxy_url = proxy_url
|
||||
self.geo_proxy_url = geo_proxy_url
|
||||
self.force_output_format = force_output_format
|
||||
self.preferred_output_format = preferred_output_format
|
||||
self.force_audio_format = force_audio_format
|
||||
self.preferred_audio_format = preferred_audio_format
|
||||
self.paused: bool = False
|
||||
self.cancelled: bool = False
|
||||
self.process: Optional[subprocess.Popen] = None
|
||||
self.current_filename: Optional[str] = None # Initialize filename storage
|
||||
self.last_file_path: Optional[str] = None # Initialize full file path storage
|
||||
self.subtitle_files: List[str] = [] # Track subtitle files that are created
|
||||
self.initial_subtitle_files: Set[Path] = set() # Track initial subtitle files before download
|
||||
|
||||
def cleanup_partial_files(self) -> None:
|
||||
"""Delete any partial files including .part and unmerged format-specific files"""
|
||||
try:
|
||||
pattern = re.compile(r"\.f\d+\.") # Pattern to match format codes like .f243.
|
||||
for file_path in self.path.iterdir():
|
||||
if file_path.suffix == ".part" or pattern.search(file_path.name):
|
||||
self._safe_delete_with_retry(file_path)
|
||||
except Exception as e:
|
||||
logger.exception(f"Error cleaning partial files: {e}")
|
||||
# Don't emit error signal for cleanup issues to avoid crashing the thread
|
||||
logger.error(f"Error cleaning partial files: {e}")
|
||||
|
||||
def _safe_delete_with_retry(self, file_path: Path, max_retries: int = 5, delay: float = 2.0) -> None:
|
||||
"""Safely delete a file with retry mechanism for file locking issues across platforms"""
|
||||
for attempt in range(max_retries):
|
||||
try:
|
||||
# Force garbage collection to release any Python-held file handles
|
||||
gc.collect()
|
||||
|
||||
if file_path.exists():
|
||||
file_path.unlink(missing_ok=True)
|
||||
logger.info(f"Successfully deleted {file_path.name}")
|
||||
return
|
||||
except PermissionError as e:
|
||||
if attempt < max_retries - 1:
|
||||
logger.warning(f"File {file_path.name} is locked, retrying in {delay} seconds... (attempt {attempt + 1}/{max_retries})")
|
||||
time.sleep(delay)
|
||||
delay = min(delay * 1.5, 5.0) # Exponential backoff, capped at 5 seconds
|
||||
else:
|
||||
logger.error(f"Failed to delete {file_path.name} after {max_retries} attempts: {e}")
|
||||
return
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting {file_path.name}: {e}")
|
||||
return
|
||||
|
||||
def _terminate_process_tree(self, process: subprocess.Popen) -> None:
|
||||
"""Terminate a process and all its children across platforms"""
|
||||
pid = process.pid
|
||||
|
||||
try:
|
||||
if sys.platform == "win32":
|
||||
# Windows: Use taskkill to kill the entire process tree
|
||||
# /T = kill child processes, /F = force kill
|
||||
# Use subprocess.run with no encoding to avoid codec issues
|
||||
subprocess.run(
|
||||
["taskkill", "/F", "/T", "/PID", str(pid)],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
creationflags=SUBPROCESS_CREATIONFLAGS,
|
||||
)
|
||||
logger.debug(f"Killed process tree on Windows (PID: {pid})")
|
||||
else:
|
||||
# Unix-like systems: Kill the process group
|
||||
try:
|
||||
# Try to kill the process group
|
||||
os.killpg(os.getpgid(pid), signal.SIGTERM)
|
||||
time.sleep(0.5)
|
||||
# Force kill if still running
|
||||
os.killpg(os.getpgid(pid), signal.SIGKILL)
|
||||
except (ProcessLookupError, PermissionError):
|
||||
# Process already terminated or no permission
|
||||
pass
|
||||
logger.debug(f"Killed process group on Unix (PID: {pid})")
|
||||
except Exception as e:
|
||||
logger.warning(f"Error killing process tree: {e}")
|
||||
# Fallback to standard termination
|
||||
try:
|
||||
process.terminate()
|
||||
process.wait(timeout=2)
|
||||
except Exception:
|
||||
try:
|
||||
process.kill()
|
||||
process.wait()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Ensure process is waited on to avoid zombies
|
||||
try:
|
||||
process.wait(timeout=3)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def cleanup_subtitle_files(self) -> None:
|
||||
"""Delete subtitle files after they have been merged into the video file"""
|
||||
deleted_count: List[int] = [0, 0]
|
||||
|
||||
def safe_delete(path: Path) -> bool:
|
||||
try:
|
||||
# Check if file exists before trying to delete
|
||||
if path.exists():
|
||||
path.unlink(missing_ok=True)
|
||||
logger.debug(f"Deleted subtitle file: {path.name}")
|
||||
return True
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.exception(f"Error deleting subtitle file {path}: {e}")
|
||||
return False
|
||||
|
||||
try:
|
||||
# --- Method 1: Delete tracked subtitle files ---
|
||||
for f in self.subtitle_files or []:
|
||||
deleted_count[0] += safe_delete(path=Path(f))
|
||||
else:
|
||||
logger.debug(f"Deleted {deleted_count[0]} of {len(self.subtitle_files)} tracked subtitle files")
|
||||
|
||||
# --- Method 2: Delete new subtitle files not in initial set ---
|
||||
new_subtitle_files: Set[Path] = {
|
||||
f for f in Path(self.path).rglob("*") if f.suffix in [".vtt", ".srt"] and f not in self.initial_subtitle_files
|
||||
}
|
||||
for subtitle_file in new_subtitle_files:
|
||||
deleted_count[1] += safe_delete(path=subtitle_file)
|
||||
else:
|
||||
logger.debug(f"Deleted {deleted_count[1]} of {len(new_subtitle_files)} new subtitle files")
|
||||
except Exception as e:
|
||||
logger.exception(f"Error cleaning subtitle files: {e}")
|
||||
|
||||
def _build_yt_dlp_command(self) -> List[str]:
|
||||
"""Build the yt-dlp command line with all options for direct execution."""
|
||||
# Use the new yt-dlp path function from ytsage_yt_dlp module
|
||||
yt_dlp_path: str = get_yt_dlp_path()
|
||||
cmd: List[str] = [yt_dlp_path]
|
||||
logger.debug(f"Using yt-dlp from: {yt_dlp_path}")
|
||||
|
||||
# Format selection strategy - use format ID if provided or fallback to resolution
|
||||
if self.format_id:
|
||||
clean_format_id: str = self.format_id.split("-drc")[0] if "-drc" in self.format_id else self.format_id
|
||||
|
||||
# If the selected format is audio-only, pass it directly.
|
||||
if self.is_audio_only:
|
||||
cmd.extend(["-f", clean_format_id])
|
||||
logger.debug(f"Using audio-only format selection: {clean_format_id}")
|
||||
# If the selected format already includes an audio track (progressive), no merge needed.
|
||||
elif self.format_has_audio:
|
||||
cmd.extend(["-f", clean_format_id])
|
||||
logger.debug(f"Using progressive format with bundled audio: {clean_format_id}")
|
||||
else:
|
||||
cmd.extend(["-f", f"{clean_format_id}+bestaudio/best"])
|
||||
logger.debug(f"Using video-only format merged with best audio: {clean_format_id}+bestaudio/best")
|
||||
else:
|
||||
# If no specific format ID, use resolution-based sorting (-S)
|
||||
res_value: str = self.resolution if self.resolution else "720" # Default to 720p if no resolution specified
|
||||
cmd.extend(["-S", f"res:{res_value}"])
|
||||
|
||||
# Force output format if enabled and merging is needed (for video)
|
||||
if self.force_output_format and not self.is_audio_only:
|
||||
if self.format_has_audio:
|
||||
# Progressive format (video with audio) - use remux to convert container
|
||||
cmd.extend(["--remux-video", self.preferred_output_format])
|
||||
logger.debug(f"Using --remux-video to force progressive format to: {self.preferred_output_format}")
|
||||
else:
|
||||
# Merging video+audio - force merge output format
|
||||
cmd.extend(["--merge-output-format", self.preferred_output_format])
|
||||
logger.debug(f"Using --merge-output-format to force merged format to: {self.preferred_output_format}")
|
||||
|
||||
# Force audio format conversion for audio-only downloads
|
||||
if self.is_audio_only and self.force_audio_format:
|
||||
cmd.append("--extract-audio")
|
||||
if self.preferred_audio_format and self.preferred_audio_format != "best":
|
||||
cmd.extend(["--audio-format", self.preferred_audio_format])
|
||||
logger.debug(f"Using --extract-audio with --audio-format {self.preferred_audio_format} for audio-only download")
|
||||
else:
|
||||
logger.debug("Using --extract-audio with best quality (no conversion) for audio-only download")
|
||||
|
||||
# Output template with resolution in filename
|
||||
# Use string concatenation instead of Path.joinpath to avoid Path object issues
|
||||
base_path: str = self.path.as_posix()
|
||||
|
||||
if self.is_playlist:
|
||||
# Create output template with playlist subfolder
|
||||
output_template: str = f"{base_path}/%(playlist_title)s/%(title)s_%(resolution)s.%(ext)s"
|
||||
else:
|
||||
output_template: str = f"{base_path}/%(title)s_%(resolution)s.%(ext)s"
|
||||
|
||||
cmd.extend(["-o", str(output_template)])
|
||||
|
||||
# Add common options
|
||||
cmd.append("--force-overwrites")
|
||||
|
||||
# Add playlist items if specified
|
||||
if self.is_playlist and self.playlist_items:
|
||||
cmd.extend(["--playlist-items", self.playlist_items])
|
||||
|
||||
# Add subtitle options if subtitles are selected
|
||||
if self.subtitle_langs:
|
||||
# Subtitles work with both audio-only and video formats
|
||||
# For audio-only formats, subtitles will be downloaded as separate files
|
||||
cmd.append("--write-subs")
|
||||
|
||||
# Get language codes from subtitle selections
|
||||
lang_codes: List[str] = []
|
||||
for sub_selection in self.subtitle_langs:
|
||||
try:
|
||||
# Extract just the language code (e.g., 'en' from 'en - Manual')
|
||||
lang_code = sub_selection.split(" - ")[0]
|
||||
lang_codes.append(lang_code)
|
||||
except Exception as e:
|
||||
logger.exception(f"Could not parse subtitle selection '{sub_selection}': {e}")
|
||||
|
||||
if lang_codes:
|
||||
cmd.extend(["--sub-langs", ",".join(lang_codes)])
|
||||
cmd.append("--write-auto-subs") # Include auto-generated subtitles
|
||||
|
||||
# Only embed subtitles if merge is enabled
|
||||
if self.merge_subs:
|
||||
cmd.append("--embed-subs")
|
||||
|
||||
# Add SponsorBlock if enabled
|
||||
if self.enable_sponsorblock and self.sponsorblock_categories:
|
||||
cmd.append("--sponsorblock-remove")
|
||||
cmd.append(",".join(self.sponsorblock_categories))
|
||||
|
||||
# Add description saving if enabled
|
||||
if self.save_description:
|
||||
cmd.append("--write-description")
|
||||
|
||||
# Add chapters embedding if enabled
|
||||
if self.embed_chapters:
|
||||
cmd.append("--embed-chapters")
|
||||
|
||||
# Add cookies if specified
|
||||
if self.cookie_file:
|
||||
cmd.extend(["--cookies", str(self.cookie_file)])
|
||||
elif self.browser_cookies:
|
||||
cmd.extend(["--cookies-from-browser", self.browser_cookies])
|
||||
|
||||
# Add proxy settings if specified
|
||||
if self.proxy_url:
|
||||
cmd.extend(["--proxy", self.proxy_url])
|
||||
|
||||
if self.geo_proxy_url:
|
||||
cmd.extend(["--geo-verification-proxy", self.geo_proxy_url])
|
||||
|
||||
# Add rate limit if specified
|
||||
if self.rate_limit:
|
||||
cmd.extend(["-r", self.rate_limit])
|
||||
|
||||
# Add download section if specified
|
||||
if self.download_section:
|
||||
cmd.extend(["--download-sections", self.download_section])
|
||||
|
||||
# Add force keyframes option if enabled
|
||||
if self.force_keyframes:
|
||||
cmd.append("--force-keyframes-at-cuts")
|
||||
|
||||
logger.debug(f"Added download section: {self.download_section}, Force keyframes: {self.force_keyframes}")
|
||||
|
||||
# Add the URL as the final argument
|
||||
cmd.append(self.url)
|
||||
|
||||
return cmd
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
logger.debug("Starting download thread")
|
||||
|
||||
# Get initial list of subtitle files to compare later
|
||||
self.initial_subtitle_files = set()
|
||||
if self.merge_subs:
|
||||
try:
|
||||
# Scan for existing subtitle files in the directory
|
||||
for file in self.path.rglob("*"):
|
||||
if file.suffix in {".vtt", ".srt"}:
|
||||
self.initial_subtitle_files.add(file)
|
||||
logger.debug(f"Found {len(self.initial_subtitle_files)} existing subtitle files before download")
|
||||
except Exception as e:
|
||||
logger.exception(f"Error scanning for initial subtitle files: {e}")
|
||||
|
||||
# Use direct CLI command
|
||||
self._run_direct_command()
|
||||
|
||||
except Exception as e:
|
||||
# Catch errors during setup
|
||||
logger.critical(f"Critical error in download thread: {e}", exc_info=True)
|
||||
self.error_signal.emit(f"Critical error in download thread: {e}")
|
||||
|
||||
def _run_direct_command(self) -> None:
|
||||
"""Run yt-dlp as a direct command line process instead of using Python API."""
|
||||
try:
|
||||
cmd: List[str] = self._build_yt_dlp_command()
|
||||
cmd_str: str = " ".join(shlex.quote(str(arg)) for arg in cmd)
|
||||
logger.debug(f"Executing command: {cmd_str}")
|
||||
|
||||
self.status_signal.emit(_("download.starting"))
|
||||
self.progress_signal.emit(0)
|
||||
|
||||
# Start the process
|
||||
# Extra logic moved to src\utils\ytsage_constants.py
|
||||
# Use start_new_session on Unix to enable process group termination
|
||||
|
||||
popen_kwargs = {
|
||||
"stdout": subprocess.PIPE,
|
||||
"stderr": subprocess.STDOUT,
|
||||
"bufsize": 1, # Line buffered
|
||||
"encoding": "utf-8",
|
||||
"errors": "replace",
|
||||
}
|
||||
|
||||
if sys.platform == "win32":
|
||||
popen_kwargs["creationflags"] = SUBPROCESS_CREATIONFLAGS
|
||||
else:
|
||||
# On Unix, start a new session so we can kill the entire process group
|
||||
popen_kwargs["start_new_session"] = True
|
||||
|
||||
self.process = subprocess.Popen(cmd, **popen_kwargs)
|
||||
|
||||
# Process output line by line to update progress
|
||||
for line in iter(self.process.stdout.readline, ""): # type: ignore
|
||||
if self.cancelled:
|
||||
# Kill the entire process tree (yt-dlp + ffmpeg children)
|
||||
self._terminate_process_tree(self.process)
|
||||
|
||||
# Add delay before cleanup to allow file handles to be released
|
||||
time.sleep(2)
|
||||
self.cleanup_partial_files()
|
||||
self.status_signal.emit(_("download.cancelled"))
|
||||
return
|
||||
|
||||
# Wait if paused
|
||||
while self.paused and not self.cancelled:
|
||||
time.sleep(0.1)
|
||||
|
||||
# Parse the line for download progress and status updates
|
||||
self._parse_output_line(line)
|
||||
|
||||
# Wait for process to complete
|
||||
return_code: int = self.process.wait()
|
||||
|
||||
# Special handling for specific errors
|
||||
# return code 127 typically means command not found
|
||||
if return_code == 127:
|
||||
self.error_signal.emit(
|
||||
_("errors.ytdlp_not_found_path")
|
||||
)
|
||||
return
|
||||
|
||||
if return_code == 0:
|
||||
self.progress_signal.emit(100)
|
||||
|
||||
# Robust file finding: Always search for the most recent file
|
||||
# This handles all post-processing scenarios (merging, remuxing, subtitle embedding, etc.)
|
||||
final_file_found = False
|
||||
|
||||
try:
|
||||
# First, check if last_file_path exists and is valid
|
||||
if self.last_file_path:
|
||||
last_path = Path(self.last_file_path)
|
||||
if last_path.exists() and last_path.is_file():
|
||||
# File exists at the tracked path
|
||||
self.current_filename = last_path.name
|
||||
final_file_found = True
|
||||
logger.info(f"Found file at tracked path: {self.last_file_path}")
|
||||
|
||||
# If not found at tracked path, search for the most recent file
|
||||
if not final_file_found:
|
||||
logger.info("Searching for most recent downloaded file...")
|
||||
potential_files = []
|
||||
|
||||
# Search in download directory and subdirectories (for playlists)
|
||||
for ext in MEDIA_EXTENSIONS:
|
||||
potential_files.extend(self.path.glob(f'*{ext}'))
|
||||
# Also check subdirectories (for playlist downloads)
|
||||
potential_files.extend(self.path.glob(f'*/*{ext}'))
|
||||
|
||||
if potential_files:
|
||||
# Sort by modification time and get the most recent
|
||||
most_recent = max(potential_files, key=lambda p: p.stat().st_mtime)
|
||||
|
||||
# Verify it was modified recently (within last 30 seconds to account for post-processing)
|
||||
time_since_modification = time.time() - most_recent.stat().st_mtime
|
||||
|
||||
if time_since_modification < 30:
|
||||
self.last_file_path = str(most_recent)
|
||||
self.current_filename = most_recent.name
|
||||
final_file_found = True
|
||||
logger.info(f"Found most recent file (modified {time_since_modification:.1f}s ago): {self.last_file_path}")
|
||||
else:
|
||||
logger.warning(f"Most recent file is too old ({time_since_modification:.1f}s), might not be the right one")
|
||||
else:
|
||||
logger.warning("No video/audio files found in download directory")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error finding final file: {e}", exc_info=True)
|
||||
|
||||
# Set completion status
|
||||
self.status_signal.emit(_("download.completed"))
|
||||
|
||||
# Clean up subtitle files if they were merged, with a small delay
|
||||
# to ensure the embedding process has completed
|
||||
if self.merge_subs:
|
||||
# Add a significant delay to ensure ffmpeg has released all file handles
|
||||
# and any post-processing is complete
|
||||
self.status_signal.emit(_("download.completed_cleaning"))
|
||||
time.sleep(3) # Increased delay to 3 seconds
|
||||
self.cleanup_subtitle_files()
|
||||
|
||||
self.finished_signal.emit()
|
||||
else:
|
||||
# Check if it was cancelled
|
||||
if self.cancelled:
|
||||
self.status_signal.emit(_("download.cancelled"))
|
||||
else:
|
||||
# Provide more descriptive error message for possible yt-dlp conflicts
|
||||
if return_code == 1:
|
||||
self.error_signal.emit(
|
||||
_("errors.download_failed_return_code_conflict", return_code=return_code)
|
||||
)
|
||||
else:
|
||||
self.error_signal.emit(
|
||||
_("errors.download_failed_return_code", return_code=return_code)
|
||||
)
|
||||
|
||||
# Add delay before cleanup to allow file handles to be released
|
||||
time.sleep(1)
|
||||
self.cleanup_partial_files()
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error in direct command: {e}")
|
||||
self.error_signal.emit(_("errors.direct_command_error", error=str(e)))
|
||||
# Add delay before cleanup to allow file handles to be released
|
||||
time.sleep(1)
|
||||
self.cleanup_partial_files()
|
||||
|
||||
def _parse_output_line(self, line: str) -> None:
|
||||
"""Parse yt-dlp command output to update progress and status."""
|
||||
line = line.strip()
|
||||
# logger.info(f"yt-dlp: {line}") # Log all output - OPTIONALLY UNCOMMENT FOR VERBOSE DEBUG
|
||||
|
||||
# Extract filename when the destination line appears
|
||||
# Use a slightly more robust regex looking for the start of the line
|
||||
dest_match = re.search(r"^\[download\] Destination:\s*(.*)", line)
|
||||
if dest_match:
|
||||
try:
|
||||
filepath = dest_match.group(1).strip()
|
||||
self.current_filename = Path(filepath).name
|
||||
self.last_file_path = filepath # Store the full path for later cleanup
|
||||
logger.debug(f"Extracted filename: {self.current_filename}") # DEBUG
|
||||
|
||||
# Check if this is an audio-only download by looking in the previous lines
|
||||
is_audio_download = False
|
||||
|
||||
# Look for audio format indicators in the current line or preceding output
|
||||
# yt-dlp typically mentions format like "Downloading format 251 - audio only"
|
||||
if " - audio only" in line:
|
||||
is_audio_download = True
|
||||
# Check if the format ID is mentioned earlier in the line
|
||||
format_match = re.search(r"Downloading format (\d+)", line)
|
||||
if format_match:
|
||||
format_id = format_match.group(1)
|
||||
logger.debug(f"Detected format ID: {format_id}")
|
||||
# Format IDs for audio typically have different patterns
|
||||
# (like 140, 251 for audio vs 137, 248 for video)
|
||||
# This is just a heuristic since format IDs can vary
|
||||
|
||||
# Determine file type based on extension and context
|
||||
ext = Path(self.current_filename).suffix.lower()
|
||||
|
||||
# Check if this is explicitly an audio stream download
|
||||
if is_audio_download or "Downloading audio" in line:
|
||||
self.status_signal.emit(_("download.downloading_audio"))
|
||||
# Video file extensions with likely video content
|
||||
elif ext in VIDEO_EXTENSIONS:
|
||||
self.status_signal.emit(_("download.downloading_video"))
|
||||
# Audio file extensions
|
||||
elif ext in AUDIO_EXTENSIONS:
|
||||
self.status_signal.emit(_("download.downloading_audio"))
|
||||
# Subtitle file extensions
|
||||
elif ext in SUBTITLE_EXTENSIONS:
|
||||
self.status_signal.emit(_("download.downloading_subtitle"))
|
||||
# Default case
|
||||
else:
|
||||
self.status_signal.emit(_("download.downloading"))
|
||||
except Exception as e:
|
||||
logger.exception(f"Error extracting filename from line '{line}': {e}")
|
||||
self.status_signal.emit(_("download.downloading_fallback")) # Fallback status
|
||||
return # Don't process this line further for speed/ETA
|
||||
|
||||
# Check for specific download types in the output
|
||||
if "Downloading video" in line:
|
||||
self.status_signal.emit(_("download.downloading_video"))
|
||||
return
|
||||
|
||||
elif "Downloading audio" in line:
|
||||
self.status_signal.emit(_("download.downloading_audio"))
|
||||
return
|
||||
|
||||
# Detect subtitle file creation
|
||||
# Look for lines like "[info] Writing video subtitles to: filename.xx.vtt"
|
||||
subtitle_match = re.search(
|
||||
r"(?:Writing|Downloading) (?:video )?subtitles.*?(?:to|:)\s*(.+\.(?:vtt|srt))(?:\s|$)",
|
||||
line,
|
||||
re.IGNORECASE,
|
||||
)
|
||||
if subtitle_match:
|
||||
subtitle_file = subtitle_match.group(1).strip()
|
||||
|
||||
# Clean up the path - remove any duplicated directory paths
|
||||
# Sometimes yt-dlp output contains malformed paths like "dir: dir/file"
|
||||
if ":" in subtitle_file and os.name == "nt": # Windows paths
|
||||
# Look for pattern like "C:\path: C:\path\file" and extract the latter
|
||||
colon_parts = subtitle_file.split(": ")
|
||||
if len(colon_parts) > 1:
|
||||
# Take the last part which should be the actual file path
|
||||
subtitle_file = colon_parts[-1].strip()
|
||||
|
||||
# Show subtitle download message
|
||||
self.status_signal.emit(_("download.downloading_subtitle"))
|
||||
# Store the subtitle file path for later deletion if merging is enabled
|
||||
if self.merge_subs:
|
||||
subtitle_path = Path(subtitle_file)
|
||||
if not subtitle_path.is_absolute():
|
||||
# If it's a relative path, make it absolute based on current path
|
||||
subtitle_path = self.path.joinpath(subtitle_file)
|
||||
self.subtitle_files.append(str(subtitle_path))
|
||||
logger.debug(f"Tracking subtitle file for later cleanup: {subtitle_path}")
|
||||
return
|
||||
|
||||
# Send status updates based on output line content
|
||||
if "Downloading webpage" in line or "Extracting URL" in line:
|
||||
self.status_signal.emit(_("download.fetching_info"))
|
||||
self.progress_signal.emit(0)
|
||||
elif "[download] Destination:" in line:
|
||||
# Extract the destination filename
|
||||
match = re.search(r"Destination: (.+)", line)
|
||||
if match:
|
||||
dest_path = match.group(1).strip()
|
||||
self.current_filename = Path(dest_path).name
|
||||
self.last_file_path = dest_path
|
||||
logger.debug(f"Captured destination filename: {self.current_filename}")
|
||||
elif "Downloading API JSON" in line:
|
||||
self.status_signal.emit(_("download.processing_playlist"))
|
||||
self.progress_signal.emit(0)
|
||||
elif "Downloading m3u8 information" in line:
|
||||
self.status_signal.emit(_("download.preparing_streams"))
|
||||
self.progress_signal.emit(0)
|
||||
elif "[download] Downloading video " in line:
|
||||
self.status_signal.emit(_("download.downloading_video"))
|
||||
elif "[download] Downloading audio " in line:
|
||||
self.status_signal.emit(_("download.downloading_audio"))
|
||||
elif "Downloading format" in line:
|
||||
# Try to detect if it's audio or video format
|
||||
if " - audio only" in line:
|
||||
self.status_signal.emit(_("download.downloading_audio"))
|
||||
elif " - video only" in line:
|
||||
self.status_signal.emit(_("download.downloading_video"))
|
||||
else:
|
||||
# Don't emit generic message - format is unclear
|
||||
pass
|
||||
|
||||
# Look for download percentage
|
||||
percent_match = re.search(r"(\d+\.\d+)%", line)
|
||||
if percent_match:
|
||||
try:
|
||||
percent = float(percent_match.group(1))
|
||||
self.progress_signal.emit(percent)
|
||||
except (ValueError, IndexError):
|
||||
pass
|
||||
|
||||
# Check for download speed and ETA
|
||||
if "[download]" in line and "%" in line:
|
||||
# Try to extract more detailed status info
|
||||
try:
|
||||
# Look for speed
|
||||
speed_match = re.search(r"at\s+(\d+\.\d+[KMG]iB/s)", line)
|
||||
speed_str = speed_match.group(1) if speed_match else "N/A"
|
||||
|
||||
# Look for ETA
|
||||
eta_match = re.search(r"ETA\s+(\d+:\d+)", line)
|
||||
eta_str = eta_match.group(1) if eta_match else "N/A"
|
||||
|
||||
# Simplify status message to only show the speed and ETA
|
||||
status = f"{_('download.speed')}: {speed_str} | {_('download.eta')}: {eta_str}"
|
||||
self.update_details.emit(status)
|
||||
except Exception as e:
|
||||
# If parsing fails, just show basic status (maybe log the error)
|
||||
logger.exception(f"Error parsing download details line: {line} -> {e}")
|
||||
pass # Keep basic status emission below if needed, or emit generic details
|
||||
|
||||
# Check for post-processing
|
||||
if "[Merger]" in line or "Merging formats" in line:
|
||||
self.status_signal.emit(_("download.merging_formats"))
|
||||
self.progress_signal.emit(95)
|
||||
# Extract the merged output filename
|
||||
merger_match = re.search(r"Merging formats into \"(.+?)\"", line)
|
||||
if merger_match:
|
||||
merged_filepath = merger_match.group(1).strip()
|
||||
self.current_filename = Path(merged_filepath).name
|
||||
self.last_file_path = merged_filepath
|
||||
logger.debug(f"Updated to merged filename: {self.current_filename}")
|
||||
elif "SponsorBlock" in line:
|
||||
self.status_signal.emit(_("download.removing_sponsor_segments"))
|
||||
self.progress_signal.emit(97)
|
||||
elif "Deleting original file" in line:
|
||||
self.progress_signal.emit(98)
|
||||
elif "has already been downloaded" in line:
|
||||
# File already exists - extract filename
|
||||
match = re.search(r"(.*?) has already been downloaded", line)
|
||||
if match:
|
||||
filename = Path(match.group(1)).name
|
||||
# Determine file type based on extension for existing file message
|
||||
ext = Path(filename).suffix.lower()
|
||||
|
||||
if ext in VIDEO_EXTENSIONS:
|
||||
self.status_signal.emit(f"⚠️ Video file already exists")
|
||||
elif ext in AUDIO_EXTENSIONS:
|
||||
self.status_signal.emit(f"⚠️ Audio file already exists")
|
||||
elif ext in SUBTITLE_EXTENSIONS:
|
||||
self.status_signal.emit(f"⚠️ Subtitle file already exists")
|
||||
else:
|
||||
self.status_signal.emit(f"⚠️ File already exists")
|
||||
|
||||
self.file_exists_signal.emit(filename)
|
||||
else:
|
||||
logger.info(f"Could not extract filename from 'already downloaded' line: {line}")
|
||||
self.status_signal.emit(_("download.file_exists")) # Fallback status
|
||||
elif "Finished downloading" in line:
|
||||
self.progress_signal.emit(100)
|
||||
|
||||
# Show completion message based on file type
|
||||
if self.current_filename:
|
||||
ext = Path(self.current_filename).suffix.lower()
|
||||
|
||||
# Video file extensions
|
||||
if ext in VIDEO_EXTENSIONS:
|
||||
self.status_signal.emit(_("download.video_completed"))
|
||||
# Audio file extensions
|
||||
elif ext in AUDIO_EXTENSIONS:
|
||||
self.status_signal.emit(_("download.audio_completed"))
|
||||
# Subtitle file extensions
|
||||
elif ext in SUBTITLE_EXTENSIONS:
|
||||
self.status_signal.emit(_("download.subtitle_completed"))
|
||||
# Default case
|
||||
else:
|
||||
self.status_signal.emit(_("download.completed"))
|
||||
else:
|
||||
self.status_signal.emit(_("download.completed"))
|
||||
|
||||
self.update_details.emit("") # Clear details label on completion
|
||||
|
||||
def pause(self) -> None:
|
||||
self.paused = True
|
||||
|
||||
def resume(self) -> None:
|
||||
self.paused = False
|
||||
|
||||
def cancel(self) -> None:
|
||||
self.cancelled = True
|
||||
# Terminate the subprocess if it's running
|
||||
if self.process:
|
||||
try:
|
||||
self.process.terminate()
|
||||
except Exception:
|
||||
pass
|
||||
@@ -0,0 +1,467 @@
|
||||
import hashlib
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import requests
|
||||
|
||||
from ..utils.ytsage_logger import logger
|
||||
from ..utils.ytsage_constants import (
|
||||
FFMPEG_7Z_DOWNLOAD_URL,
|
||||
FFMPEG_7Z_SHA256_URL,
|
||||
FFMPEG_ZIP_DOWNLOAD_URL,
|
||||
FFMPEG_ZIP_SHA256_URL,
|
||||
OS_NAME,
|
||||
SUBPROCESS_CREATIONFLAGS,
|
||||
)
|
||||
|
||||
|
||||
def check_7zip_installed() -> bool:
|
||||
"""Check if 7-Zip is installed on Windows."""
|
||||
try:
|
||||
subprocess.run(["7z", "--help"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, creationflags=SUBPROCESS_CREATIONFLAGS)
|
||||
return True
|
||||
except (subprocess.SubprocessError, FileNotFoundError):
|
||||
return False
|
||||
|
||||
|
||||
def download_file(url, dest_path, progress_callback=None) -> bool:
|
||||
"""Download a file from URL to destination path with progress indication."""
|
||||
try:
|
||||
response = requests.get(url, stream=True, timeout=30) # Added timeout
|
||||
response.raise_for_status() # Check for HTTP errors
|
||||
total_size = int(response.headers.get("content-length", 0))
|
||||
|
||||
with open(dest_path, "wb") as f:
|
||||
if total_size == 0:
|
||||
f.write(response.content)
|
||||
else:
|
||||
downloaded = 0
|
||||
for data in response.iter_content(chunk_size=8192):
|
||||
downloaded += len(data)
|
||||
f.write(data)
|
||||
if progress_callback:
|
||||
progress = int((downloaded / total_size) * 100)
|
||||
progress_callback(f"⚡ Downloading FFmpeg components... {progress}%")
|
||||
return True
|
||||
except requests.RequestException as e:
|
||||
logger.info(f"Download error: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def get_file_sha256(file_path) -> str:
|
||||
"""Calculate SHA-256 hash of a file."""
|
||||
sha256_hash = hashlib.sha256()
|
||||
with open(file_path, "rb") as f:
|
||||
for chunk in iter(lambda: f.read(4096), b""):
|
||||
sha256_hash.update(chunk)
|
||||
return sha256_hash.hexdigest()
|
||||
|
||||
|
||||
def verify_sha256(file_path, expected_hash_url) -> bool:
|
||||
"""Verify file SHA-256 hash against expected hash from URL."""
|
||||
try:
|
||||
# Download the SHA-256 hash
|
||||
response = requests.get(expected_hash_url, timeout=10)
|
||||
response.raise_for_status()
|
||||
expected_hash = response.text.strip().split()[0] # Get just the hash part
|
||||
|
||||
# Calculate actual hash
|
||||
actual_hash = get_file_sha256(file_path)
|
||||
|
||||
# Compare hashes
|
||||
if actual_hash.lower() == expected_hash.lower():
|
||||
logger.info("SHA-256 verification successful!")
|
||||
return True
|
||||
else:
|
||||
logger.error(f"SHA-256 verification failed!")
|
||||
logger.info(f"Expected: {expected_hash}")
|
||||
logger.info(f"Actual: {actual_hash}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.info(f"⚠️ SHA-256 verification error: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def get_ffmpeg_install_path() -> Path:
|
||||
"""
|
||||
Get the FFmpeg installation path.
|
||||
For Windows, tries to find the latest essentials build dynamically.
|
||||
"""
|
||||
if OS_NAME == "Windows":
|
||||
ffmpeg_base = Path(os.getenv("LOCALAPPDATA")) / "ffmpeg" # type: ignore
|
||||
|
||||
# If the directory exists, look for any ffmpeg-*-essentials_build folder
|
||||
if ffmpeg_base.exists():
|
||||
# Find all directories matching the pattern
|
||||
essentials_dirs = list(ffmpeg_base.glob("ffmpeg-*-essentials_build"))
|
||||
if essentials_dirs:
|
||||
# Sort by name (which includes version) and take the latest
|
||||
latest_dir = sorted(essentials_dirs, reverse=True)[0]
|
||||
bin_dir = latest_dir / "bin"
|
||||
if bin_dir.exists():
|
||||
return bin_dir
|
||||
|
||||
# Fallback: return default path (even if it doesn't exist yet)
|
||||
return ffmpeg_base / "ffmpeg-essentials_build" / "bin"
|
||||
|
||||
elif OS_NAME == "Darwin":
|
||||
paths = ["/usr/local/bin", "/opt/homebrew/bin", "/usr/bin"]
|
||||
for path in paths:
|
||||
if Path(path).joinpath("ffmpeg").exists():
|
||||
return Path(path)
|
||||
return Path("/usr/local/bin") # Default Homebrew path
|
||||
else:
|
||||
return Path("/usr/bin") # Standard Linux path
|
||||
|
||||
|
||||
|
||||
def get_ffmpeg_path() -> str | Path:
|
||||
"""
|
||||
Get the FFmpeg executable path, either from PATH or installation directory.
|
||||
Returns:
|
||||
str: Path to FFmpeg executable or 'ffmpeg' if found in PATH but path unknown
|
||||
"""
|
||||
try:
|
||||
# First try to find ffmpeg in PATH using 'where' on Windows or 'which' on Unix
|
||||
if OS_NAME == "Windows":
|
||||
# On Windows, use 'where' command and hide console window
|
||||
# Extra logic moved to src\utils\ytsage_constants.py
|
||||
|
||||
result = subprocess.run(
|
||||
["where", "ffmpeg"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
creationflags=SUBPROCESS_CREATIONFLAGS,
|
||||
)
|
||||
if result.returncode == 0 and result.stdout.strip():
|
||||
ffmpeg_path = result.stdout.strip().split("\n")[0]
|
||||
return ffmpeg_path
|
||||
else:
|
||||
# On Unix systems, use 'which' command
|
||||
result = subprocess.run(["which", "ffmpeg"], capture_output=True, text=True, check=False)
|
||||
if result.returncode == 0 and result.stdout.strip():
|
||||
ffmpeg_path = result.stdout.strip()
|
||||
return ffmpeg_path
|
||||
except Exception as e:
|
||||
logger.exception(f"Error finding ffmpeg in PATH: {e}")
|
||||
|
||||
# If not found in PATH, check the installation directory
|
||||
ffmpeg_install_path = get_ffmpeg_install_path()
|
||||
if OS_NAME == "Windows":
|
||||
ffmpeg_exe = Path(ffmpeg_install_path).joinpath("ffmpeg.exe")
|
||||
else:
|
||||
ffmpeg_exe = Path(ffmpeg_install_path).joinpath("ffmpeg")
|
||||
|
||||
if ffmpeg_exe.exists():
|
||||
return ffmpeg_exe
|
||||
|
||||
# Return command name as fallback
|
||||
return "ffmpeg"
|
||||
|
||||
|
||||
def check_ffmpeg_installed() -> bool:
|
||||
"""Check if FFmpeg is installed and accessible."""
|
||||
try:
|
||||
# First try the PATH
|
||||
result = subprocess.run(
|
||||
["ffmpeg", "-version"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
check=True,
|
||||
creationflags=SUBPROCESS_CREATIONFLAGS,
|
||||
timeout=5,
|
||||
) # Added timeout
|
||||
return True
|
||||
except (subprocess.SubprocessError, FileNotFoundError):
|
||||
# If not in PATH, check the installation directory
|
||||
ffmpeg_path = get_ffmpeg_install_path()
|
||||
if OS_NAME == "Windows":
|
||||
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg.exe")
|
||||
else:
|
||||
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg")
|
||||
|
||||
if ffmpeg_exe.exists():
|
||||
# Add to PATH if found
|
||||
os.environ["PATH"] = f"{ffmpeg_path}{os.pathsep}{os.environ.get('PATH', '')}"
|
||||
return True
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.info(f"FFmpeg check error: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def install_ffmpeg_windows(progress_callback=None) -> bool:
|
||||
"""Install FFmpeg on Windows using essentials build with 7z method primarily, with zip as fallback."""
|
||||
# Check if already installed
|
||||
if check_ffmpeg_installed():
|
||||
logger.info("FFmpeg is already installed!")
|
||||
if progress_callback:
|
||||
progress_callback("✅ FFmpeg is already installed!")
|
||||
return True
|
||||
|
||||
try:
|
||||
# Define variables for essentials build
|
||||
extract_dir = Path(os.getenv("LOCALAPPDATA")) / "ffmpeg" # type: ignore
|
||||
|
||||
# Create extraction directory if it doesn't exist
|
||||
extract_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Try 7z method first (smaller size)
|
||||
use_7zip = check_7zip_installed()
|
||||
success = False
|
||||
|
||||
if use_7zip:
|
||||
logger.info("Using 7-Zip method (smaller download size)...")
|
||||
if progress_callback:
|
||||
progress_callback("⚡ Using 7-Zip method...")
|
||||
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".7z").name
|
||||
|
||||
# Download 7z file
|
||||
if progress_callback:
|
||||
progress_callback("⬇ Downloading FFmpeg (7z)...")
|
||||
if download_file(
|
||||
FFMPEG_7Z_DOWNLOAD_URL,
|
||||
temp_file,
|
||||
progress_callback=progress_callback,
|
||||
):
|
||||
# Verify SHA-256 hash for 7z file
|
||||
if progress_callback:
|
||||
progress_callback("🔐 Verifying download integrity...")
|
||||
if verify_sha256(temp_file, FFMPEG_7Z_SHA256_URL):
|
||||
logger.info("Extracting FFmpeg components from 7z archive...")
|
||||
if progress_callback:
|
||||
progress_callback("⚙ Extracting FFmpeg...")
|
||||
try:
|
||||
subprocess.run(
|
||||
["7z", "x", temp_file, f"-o{extract_dir}", "-y"],
|
||||
creationflags=SUBPROCESS_CREATIONFLAGS,
|
||||
timeout=300,
|
||||
check=True,
|
||||
)
|
||||
success = True
|
||||
except Exception as e:
|
||||
logger.exception(f"7z extraction failed: {e}, trying zip fallback...")
|
||||
if progress_callback:
|
||||
progress_callback("❌ 7z extraction failed, trying zip fallback...")
|
||||
else:
|
||||
logger.error("SHA-256 verification failed for 7z file, trying zip fallback...")
|
||||
if progress_callback:
|
||||
progress_callback("❌ SHA-256 verification failed, trying zip fallback...")
|
||||
|
||||
# Clean up temp file
|
||||
try:
|
||||
Path(temp_file).unlink(missing_ok=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Fallback to zip method if 7z failed or not available
|
||||
if not success:
|
||||
logger.info("Using ZIP method as fallback...")
|
||||
if progress_callback:
|
||||
progress_callback("📦 Using ZIP method...")
|
||||
temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".zip").name
|
||||
|
||||
# Download zip file
|
||||
if progress_callback:
|
||||
progress_callback("⬇ Downloading FFmpeg (zip)...")
|
||||
if not download_file(
|
||||
FFMPEG_ZIP_DOWNLOAD_URL,
|
||||
temp_file,
|
||||
progress_callback=progress_callback,
|
||||
):
|
||||
logger.error("Failed to download FFmpeg (both 7z and zip methods failed)")
|
||||
if progress_callback:
|
||||
progress_callback("❌ Failed to download FFmpeg")
|
||||
return False
|
||||
|
||||
# Verify SHA-256 hash for zip
|
||||
if progress_callback:
|
||||
progress_callback("🔐 Verifying download integrity...")
|
||||
if not verify_sha256(temp_file, FFMPEG_ZIP_SHA256_URL):
|
||||
logger.warning("SHA-256 verification failed for zip file, proceeding anyway...")
|
||||
if progress_callback:
|
||||
progress_callback("⚠️ SHA-256 verification failed, proceeding anyway...")
|
||||
|
||||
logger.info("Extracting FFmpeg components from zip archive...")
|
||||
if progress_callback:
|
||||
progress_callback("⚙ Extracting FFmpeg...")
|
||||
try:
|
||||
import zipfile
|
||||
with zipfile.ZipFile(temp_file, "r") as zip_ref:
|
||||
zip_ref.extractall(extract_dir)
|
||||
success = True
|
||||
except Exception as e:
|
||||
logger.exception(f"Extraction failed: {e}")
|
||||
if progress_callback:
|
||||
progress_callback(f"❌ Extraction failed: {e}")
|
||||
return False
|
||||
finally:
|
||||
# Clean up temp file
|
||||
try:
|
||||
Path(temp_file).unlink(missing_ok=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not success:
|
||||
logger.error("Both 7z and zip methods failed")
|
||||
if progress_callback:
|
||||
progress_callback("❌ Installation failed")
|
||||
return False
|
||||
|
||||
# Find the bin directory in the extracted essentials build
|
||||
logger.info("Locating FFmpeg binaries...")
|
||||
if progress_callback:
|
||||
progress_callback("🔍 Locating FFmpeg binaries...")
|
||||
|
||||
bin_dir = None
|
||||
# Look for any directory matching ffmpeg-*-essentials_build pattern
|
||||
for item in extract_dir.iterdir():
|
||||
if item.is_dir() and "essentials_build" in item.name.lower():
|
||||
potential_bin = item / "bin"
|
||||
if potential_bin.exists():
|
||||
bin_dir = potential_bin
|
||||
logger.info(f"Found FFmpeg bin directory: {bin_dir}")
|
||||
break
|
||||
|
||||
if not bin_dir:
|
||||
logger.error("Could not locate FFmpeg bin directory")
|
||||
if progress_callback:
|
||||
progress_callback("❌ Could not locate FFmpeg bin directory")
|
||||
return False
|
||||
|
||||
logger.info("Configuring system paths...")
|
||||
if progress_callback:
|
||||
progress_callback("🔧 Configuring system paths...")
|
||||
|
||||
# Add to System Path
|
||||
user_path = os.environ.get("PATH", "")
|
||||
path_parts = user_path.split(os.pathsep)
|
||||
|
||||
# Remove old FFmpeg paths and add new one
|
||||
cleaned_paths = [p for p in path_parts if "ffmpeg" not in p.lower() or str(bin_dir) in p]
|
||||
if str(bin_dir) not in cleaned_paths:
|
||||
cleaned_paths.insert(0, str(bin_dir))
|
||||
|
||||
new_path = os.pathsep.join(cleaned_paths)
|
||||
|
||||
try:
|
||||
subprocess.run(
|
||||
["setx", "PATH", new_path],
|
||||
creationflags=SUBPROCESS_CREATIONFLAGS,
|
||||
timeout=30,
|
||||
check=True,
|
||||
)
|
||||
os.environ["PATH"] = new_path
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to update PATH permanently: {e}")
|
||||
# Still update for current session
|
||||
os.environ["PATH"] = new_path
|
||||
|
||||
# Verify installation
|
||||
if progress_callback:
|
||||
progress_callback("✅ Verifying installation...")
|
||||
if not check_ffmpeg_installed():
|
||||
logger.error("FFmpeg installation verification failed")
|
||||
if progress_callback:
|
||||
progress_callback("⚠️ Installation completed but verification failed")
|
||||
return True # Still return True as files were extracted
|
||||
|
||||
logger.info("FFmpeg installation completed successfully!")
|
||||
if progress_callback:
|
||||
progress_callback("✅ FFmpeg installation completed successfully!")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error installing FFmpeg: {e}")
|
||||
if progress_callback:
|
||||
progress_callback(f"❌ Error installing FFmpeg: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def install_ffmpeg_macos() -> bool:
|
||||
"""Install FFmpeg on macOS using Homebrew."""
|
||||
try:
|
||||
# Check if Homebrew is installed
|
||||
try:
|
||||
subprocess.run(
|
||||
["brew", "--version"],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
check=True,
|
||||
timeout=5,
|
||||
)
|
||||
except (subprocess.SubprocessError, FileNotFoundError):
|
||||
logger.info("Installing Homebrew...")
|
||||
brew_install_cmd = '/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"'
|
||||
subprocess.run(brew_install_cmd, shell=True, check=True, timeout=300)
|
||||
|
||||
# Install FFmpeg
|
||||
logger.info("Installing FFmpeg...")
|
||||
subprocess.run(["brew", "install", "ffmpeg"], check=True, timeout=300)
|
||||
|
||||
# Verify installation
|
||||
if not check_ffmpeg_installed():
|
||||
logger.error("FFmpeg installation verification failed")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error installing FFmpeg: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def install_ffmpeg_linux() -> bool:
|
||||
"""Install FFmpeg on Linux using appropriate package manager."""
|
||||
try:
|
||||
# Detect the package manager
|
||||
if shutil.which("apt"):
|
||||
# Debian/Ubuntu
|
||||
subprocess.run(["sudo", "apt", "update"], check=True, timeout=60)
|
||||
subprocess.run(["sudo", "apt", "install", "-y", "ffmpeg"], check=True, timeout=300)
|
||||
elif shutil.which("dnf"):
|
||||
# Fedora
|
||||
subprocess.run(["sudo", "dnf", "install", "-y", "ffmpeg"], check=True, timeout=300)
|
||||
elif shutil.which("pacman"):
|
||||
# Arch Linux
|
||||
subprocess.run(
|
||||
["sudo", "pacman", "-S", "--noconfirm", "ffmpeg"],
|
||||
check=True,
|
||||
timeout=300,
|
||||
)
|
||||
elif shutil.which("snap"):
|
||||
# Universal snap package
|
||||
subprocess.run(["sudo", "snap", "install", "ffmpeg"], check=True, timeout=300)
|
||||
else:
|
||||
logger.error("No supported package manager found")
|
||||
return False
|
||||
|
||||
# Verify installation
|
||||
if not check_ffmpeg_installed():
|
||||
logger.error("FFmpeg installation verification failed")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error installing FFmpeg: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def auto_install_ffmpeg(progress_callback=None) -> bool:
|
||||
"""Automatically install FFmpeg based on the operating system."""
|
||||
if OS_NAME == "Windows":
|
||||
return install_ffmpeg_windows(progress_callback=progress_callback)
|
||||
elif OS_NAME == "Darwin":
|
||||
return install_ffmpeg_macos()
|
||||
elif OS_NAME == "Linux":
|
||||
return install_ffmpeg_linux()
|
||||
else:
|
||||
logger.info(f"Unsupported operating system: {OS_NAME}")
|
||||
if progress_callback:
|
||||
progress_callback(f"❌ Unsupported operating system: {OS_NAME}")
|
||||
return False
|
||||
@@ -0,0 +1,826 @@
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from importlib.metadata import PackageNotFoundError
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional, Union
|
||||
|
||||
import requests
|
||||
from packaging import version
|
||||
|
||||
from .ytsage_ffmpeg import check_ffmpeg_installed, get_ffmpeg_install_path
|
||||
from .ytsage_yt_dlp import get_yt_dlp_path
|
||||
from ..utils.ytsage_constants import (
|
||||
APP_CONFIG_FILE,
|
||||
OS_NAME,
|
||||
SUBPROCESS_CREATIONFLAGS,
|
||||
USER_HOME_DIR,
|
||||
YTDLP_APP_BIN_PATH,
|
||||
YTDLP_DOWNLOAD_URL,
|
||||
)
|
||||
from ..utils.ytsage_localization import _
|
||||
from ..utils.ytsage_logger import logger
|
||||
|
||||
try:
|
||||
from importlib.metadata import PackageNotFoundError as ImportlibPackageNotFoundError
|
||||
from importlib.metadata import version as importlib_version
|
||||
|
||||
def get_version(package_name: str) -> str:
|
||||
return importlib_version(package_name)
|
||||
|
||||
PackageNotFoundError = ImportlibPackageNotFoundError
|
||||
except ImportError:
|
||||
# Fallback for older Python versions
|
||||
import pkg_resources
|
||||
|
||||
def get_version(package_name: str) -> str:
|
||||
return pkg_resources.get_distribution(package_name).version
|
||||
|
||||
PackageNotFoundError = pkg_resources.DistributionNotFound
|
||||
|
||||
|
||||
# Cache for version information to avoid delays
|
||||
_version_cache: Dict[str, Dict[str, Any]] = {
|
||||
"ytdlp": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
|
||||
"ffmpeg": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
|
||||
"deno": {"version": None, "path": None, "last_check": 0, "path_mtime": 0},
|
||||
}
|
||||
|
||||
# Cache expiry time in seconds (5 minutes)
|
||||
CACHE_EXPIRY: int = 300
|
||||
|
||||
|
||||
def get_file_mtime(filepath: Optional[Union[str, Path]]) -> float:
|
||||
"""Get file modification time safely."""
|
||||
try:
|
||||
if filepath and Path(filepath).exists():
|
||||
return Path(filepath).stat().st_mtime
|
||||
except Exception:
|
||||
pass
|
||||
return 0.0
|
||||
|
||||
|
||||
def should_refresh_cache(tool_name: str, current_path: Optional[str]) -> bool:
|
||||
"""Determine if cache should be refreshed for a tool."""
|
||||
cache: Dict[str, Any] = _version_cache.get(tool_name, {})
|
||||
current_time: float = time.time()
|
||||
|
||||
# Always refresh if no cached data
|
||||
if not cache.get("version"):
|
||||
return True
|
||||
|
||||
# Refresh if path changed
|
||||
if cache.get("path") != current_path:
|
||||
return True
|
||||
|
||||
# Refresh if file was modified
|
||||
current_mtime = get_file_mtime(current_path)
|
||||
if current_mtime > cache.get("path_mtime", 0):
|
||||
return True
|
||||
|
||||
# Refresh if cache expired
|
||||
if current_time - cache.get("last_check", 0) > CACHE_EXPIRY:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def update_version_cache(tool_name: str, version_info: str, path: Optional[str], force_save: bool = False) -> None:
|
||||
"""Update the version cache and optionally save to config."""
|
||||
current_time: float = time.time()
|
||||
current_mtime: float = get_file_mtime(path)
|
||||
|
||||
_version_cache[tool_name] = {
|
||||
"version": version_info,
|
||||
"path": path,
|
||||
"last_check": current_time,
|
||||
"path_mtime": current_mtime,
|
||||
}
|
||||
|
||||
# Save to persistent config
|
||||
if force_save:
|
||||
save_version_cache_to_config()
|
||||
|
||||
|
||||
def load_version_cache_from_config() -> None:
|
||||
"""Load cached version info from config file."""
|
||||
from ..utils.ytsage_config_manager import ConfigManager
|
||||
|
||||
try:
|
||||
cached_versions = ConfigManager.get("cached_versions") or {}
|
||||
|
||||
for tool_name, cache_data in cached_versions.items():
|
||||
if tool_name in _version_cache:
|
||||
_version_cache[tool_name].update(cache_data)
|
||||
except Exception as e:
|
||||
logger.exception(f"Error loading version cache: {e}")
|
||||
|
||||
|
||||
def save_version_cache_to_config() -> None:
|
||||
"""Save version cache to config file."""
|
||||
from ..utils.ytsage_config_manager import ConfigManager
|
||||
|
||||
try:
|
||||
ConfigManager.set("cached_versions", _version_cache.copy())
|
||||
except Exception as e:
|
||||
logger.exception(f"Error saving version cache: {e}")
|
||||
|
||||
|
||||
def get_ytdlp_version_cached() -> str:
|
||||
"""Get yt-dlp version with caching support."""
|
||||
try:
|
||||
current_path = get_yt_dlp_path()
|
||||
|
||||
# Check if we need to refresh cache
|
||||
if not should_refresh_cache("ytdlp", current_path):
|
||||
cached_version = _version_cache["ytdlp"].get("version")
|
||||
if cached_version:
|
||||
return cached_version
|
||||
|
||||
# Get fresh version info
|
||||
version_info = get_ytdlp_version_direct(current_path)
|
||||
|
||||
# Update cache
|
||||
update_version_cache("ytdlp", version_info, current_path)
|
||||
|
||||
return version_info
|
||||
except Exception as e:
|
||||
logger.exception(f"Error getting cached yt-dlp version: {e}")
|
||||
return "Error getting version"
|
||||
|
||||
|
||||
def get_ffmpeg_version_cached() -> str:
|
||||
"""Get FFmpeg version with caching support."""
|
||||
try:
|
||||
# Try to find ffmpeg path
|
||||
current_path = "ffmpeg" # Default to system PATH
|
||||
|
||||
# Check if we need to refresh cache
|
||||
if not should_refresh_cache("ffmpeg", current_path):
|
||||
cached_version = _version_cache["ffmpeg"].get("version")
|
||||
if cached_version:
|
||||
return cached_version
|
||||
|
||||
# Get fresh version info
|
||||
version_info = get_ffmpeg_version_direct()
|
||||
|
||||
# Update cache
|
||||
update_version_cache("ffmpeg", version_info, current_path)
|
||||
|
||||
return version_info
|
||||
except Exception as e:
|
||||
logger.exception(f"Error getting cached FFmpeg version: {e}")
|
||||
return "Error getting version"
|
||||
|
||||
|
||||
def get_deno_version_cached() -> str:
|
||||
"""Get Deno version with caching support."""
|
||||
try:
|
||||
from .ytsage_deno import get_deno_path
|
||||
|
||||
current_path = get_deno_path()
|
||||
|
||||
# Check if we need to refresh cache
|
||||
if not should_refresh_cache("deno", current_path):
|
||||
cached_version = _version_cache["deno"].get("version")
|
||||
if cached_version:
|
||||
return cached_version
|
||||
|
||||
# Get fresh version info
|
||||
from .ytsage_deno import get_deno_version_direct
|
||||
version_info = get_deno_version_direct(current_path)
|
||||
|
||||
# Update cache
|
||||
update_version_cache("deno", version_info, current_path)
|
||||
|
||||
return version_info
|
||||
except Exception as e:
|
||||
logger.exception(f"Error getting cached Deno version: {e}")
|
||||
return "Error getting version"
|
||||
|
||||
|
||||
def refresh_version_cache(force=False) -> bool:
|
||||
"""Manually refresh version cache for all tools."""
|
||||
try:
|
||||
# Refresh yt-dlp
|
||||
current_path = get_yt_dlp_path()
|
||||
version_info = get_ytdlp_version_direct(current_path)
|
||||
update_version_cache("ytdlp", version_info, current_path, force_save=True)
|
||||
|
||||
# Refresh FFmpeg
|
||||
version_info = get_ffmpeg_version_direct()
|
||||
update_version_cache("ffmpeg", version_info, "ffmpeg", force_save=True)
|
||||
|
||||
# Refresh Deno
|
||||
from .ytsage_deno import get_deno_path, get_deno_version_direct
|
||||
deno_path = get_deno_path()
|
||||
version_info = get_deno_version_direct(deno_path)
|
||||
update_version_cache("deno", version_info, deno_path, force_save=True)
|
||||
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.exception(f"Error refreshing version cache: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def get_ytdlp_version() -> str:
|
||||
"""Get the version of yt-dlp (uses cached version for performance)."""
|
||||
return get_ytdlp_version_cached()
|
||||
|
||||
|
||||
def get_ffmpeg_version() -> str:
|
||||
"""Get the version of FFmpeg (uses cached version for performance)."""
|
||||
return get_ffmpeg_version_cached()
|
||||
|
||||
|
||||
def get_deno_version() -> str:
|
||||
"""Get the version of Deno (uses cached version for performance)."""
|
||||
return get_deno_version_cached()
|
||||
|
||||
|
||||
def get_ytdlp_version_direct(yt_dlp_path: Optional[str] = None) -> str:
|
||||
"""Get yt-dlp version directly without caching."""
|
||||
try:
|
||||
if yt_dlp_path is None:
|
||||
yt_dlp_path = get_yt_dlp_path()
|
||||
|
||||
if not yt_dlp_path or yt_dlp_path == "yt-dlp":
|
||||
return "Not found"
|
||||
|
||||
# Extra logic moved to src\utils\ytsage_constants.py
|
||||
result = subprocess.run(
|
||||
[yt_dlp_path, "--version"], capture_output=True, text=True, timeout=10, creationflags=SUBPROCESS_CREATIONFLAGS
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
return result.stdout.strip()
|
||||
else:
|
||||
return "Error getting version"
|
||||
except Exception as e:
|
||||
logger.exception(f"Error getting yt-dlp version: {e}")
|
||||
return "Error getting version"
|
||||
|
||||
|
||||
def get_ffmpeg_version_direct() -> str:
|
||||
"""Get FFmpeg version directly without caching."""
|
||||
try:
|
||||
# Extra logic moved to src\utils\ytsage_constants.py
|
||||
result = subprocess.run(
|
||||
["ffmpeg", "-version"], capture_output=True, text=True, timeout=10, creationflags=SUBPROCESS_CREATIONFLAGS
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
# Parse the first line to get version info
|
||||
lines = result.stdout.split("\n")
|
||||
if lines:
|
||||
first_line = lines[0]
|
||||
# Extract version from something like "ffmpeg version 4.4.2 Copyright..."
|
||||
if "version" in first_line:
|
||||
parts = first_line.split()
|
||||
for i, part in enumerate(parts):
|
||||
if part == "version" and i + 1 < len(parts):
|
||||
return parts[i + 1]
|
||||
return first_line.strip()
|
||||
return "Unknown version"
|
||||
else:
|
||||
return "Not found"
|
||||
except FileNotFoundError:
|
||||
# If ffmpeg is not in PATH, try the installation directory
|
||||
try:
|
||||
ffmpeg_path = get_ffmpeg_install_path()
|
||||
if OS_NAME == "Windows":
|
||||
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg.exe")
|
||||
else:
|
||||
ffmpeg_exe = Path(ffmpeg_path).joinpath("ffmpeg")
|
||||
|
||||
if ffmpeg_exe.exists():
|
||||
result = subprocess.run(
|
||||
[ffmpeg_exe, "-version"], capture_output=True, text=True, timeout=10, creationflags=SUBPROCESS_CREATIONFLAGS
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
lines = result.stdout.split("\n")
|
||||
if lines:
|
||||
first_line = lines[0]
|
||||
if "version" in first_line:
|
||||
parts = first_line.split()
|
||||
for i, part in enumerate(parts):
|
||||
if part == "version" and i + 1 < len(parts):
|
||||
return parts[i + 1]
|
||||
return first_line.strip()
|
||||
return "Unknown version"
|
||||
return "Not found"
|
||||
except Exception as e:
|
||||
logger.exception(f"Error getting FFmpeg version from install path: {e}")
|
||||
return "Not found"
|
||||
except Exception as e:
|
||||
logger.exception(f"Error getting FFmpeg version: {e}")
|
||||
return "Error getting version"
|
||||
|
||||
|
||||
# get_app_data_dir() moved to src\utils\ytsage_constants.py
|
||||
# get_config_file_path() moved to src\utils\ytsage_constants.py
|
||||
# ensure_app_data_dir() moved to src\utils\ytsage_constants.py
|
||||
# load_config() and save_config() removed - use ConfigManager instead
|
||||
|
||||
|
||||
def check_ffmpeg() -> bool:
|
||||
"""Check if FFmpeg is installed and accessible with enhanced error handling."""
|
||||
try:
|
||||
# Use the enhanced FFmpeg check from ytsage_ffmpeg
|
||||
if check_ffmpeg_installed():
|
||||
return True
|
||||
|
||||
# For Windows, try to add the FFmpeg path to environment
|
||||
if OS_NAME == "Windows":
|
||||
ffmpeg_path = get_ffmpeg_install_path()
|
||||
if ffmpeg_path.joinpath("ffmpeg.exe").exists():
|
||||
try:
|
||||
# Add to current session PATH
|
||||
os.environ["PATH"] = f"{ffmpeg_path}{os.pathsep}{os.environ.get('PATH', '')}"
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.exception(f"Error updating PATH: {e}")
|
||||
return False
|
||||
|
||||
# For macOS, check common paths
|
||||
elif OS_NAME == "Darwin":
|
||||
common_paths = [
|
||||
"/usr/local/bin/ffmpeg",
|
||||
"/opt/homebrew/bin/ffmpeg",
|
||||
"/usr/bin/ffmpeg",
|
||||
]
|
||||
for path in common_paths:
|
||||
if Path(path).exists():
|
||||
try:
|
||||
ffmpeg_dir = Path(path).parent
|
||||
os.environ["PATH"] = f"{ffmpeg_dir}{os.pathsep}{os.environ.get('PATH', '')}"
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.exception(f"Error updating PATH: {e}")
|
||||
continue
|
||||
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error checking FFmpeg: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def load_saved_path(main_window_instance: Any) -> None:
|
||||
"""Load saved download path with enhanced error handling."""
|
||||
try:
|
||||
if APP_CONFIG_FILE.exists():
|
||||
try:
|
||||
with open(APP_CONFIG_FILE, "r", encoding="utf-8") as f:
|
||||
config = json.load(f)
|
||||
saved_path = config.get("download_path", "")
|
||||
if Path(saved_path).exists() and os.access(saved_path, os.W_OK):
|
||||
main_window_instance.last_path = saved_path
|
||||
return
|
||||
except (json.JSONDecodeError, UnicodeError) as e:
|
||||
logger.exception(f"Error reading config file: {e}")
|
||||
# If config file is corrupted, try to remove it
|
||||
try:
|
||||
APP_CONFIG_FILE.unlink(missing_ok=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Fallback to Downloads folder
|
||||
downloads_path = USER_HOME_DIR / "Downloads"
|
||||
if downloads_path.exists() and os.access(downloads_path, os.W_OK):
|
||||
main_window_instance.last_path = downloads_path
|
||||
else:
|
||||
# Final fallback to temp directory if Downloads is not accessible
|
||||
main_window_instance.last_path = tempfile.gettempdir()
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error loading saved settings: {e}")
|
||||
main_window_instance.last_path = tempfile.gettempdir()
|
||||
|
||||
|
||||
def save_path(main_window_instance: Any, path: Union[str, Path]) -> bool:
|
||||
"""Save download path with enhanced error handling."""
|
||||
try:
|
||||
# Verify the path is valid and writable
|
||||
if not Path(path).exists():
|
||||
try:
|
||||
Path(path).mkdir(exist_ok=True)
|
||||
except Exception as e:
|
||||
logger.exception(f"Error creating directory: {e}")
|
||||
return False
|
||||
|
||||
if not os.access(path, os.W_OK):
|
||||
logger.info("Path is not writable")
|
||||
return False
|
||||
|
||||
# Save the config
|
||||
config = {"download_path": path}
|
||||
with open(APP_CONFIG_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump(config, f, ensure_ascii=False)
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error saving settings: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def update_yt_dlp() -> bool:
|
||||
"""Check for yt-dlp updates and update if a newer version is available."""
|
||||
try:
|
||||
# Get the yt-dlp path
|
||||
yt_dlp_path: Path = get_yt_dlp_path()
|
||||
|
||||
# Extra logic moved to src\utils\ytsage_constants.py
|
||||
|
||||
# For binaries downloaded with our app, use direct binary update approach
|
||||
# Check if this is an app-managed binary by comparing paths safely
|
||||
is_app_managed: bool = False
|
||||
try:
|
||||
# Only compare if both files exist
|
||||
if yt_dlp_path.exists() and YTDLP_APP_BIN_PATH.exists():
|
||||
is_app_managed = yt_dlp_path.samefile(YTDLP_APP_BIN_PATH)
|
||||
elif str(yt_dlp_path) == str(YTDLP_APP_BIN_PATH):
|
||||
# If paths are identical as strings, consider it app-managed
|
||||
is_app_managed = True
|
||||
else:
|
||||
# If app binary doesn't exist, this is definitely not app-managed
|
||||
is_app_managed = False
|
||||
except (OSError, IOError) as e:
|
||||
logger.debug(f"Error comparing paths in update_yt_dlp: {e}")
|
||||
is_app_managed = False
|
||||
|
||||
if is_app_managed:
|
||||
# We're using a binary installed by our app, update directly
|
||||
logger.info(f"Updating yt-dlp binary at {yt_dlp_path}")
|
||||
|
||||
# Determine the URL based on OS
|
||||
# Extra logic moved to src\utils\ytsage_constants.py
|
||||
|
||||
# Download the latest version
|
||||
try:
|
||||
response = requests.get(YTDLP_DOWNLOAD_URL, stream=True)
|
||||
if response.status_code == 200:
|
||||
# Create a temporary file
|
||||
temp_file = f"{yt_dlp_path}.new"
|
||||
|
||||
with open(temp_file, "wb") as f:
|
||||
for chunk in response.iter_content(chunk_size=8192):
|
||||
f.write(chunk)
|
||||
|
||||
# Make executable on Unix systems
|
||||
if OS_NAME != "Windows":
|
||||
os.chmod(temp_file, 0o755)
|
||||
|
||||
# Replace the old file with the new one
|
||||
try:
|
||||
# On Windows, we need to remove the old file first
|
||||
if OS_NAME == "Windows" and yt_dlp_path.exists():
|
||||
yt_dlp_path.unlink(missing_ok=True)
|
||||
|
||||
Path(temp_file).rename(yt_dlp_path)
|
||||
logger.info("yt-dlp binary successfully updated")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.exception(f"Error replacing yt-dlp binary: {e}")
|
||||
return False
|
||||
else:
|
||||
logger.info(f"Failed to download latest yt-dlp: HTTP {response.status_code}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.exception(f"Error downloading yt-dlp update: {e}")
|
||||
return False
|
||||
else:
|
||||
# We're using a system-installed yt-dlp, use pip to update
|
||||
logger.info("Using pip to update yt-dlp")
|
||||
|
||||
# Get current version
|
||||
try:
|
||||
current_version = get_version("yt-dlp")
|
||||
logger.info(f"Current yt-dlp version: {current_version}")
|
||||
except PackageNotFoundError:
|
||||
logger.info("yt-dlp not installed via pip, attempting update anyway")
|
||||
current_version = "0.0.0" # Assume very old version to force update
|
||||
|
||||
# Get the latest version from PyPI JSON API
|
||||
try:
|
||||
response = requests.get("https://pypi.org/pypi/yt-dlp/json", timeout=10)
|
||||
if response.status_code == 200:
|
||||
data = response.json()
|
||||
latest_version = data["info"]["version"]
|
||||
logger.info(f"Latest available yt-dlp version: {latest_version}")
|
||||
|
||||
# Compare versions and update if needed
|
||||
if version.parse(latest_version) > version.parse(current_version):
|
||||
logger.info(f"Updating yt-dlp from {current_version} to {latest_version}...")
|
||||
update_result = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-m",
|
||||
"pip",
|
||||
"install",
|
||||
"--upgrade",
|
||||
"yt-dlp",
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
creationflags=SUBPROCESS_CREATIONFLAGS,
|
||||
)
|
||||
if update_result.returncode == 0:
|
||||
logger.info("yt-dlp successfully updated")
|
||||
return True
|
||||
else:
|
||||
logger.error(f"Error updating yt-dlp: {update_result.stderr}")
|
||||
else:
|
||||
logger.info("yt-dlp is already up to date")
|
||||
return True
|
||||
else:
|
||||
logger.info(f"Failed to get latest version info: HTTP {response.status_code}")
|
||||
except Exception as e:
|
||||
logger.exception(f"Error checking for yt-dlp updates: {e}")
|
||||
except Exception as e:
|
||||
logger.exception(f"Unexpected error during yt-dlp update: {e}")
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def should_check_for_auto_update() -> bool:
|
||||
"""Check if auto-update should be performed based on user settings."""
|
||||
from ..utils.ytsage_config_manager import ConfigManager
|
||||
|
||||
try:
|
||||
# Check if auto-update is enabled
|
||||
if not ConfigManager.get("auto_update_ytdlp"):
|
||||
return False
|
||||
|
||||
frequency: str = ConfigManager.get("auto_update_frequency") or "daily"
|
||||
last_check: float = ConfigManager.get("last_update_check") or 0
|
||||
current_time: float = time.time()
|
||||
|
||||
# Calculate time since last check
|
||||
time_diff: float = current_time - last_check
|
||||
|
||||
if frequency == "startup":
|
||||
# Always check on startup if we haven't checked in the last hour
|
||||
return time_diff > 3600 # 1 hour
|
||||
elif frequency == "daily":
|
||||
return time_diff > 86400 # 24 hours
|
||||
elif frequency == "weekly":
|
||||
return time_diff > 604800 # 7 days
|
||||
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.exception(f"Error checking auto-update schedule: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def check_and_update_ytdlp_auto() -> bool:
|
||||
"""Perform automatic yt-dlp update check and update if needed."""
|
||||
from ..utils.ytsage_config_manager import ConfigManager
|
||||
|
||||
try:
|
||||
logger.info("Performing automatic yt-dlp update check...")
|
||||
|
||||
# Get current version
|
||||
current_version = get_ytdlp_version()
|
||||
if "Error" in current_version:
|
||||
logger.info("Could not determine current yt-dlp version, skipping auto-update")
|
||||
return False
|
||||
|
||||
# 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"Current yt-dlp version: {current_version}")
|
||||
logger.info(f"Latest yt-dlp version: {latest_version}")
|
||||
|
||||
# Compare versions
|
||||
if version.parse(latest_version) > version.parse(current_version):
|
||||
logger.info(f"Auto-updating yt-dlp from {current_version} to {latest_version}...")
|
||||
|
||||
# Perform the update
|
||||
if update_yt_dlp():
|
||||
logger.info("Auto-update completed successfully!")
|
||||
# Update the last check timestamp
|
||||
ConfigManager.set("last_update_check", time.time())
|
||||
return True
|
||||
else:
|
||||
logger.info("Auto-update failed")
|
||||
return False
|
||||
else:
|
||||
logger.info("yt-dlp is already up to date")
|
||||
# Still update the timestamp even if no update was needed
|
||||
ConfigManager.set("last_update_check", time.time())
|
||||
return True
|
||||
|
||||
except requests.RequestException as e:
|
||||
logger.info(f"Network error during auto-update check: {e}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.exception(f"Error during auto-update check: {e}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.critical(f"Critical error in auto-update: {e}", exc_info=True)
|
||||
return False
|
||||
|
||||
|
||||
def get_auto_update_settings() -> Dict[str, Any]:
|
||||
"""Get current auto-update settings from config."""
|
||||
from ..utils.ytsage_config_manager import ConfigManager
|
||||
|
||||
enabled: Optional[bool] = ConfigManager.get("auto_update_ytdlp")
|
||||
frequency: Optional[str] = ConfigManager.get("auto_update_frequency")
|
||||
last_check: Optional[float] = ConfigManager.get("last_update_check")
|
||||
|
||||
return {
|
||||
"enabled": enabled if enabled is not None else True,
|
||||
"frequency": frequency if frequency is not None else "daily",
|
||||
"last_check": last_check if last_check is not None else 0,
|
||||
}
|
||||
|
||||
|
||||
def update_auto_update_settings(enabled: bool, frequency: str) -> bool:
|
||||
"""Update auto-update settings in config."""
|
||||
try:
|
||||
from ..utils.ytsage_config_manager import ConfigManager
|
||||
|
||||
ConfigManager.set("auto_update_ytdlp", enabled)
|
||||
ConfigManager.set("auto_update_frequency", frequency)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.exception(f"Error updating auto-update settings: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def parse_yt_dlp_error(error_message: str) -> str:
|
||||
"""
|
||||
Parse yt-dlp error messages and return user-friendly error messages.
|
||||
|
||||
Args:
|
||||
error_message: The raw error message from yt-dlp
|
||||
|
||||
Returns:
|
||||
str: A user-friendly error message with actionable advice
|
||||
"""
|
||||
error_str = error_message.lower()
|
||||
|
||||
# Private video errors
|
||||
if any(keyword in error_str for keyword in ["private video", "login_required", "sign in if you"]):
|
||||
return _("ytdlp_errors.private_video")
|
||||
|
||||
# Age-restricted content
|
||||
if any(keyword in error_str for keyword in ["age restricted", "age-restricted", "confirm your age"]):
|
||||
return _("ytdlp_errors.age_restricted")
|
||||
|
||||
# Geo-blocked content
|
||||
if any(
|
||||
keyword in error_str
|
||||
for keyword in [
|
||||
"not available in your country",
|
||||
"geo-blocked",
|
||||
"video is not available",
|
||||
"not made this video available in your country",
|
||||
]
|
||||
):
|
||||
return _("ytdlp_errors.geo_blocked")
|
||||
|
||||
# Removed/deleted videos
|
||||
if any(keyword in error_str for keyword in ["video unavailable", "this video has been removed", "video does not exist"]):
|
||||
return _("ytdlp_errors.video_unavailable")
|
||||
|
||||
# Live stream errors
|
||||
if any(keyword in error_str for keyword in ["live stream", "livestream", "is live"]):
|
||||
return _("ytdlp_errors.live_stream")
|
||||
|
||||
# Playlist errors
|
||||
if any(keyword in error_str for keyword in ["playlist", "no entries"]):
|
||||
return _("ytdlp_errors.playlist_error")
|
||||
|
||||
# Network/connection errors
|
||||
if any(keyword in error_str for keyword in ["network error", "connection", "timeout", "unable to download"]):
|
||||
return _("ytdlp_errors.network_error")
|
||||
|
||||
# Invalid URL
|
||||
if any(keyword in error_str for keyword in ["invalid url", "unsupported url", "no video found"]):
|
||||
return _("ytdlp_errors.invalid_url")
|
||||
|
||||
# YouTube premium content
|
||||
if any(keyword in error_str for keyword in ["youtube premium", "premium", "members only"]):
|
||||
return _("ytdlp_errors.premium_content")
|
||||
|
||||
# Copyright/DMCA
|
||||
if any(keyword in error_str for keyword in ["copyright", "dmca", "blocked"]):
|
||||
return _("ytdlp_errors.copyright_blocked")
|
||||
|
||||
# Extraction errors (could be temporary)
|
||||
if any(keyword in error_str for keyword in ["unable to extract", "extraction failed"]):
|
||||
return _("ytdlp_errors.extraction_failed")
|
||||
|
||||
# Generic fallback with the original error for debugging
|
||||
return _("ytdlp_errors.generic_error", error=error_message)
|
||||
|
||||
|
||||
def validate_video_url(url: str) -> tuple[bool, str]:
|
||||
"""
|
||||
Validate a video URL for supported platforms.
|
||||
|
||||
Args:
|
||||
url: The URL string to validate
|
||||
|
||||
Returns:
|
||||
tuple[bool, str]: (is_valid, error_message)
|
||||
- is_valid: True if URL is valid, False otherwise
|
||||
- error_message: Empty string if valid, error description if invalid
|
||||
|
||||
Example:
|
||||
>>> is_valid, error = validate_video_url("https://youtube.com/watch?v=xxx")
|
||||
>>> if not is_valid:
|
||||
... print(error)
|
||||
"""
|
||||
from urllib.parse import urlparse
|
||||
|
||||
# Check if URL is empty
|
||||
if not url or not url.strip():
|
||||
return False, _("url_validation.empty_url")
|
||||
|
||||
url = url.strip()
|
||||
|
||||
# Check basic URL structure
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
except Exception as e:
|
||||
logger.debug(f"URL parsing error: {e}")
|
||||
return False, _("url_validation.invalid_format")
|
||||
|
||||
# Check if scheme is http or https
|
||||
if parsed.scheme not in ['http', 'https']:
|
||||
return False, _("url_validation.invalid_scheme")
|
||||
|
||||
# Check if netloc (domain) exists
|
||||
if not parsed.netloc:
|
||||
return False, _("url_validation.missing_domain")
|
||||
|
||||
# YTSage focuses on YouTube and YouTube Music only
|
||||
# Supported YouTube domains
|
||||
youtube_domains = [
|
||||
'youtube.com',
|
||||
'www.youtube.com',
|
||||
'youtu.be',
|
||||
'm.youtube.com',
|
||||
'music.youtube.com', # YouTube Music
|
||||
'gaming.youtube.com', # YouTube Gaming (redirects to main)
|
||||
]
|
||||
|
||||
# Check if domain is YouTube
|
||||
netloc_lower = parsed.netloc.lower()
|
||||
is_youtube = any(
|
||||
netloc_lower == domain or netloc_lower.endswith('.' + domain)
|
||||
for domain in youtube_domains
|
||||
)
|
||||
|
||||
if not is_youtube:
|
||||
return False, _("url_validation.unsupported_platform", domain=parsed.netloc)
|
||||
|
||||
# Optional: Validate YouTube URL patterns
|
||||
# Common YouTube URL patterns:
|
||||
# - /watch?v=VIDEO_ID
|
||||
# - /playlist?list=PLAYLIST_ID
|
||||
# - /shorts/VIDEO_ID
|
||||
# - youtu.be/VIDEO_ID
|
||||
valid_patterns = [
|
||||
'/watch',
|
||||
'/playlist',
|
||||
'/shorts/',
|
||||
'/live/',
|
||||
'/channel/',
|
||||
'/c/',
|
||||
'/user/',
|
||||
'@', # New handle format
|
||||
]
|
||||
|
||||
# For youtu.be, the path itself is the video ID
|
||||
if 'youtu.be' in netloc_lower:
|
||||
if not parsed.path or parsed.path == '/':
|
||||
return False, _("url_validation.invalid_youtu_be")
|
||||
return True, ""
|
||||
|
||||
# For youtube.com domains, check for valid patterns
|
||||
if any(pattern in url.lower() for pattern in valid_patterns):
|
||||
return True, ""
|
||||
|
||||
# If it's a YouTube domain but doesn't match known patterns, still allow it
|
||||
# (yt-dlp might support formats we don't know about)
|
||||
logger.info(f"YouTube URL doesn't match known patterns but allowing: {url}")
|
||||
return True, ""
|
||||
|
||||
@@ -0,0 +1,714 @@
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import requests
|
||||
from PySide6.QtCore import Qt, QThread, Signal
|
||||
from PySide6.QtGui import QIcon
|
||||
from PySide6.QtWidgets import (
|
||||
QDialog,
|
||||
QFileDialog,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QMessageBox,
|
||||
QProgressBar,
|
||||
QPushButton,
|
||||
QRadioButton,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from ..utils.ytsage_logger import logger
|
||||
from ..utils.ytsage_constants import (
|
||||
APP_BIN_DIR,
|
||||
ICON_PATH,
|
||||
OS_FULL_NAME,
|
||||
OS_NAME,
|
||||
SUBPROCESS_CREATIONFLAGS,
|
||||
YTDLP_APP_BIN_PATH,
|
||||
YTDLP_DOWNLOAD_URL,
|
||||
YTDLP_SHA256_URL,
|
||||
)
|
||||
from .ytsage_ffmpeg import get_file_sha256
|
||||
from ..utils.ytsage_localization import _
|
||||
|
||||
# YTDLP_URLS moved to src\utils\ytsage_constants.py
|
||||
# get_ytdlp_install_dir() moved to src\utils\ytsage_constants.py
|
||||
# get_ytdlp_executable_path() moved to src\utils\ytsage_constants.py
|
||||
# get_os_type() moved to src\utils\ytsage_constants.py
|
||||
# ensure_install_dir_exists() moved to src\utils\ytsage_constants.py
|
||||
|
||||
|
||||
def verify_ytdlp_sha256(file_path: Path, download_url: str) -> bool:
|
||||
"""
|
||||
Verify yt-dlp file SHA256 hash against official checksums.
|
||||
|
||||
Args:
|
||||
file_path: Path to the downloaded yt-dlp file
|
||||
download_url: The URL used to download the file (to determine the filename)
|
||||
|
||||
Returns:
|
||||
bool: True if verification successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
# Download the SHA2-256SUMS file
|
||||
logger.info(f"Downloading SHA256 checksums from: {YTDLP_SHA256_URL}")
|
||||
response = requests.get(YTDLP_SHA256_URL, timeout=10)
|
||||
response.raise_for_status()
|
||||
checksum_content = response.text
|
||||
|
||||
# Extract filename from download URL (e.g., yt-dlp.exe, yt-dlp_macos, yt-dlp)
|
||||
filename = download_url.split("/")[-1]
|
||||
logger.info(f"Looking for checksum for file: {filename}")
|
||||
|
||||
# Parse the checksum file to find the matching hash
|
||||
expected_hash = None
|
||||
for line in checksum_content.strip().split("\n"):
|
||||
if filename in line:
|
||||
# Format: "hash filename"
|
||||
parts = line.strip().split()
|
||||
if len(parts) >= 2 and parts[1] == filename:
|
||||
expected_hash = parts[0]
|
||||
break
|
||||
|
||||
if not expected_hash:
|
||||
logger.error(f"Could not find SHA256 hash for {filename} in checksums file")
|
||||
return False
|
||||
|
||||
# Calculate actual hash of downloaded file
|
||||
logger.info("Calculating SHA256 hash of downloaded file...")
|
||||
actual_hash = get_file_sha256(file_path)
|
||||
|
||||
# Compare hashes
|
||||
if actual_hash.lower() == expected_hash.lower():
|
||||
logger.info("✓ SHA256 verification successful!")
|
||||
logger.info(f" Expected: {expected_hash}")
|
||||
logger.info(f" Actual: {actual_hash}")
|
||||
return True
|
||||
else:
|
||||
logger.error("✗ SHA256 verification failed!")
|
||||
logger.error(f" Expected: {expected_hash}")
|
||||
logger.error(f" Actual: {actual_hash}")
|
||||
return False
|
||||
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"Failed to download SHA256 checksums: {e}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.exception(f"Error during SHA256 verification: {e}")
|
||||
return False
|
||||
|
||||
|
||||
class DownloadYtdlpThread(QThread):
|
||||
progress_signal = Signal(int)
|
||||
finished_signal = Signal(bool, str)
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
|
||||
def run(self) -> None:
|
||||
try:
|
||||
# Extra logic moved to src\utils\ytsage_constants.py
|
||||
exe_path = YTDLP_APP_BIN_PATH
|
||||
|
||||
# Download with progress reporting
|
||||
logger.info(f"Downloading yt-dlp from: {YTDLP_DOWNLOAD_URL}")
|
||||
response = requests.get(YTDLP_DOWNLOAD_URL, stream=True)
|
||||
response.raise_for_status()
|
||||
total_size = int(response.headers.get("content-length", 0))
|
||||
block_size = 1024 # 1 Kibibyte
|
||||
|
||||
if total_size == 0:
|
||||
self.progress_signal.emit(100)
|
||||
|
||||
with open(exe_path, "wb") as f:
|
||||
downloaded = 0
|
||||
for data in response.iter_content(block_size):
|
||||
f.write(data)
|
||||
downloaded += len(data)
|
||||
if total_size > 0:
|
||||
progress = int(downloaded / total_size * 100)
|
||||
self.progress_signal.emit(progress)
|
||||
|
||||
logger.info("Download complete, verifying SHA256 hash...")
|
||||
|
||||
# Verify SHA256 hash
|
||||
if not verify_ytdlp_sha256(exe_path, YTDLP_DOWNLOAD_URL):
|
||||
# Hash verification failed - delete the downloaded file
|
||||
logger.error("SHA256 verification failed! Removing downloaded file.")
|
||||
if Path(exe_path).exists():
|
||||
Path(exe_path).unlink()
|
||||
self.finished_signal.emit(
|
||||
False,
|
||||
"SHA256 verification failed. The downloaded file may be corrupted or tampered with."
|
||||
)
|
||||
return
|
||||
|
||||
# Make executable on macOS and Linux
|
||||
if OS_NAME != "Windows":
|
||||
os.chmod(exe_path, 0o755)
|
||||
|
||||
logger.info("yt-dlp downloaded and verified successfully!")
|
||||
self.finished_signal.emit(True, str(exe_path))
|
||||
|
||||
except Exception as e:
|
||||
logger.exception(f"Error downloading yt-dlp: {e}")
|
||||
self.finished_signal.emit(False, str(e))
|
||||
|
||||
|
||||
class YtdlpSetupDialog(QDialog):
|
||||
setup_complete = Signal(str) # Signal emitting the path to yt-dlp
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle(_("ytdlp_setup.required_title"))
|
||||
self.setMinimumWidth(520)
|
||||
self.setMinimumHeight(350)
|
||||
self.resize(520, 380)
|
||||
|
||||
# Set the window icon to match the main app
|
||||
if parent and parent.windowIcon():
|
||||
self.setWindowIcon(parent.windowIcon())
|
||||
else:
|
||||
# icon_path logic moved to src\utils\ytsage_constants.py
|
||||
icon_path = ICON_PATH
|
||||
if Path.exists(icon_path):
|
||||
self.setWindowIcon(QIcon(str(icon_path)))
|
||||
|
||||
self.init_ui()
|
||||
|
||||
# Apply dark theme styling to match app
|
||||
self.setStyleSheet(
|
||||
"""
|
||||
QDialog {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel {
|
||||
color: #cccccc;
|
||||
line-height: 1.4;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 10px 20px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 6px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
font-size: 13px;
|
||||
min-width: 120px;
|
||||
min-height: 20px;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
QPushButton:pressed {
|
||||
background-color: #800000;
|
||||
}
|
||||
QPushButton:disabled {
|
||||
background-color: #666666;
|
||||
color: #999999;
|
||||
}
|
||||
QProgressBar {
|
||||
border: 2px solid #1d1e22;
|
||||
border-radius: 6px;
|
||||
text-align: center;
|
||||
color: white;
|
||||
background-color: #1d1e22;
|
||||
height: 25px;
|
||||
font-weight: bold;
|
||||
}
|
||||
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;
|
||||
}
|
||||
QRadioButton {
|
||||
color: #ffffff;
|
||||
spacing: 10px;
|
||||
padding: 8px;
|
||||
font-size: 13px;
|
||||
}
|
||||
QRadioButton::indicator {
|
||||
width: 18px;
|
||||
height: 18px;
|
||||
border-radius: 9px;
|
||||
}
|
||||
QRadioButton::indicator:unchecked {
|
||||
border: 2px solid #666666;
|
||||
background: #1d1e22;
|
||||
}
|
||||
QRadioButton::indicator:checked {
|
||||
border: 2px solid #c90000;
|
||||
background: #c90000;
|
||||
}
|
||||
"""
|
||||
)
|
||||
|
||||
def init_ui(self) -> None:
|
||||
layout = QVBoxLayout()
|
||||
layout.setSpacing(15)
|
||||
layout.setContentsMargins(25, 25, 25, 25)
|
||||
|
||||
# Header title
|
||||
title_label = QLabel(_("ytdlp_setup.required_title"))
|
||||
title_label.setStyleSheet("font-size: 16px; font-weight: bold; color: #ffffff; padding: 5px 0;")
|
||||
title_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
layout.addWidget(title_label)
|
||||
|
||||
# Information label with improved styling
|
||||
# os_name logic moved to src\utils\ytsage_constants.py
|
||||
|
||||
info_label = QLabel(
|
||||
_("ytdlp_setup.description", os_name=OS_FULL_NAME)
|
||||
)
|
||||
info_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
info_label.setWordWrap(True)
|
||||
info_label.setStyleSheet("font-size: 13px; color: #cccccc; padding: 5px; line-height: 1.4;")
|
||||
layout.addWidget(info_label)
|
||||
|
||||
# Radio buttons with minimal spacing
|
||||
option_widget = QWidget()
|
||||
option_layout = QVBoxLayout(option_widget)
|
||||
option_layout.setSpacing(8)
|
||||
option_layout.setContentsMargins(0, 0, 0, 0)
|
||||
|
||||
self.auto_radio = QRadioButton(_("ytdlp_setup.option_auto"))
|
||||
self.auto_radio.setChecked(True)
|
||||
self.manual_radio = QRadioButton(_("ytdlp_setup.option_manual"))
|
||||
|
||||
option_layout.addWidget(self.auto_radio)
|
||||
option_layout.addWidget(self.manual_radio)
|
||||
layout.addWidget(option_widget)
|
||||
|
||||
# Progress bar with proper sizing
|
||||
self.progress_bar = QProgressBar()
|
||||
self.progress_bar.setVisible(False)
|
||||
self.progress_bar.setFixedHeight(20) # Fixed height for consistency
|
||||
self.progress_bar.setStyleSheet(
|
||||
"""
|
||||
QProgressBar {
|
||||
border: 1px solid #3d3d3d;
|
||||
border-radius: 8px;
|
||||
background-color: #1d1e22;
|
||||
text-align: center;
|
||||
color: #ffffff;
|
||||
font-size: 12px;
|
||||
font-weight: bold;
|
||||
height: 20px;
|
||||
}
|
||||
QProgressBar::chunk {
|
||||
background-color: #c90000;
|
||||
border-radius: 6px;
|
||||
margin: 1px;
|
||||
}
|
||||
"""
|
||||
)
|
||||
layout.addWidget(self.progress_bar)
|
||||
|
||||
# Status label with better spacing
|
||||
self.status_label = QLabel("")
|
||||
self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
self.status_label.setStyleSheet("font-size: 12px; color: #cccccc; padding: 8px 0;")
|
||||
self.status_label.setWordWrap(True)
|
||||
layout.addWidget(self.status_label)
|
||||
|
||||
# Add stretch to push buttons to bottom
|
||||
layout.addStretch()
|
||||
|
||||
# Button layout with improved spacing
|
||||
button_layout = QHBoxLayout()
|
||||
button_layout.setSpacing(15)
|
||||
button_layout.setContentsMargins(0, 10, 0, 0) # Add top margin for buttons
|
||||
|
||||
self.setup_button = QPushButton(_("ytdlp_setup.setup_button"))
|
||||
self.setup_button.clicked.connect(self.setup_ytdlp)
|
||||
|
||||
self.cancel_button = QPushButton(_("buttons.cancel"))
|
||||
self.cancel_button.clicked.connect(self.reject)
|
||||
|
||||
button_layout.addWidget(self.setup_button)
|
||||
button_layout.addWidget(self.cancel_button)
|
||||
layout.addLayout(button_layout)
|
||||
|
||||
self.setLayout(layout)
|
||||
|
||||
def setup_ytdlp(self) -> None:
|
||||
if self.auto_radio.isChecked():
|
||||
self.download_ytdlp()
|
||||
else:
|
||||
self.select_ytdlp_path()
|
||||
|
||||
def download_ytdlp(self) -> None:
|
||||
self.progress_bar.setVisible(True)
|
||||
self.progress_bar.setValue(0)
|
||||
self.status_label.setText(_("ytdlp_setup.downloading"))
|
||||
self.setup_button.setEnabled(False)
|
||||
self.cancel_button.setEnabled(False)
|
||||
|
||||
self.download_thread = DownloadYtdlpThread()
|
||||
self.download_thread.progress_signal.connect(self.update_progress)
|
||||
self.download_thread.finished_signal.connect(self.download_finished)
|
||||
self.download_thread.start()
|
||||
|
||||
def update_progress(self, value) -> None:
|
||||
self.progress_bar.setValue(value)
|
||||
|
||||
def download_finished(self, success, result) -> None:
|
||||
self.setup_button.setEnabled(True)
|
||||
self.cancel_button.setEnabled(True)
|
||||
|
||||
if success:
|
||||
self.status_label.setText(_("ytdlp_setup.success"))
|
||||
self.setup_complete.emit(result)
|
||||
self.accept()
|
||||
else:
|
||||
self.status_label.setText(_("ytdlp_setup.error", error=result))
|
||||
error_dialog = QMessageBox(self)
|
||||
error_dialog.setIcon(QMessageBox.Icon.Critical)
|
||||
error_dialog.setWindowTitle(_("ytdlp_setup.download_failed_title"))
|
||||
error_dialog.setText(_("ytdlp_setup.download_failed_message", error=result))
|
||||
# Set the window icon to match the main dialog
|
||||
error_dialog.setWindowIcon(self.windowIcon())
|
||||
error_dialog.setStyleSheet(
|
||||
"""
|
||||
QMessageBox {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel {
|
||||
color: #ffffff;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 8px 15px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
"""
|
||||
)
|
||||
error_dialog.exec()
|
||||
|
||||
def select_ytdlp_path(self) -> None:
|
||||
if OS_NAME == "Windows":
|
||||
file_filter = _("ytdlp_setup.file_filter_windows")
|
||||
else:
|
||||
file_filter = _("ytdlp_setup.file_filter_all")
|
||||
|
||||
# Apply style to QFileDialog
|
||||
file_dialog = QFileDialog(self)
|
||||
file_dialog.setStyleSheet(
|
||||
"""
|
||||
QFileDialog {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel, QCheckBox, QListView, QTreeView, QComboBox, QLineEdit {
|
||||
color: #ffffff;
|
||||
background-color: #1b2021;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 8px 15px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
"""
|
||||
)
|
||||
|
||||
file_path, _ = file_dialog.getOpenFileName(
|
||||
self, _("ytdlp_setup.select_executable_title"), "", file_filter
|
||||
)
|
||||
|
||||
if file_path:
|
||||
logger.debug(f"User selected file: {file_path}")
|
||||
# Verify the selected file
|
||||
try:
|
||||
# Extra logic moved to src\utils\ytsage_constants.py
|
||||
# Try to run yt-dlp --version
|
||||
logger.debug(f"Verifying file with --version command")
|
||||
result = subprocess.run(
|
||||
[file_path, "--version"], capture_output=True, text=True, check=False, creationflags=SUBPROCESS_CREATIONFLAGS
|
||||
)
|
||||
logger.debug(f"Version check result: {result.returncode}, Output: {result.stdout.strip()}")
|
||||
|
||||
if result.returncode == 0:
|
||||
# File is valid, copy it to our app's bin directory
|
||||
try:
|
||||
# Ensure the bin directory exists
|
||||
logger.debug(f"Install directory: {APP_BIN_DIR}")
|
||||
|
||||
# Determine the target filename based on OS
|
||||
target_path = YTDLP_APP_BIN_PATH
|
||||
logger.debug(f"Target path: {target_path}")
|
||||
|
||||
# Copy the file
|
||||
shutil.copy2(file_path, target_path)
|
||||
logger.debug(f"File copied successfully")
|
||||
|
||||
# Set executable permissions on Unix systems
|
||||
if OS_NAME != "Windows":
|
||||
os.chmod(target_path, 0o755)
|
||||
logger.debug(f"Permissions set on Unix system")
|
||||
|
||||
# Return the path of the copied file
|
||||
self.status_label.setText(_("ytdlp_setup.copied_to", path=target_path))
|
||||
logger.debug(f"Emitting setup_complete signal with path: {target_path}")
|
||||
self.setup_complete.emit(target_path)
|
||||
self.accept()
|
||||
except Exception as copy_error:
|
||||
logger.debug(f"Error copying file: {copy_error}", exc_info=True)
|
||||
error_dialog = QMessageBox(self)
|
||||
error_dialog.setIcon(QMessageBox.Icon.Critical)
|
||||
error_dialog.setWindowTitle(_("ytdlp_setup.setup_error_title"))
|
||||
error_dialog.setText(_("ytdlp_setup.copy_error", error=copy_error))
|
||||
error_dialog.setStyleSheet(
|
||||
"""
|
||||
QMessageBox {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel {
|
||||
color: #ffffff;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 8px 15px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
"""
|
||||
)
|
||||
error_dialog.exec()
|
||||
else:
|
||||
logger.debug(f"File verification failed with return code: {result.returncode}")
|
||||
error_dialog = QMessageBox(self)
|
||||
error_dialog.setIcon(QMessageBox.Icon.Warning)
|
||||
error_dialog.setWindowTitle(_("ytdlp_setup.invalid_executable_title"))
|
||||
error_dialog.setText(_("ytdlp_setup.invalid_executable_message"))
|
||||
error_dialog.setStyleSheet(
|
||||
"""
|
||||
QMessageBox {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel {
|
||||
color: #ffffff;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 8px 15px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
"""
|
||||
)
|
||||
error_dialog.exec()
|
||||
except Exception as e:
|
||||
logger.debug(f"Exception during verification: {e}", exc_info=True)
|
||||
error_dialog = QMessageBox(self)
|
||||
error_dialog.setIcon(QMessageBox.Icon.Critical)
|
||||
error_dialog.setWindowTitle(_("main_ui.error_title"))
|
||||
error_dialog.setText(_("ytdlp_setup.verify_error", error=e))
|
||||
error_dialog.setStyleSheet(
|
||||
"""
|
||||
QMessageBox {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel {
|
||||
color: #ffffff;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 8px 15px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
"""
|
||||
)
|
||||
error_dialog.exec()
|
||||
|
||||
|
||||
def check_ytdlp_binary() -> Optional[Path]:
|
||||
"""
|
||||
Check if yt-dlp binary exists in the app's bin directory ONLY.
|
||||
We now ignore system PATH and only use our managed binary.
|
||||
Returns:
|
||||
Path or None: Path to yt-dlp binary if found in app bin, None otherwise
|
||||
"""
|
||||
exe_path = YTDLP_APP_BIN_PATH
|
||||
if exe_path.exists():
|
||||
# Make sure it's executable on Unix systems
|
||||
if OS_NAME != "Windows" and not os.access(exe_path, os.X_OK):
|
||||
try:
|
||||
os.chmod(exe_path, 0o755)
|
||||
logger.info(f"Fixed permissions on yt-dlp at {exe_path}")
|
||||
except Exception as e:
|
||||
logger.exception(f"Could not set executable permissions on {exe_path}: {e}")
|
||||
logger.info(f"Found yt-dlp in app bin directory: {exe_path}")
|
||||
return exe_path
|
||||
|
||||
# Binary not found in app directory - return None to trigger setup
|
||||
logger.warning(f"yt-dlp binary not found in app bin directory: {exe_path}")
|
||||
return None
|
||||
|
||||
|
||||
def check_ytdlp_installed() -> bool:
|
||||
"""
|
||||
Check if yt-dlp is installed and accessible.
|
||||
Returns:
|
||||
bool: True if yt-dlp is found and working, False otherwise
|
||||
"""
|
||||
try:
|
||||
ytdlp_path = check_ytdlp_binary()
|
||||
if ytdlp_path:
|
||||
# Try to run yt-dlp --version to verify it's working
|
||||
try:
|
||||
# Extra logic moved to src\utils\ytsage_constants.py
|
||||
result = subprocess.run(
|
||||
[ytdlp_path, "--version"], capture_output=True, text=True, timeout=5, creationflags=SUBPROCESS_CREATIONFLAGS
|
||||
)
|
||||
return result.returncode == 0
|
||||
except Exception:
|
||||
return False
|
||||
return False
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def get_yt_dlp_path() -> Path:
|
||||
"""
|
||||
Get the yt-dlp path, either from the app's bin directory or system PATH.
|
||||
This replaces the function in ytsage_utils.py.
|
||||
Returns:
|
||||
str: Path to yt-dlp binary
|
||||
"""
|
||||
# First check if we have yt-dlp in our app's bin directory or system PATH
|
||||
ytdlp_path = check_ytdlp_binary()
|
||||
if ytdlp_path:
|
||||
logger.info(f"Using yt-dlp from: {ytdlp_path}")
|
||||
return ytdlp_path
|
||||
|
||||
# If not found anywhere, fall back to the command name as a last resort
|
||||
logger.info("yt-dlp not found in app directory or PATH, falling back to command name")
|
||||
return "yt-dlp" # type: ignore[return-value]
|
||||
|
||||
|
||||
def setup_ytdlp(parent_widget=None):
|
||||
"""
|
||||
Show the yt-dlp setup dialog and handle the result.
|
||||
Returns:
|
||||
str: Path to yt-dlp binary
|
||||
"""
|
||||
logger.debug("Starting yt-dlp setup dialog")
|
||||
dialog = YtdlpSetupDialog(parent_widget)
|
||||
|
||||
# Store the setup result from the signal
|
||||
setup_result = {"path": None}
|
||||
|
||||
def on_setup_complete(path) -> None:
|
||||
logger.debug(f"Received setup_complete signal with path: {path}")
|
||||
setup_result["path"] = path
|
||||
|
||||
# Connect to the setup_complete signal
|
||||
dialog.setup_complete.connect(on_setup_complete)
|
||||
|
||||
# Show the dialog
|
||||
result = dialog.exec()
|
||||
logger.debug(f"Dialog result: {result} (Accepted={QDialog.DialogCode.Accepted})")
|
||||
|
||||
if result == QDialog.DialogCode.Accepted:
|
||||
# First check if we received a path from the signal
|
||||
if setup_result["path"]:
|
||||
path_obj = Path(setup_result["path"]) if isinstance(setup_result["path"], str) else setup_result["path"]
|
||||
if path_obj.exists():
|
||||
logger.debug(f"Using path from signal: {setup_result['path']}")
|
||||
return str(setup_result["path"])
|
||||
|
||||
# Get the expected path for verification as fallback
|
||||
expected_path = YTDLP_APP_BIN_PATH
|
||||
logger.debug(f"Expected yt-dlp path: {expected_path}")
|
||||
|
||||
# Verify the path exists after dialog is accepted
|
||||
if expected_path.exists():
|
||||
logger.debug(f"yt-dlp successfully found at expected path: {expected_path}")
|
||||
return str(expected_path)
|
||||
else:
|
||||
logger.debug(f"Expected path does not exist, trying alternate detection")
|
||||
# Try to use the get_yt_dlp_path function to find yt-dlp elsewhere
|
||||
yt_dlp_path = get_yt_dlp_path()
|
||||
logger.debug(f"Alternate detection result: {yt_dlp_path}")
|
||||
if yt_dlp_path != "yt-dlp":
|
||||
path_obj = Path(yt_dlp_path) if isinstance(yt_dlp_path, str) else yt_dlp_path
|
||||
if path_obj.exists():
|
||||
logger.debug(f"yt-dlp found at alternate location: {yt_dlp_path}")
|
||||
return str(yt_dlp_path)
|
||||
|
||||
# Something went wrong, show an error message
|
||||
logger.debug(f"Setup failed, showing error dialog")
|
||||
if parent_widget:
|
||||
error_dialog = QMessageBox(parent_widget)
|
||||
error_dialog.setIcon(QMessageBox.Icon.Warning)
|
||||
error_dialog.setWindowTitle(_("ytdlp_setup.setup_failed_title"))
|
||||
error_dialog.setText(_("ytdlp_setup.setup_failed_message"))
|
||||
# Set the window icon to match the parent
|
||||
error_dialog.setWindowIcon(parent_widget.windowIcon())
|
||||
error_dialog.setStyleSheet(
|
||||
"""
|
||||
QMessageBox {
|
||||
background-color: #15181b;
|
||||
color: #ffffff;
|
||||
}
|
||||
QLabel {
|
||||
color: #ffffff;
|
||||
}
|
||||
QPushButton {
|
||||
padding: 8px 15px;
|
||||
background-color: #c90000;
|
||||
border: none;
|
||||
border-radius: 4px;
|
||||
color: white;
|
||||
font-weight: bold;
|
||||
}
|
||||
QPushButton:hover {
|
||||
background-color: #a50000;
|
||||
}
|
||||
"""
|
||||
)
|
||||
error_dialog.exec()
|
||||
logger.warning(f"yt-dlp setup failed, path does not exist: {expected_path}")
|
||||
else:
|
||||
logger.debug("User cancelled the setup dialog")
|
||||
|
||||
# User cancelled or setup failed, return the fallback command
|
||||
logger.debug("Returning fallback command 'yt-dlp'")
|
||||
return "yt-dlp"
|
||||
Reference in New Issue
Block a user