1256 lines
50 KiB
Python
1256 lines
50 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
import subprocess
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Set
|
||
|
||
from PyQt6.QtCore import QEventLoop, QSettings, QSize, Qt, QTimer, QUrl
|
||
from PyQt6.QtGui import QAction, QBrush, QColor, QDesktopServices, QKeySequence, QShortcut
|
||
from PyQt6.QtWidgets import (
|
||
QApplication,
|
||
QAbstractItemView,
|
||
QFileDialog,
|
||
QFrame,
|
||
QHBoxLayout,
|
||
QHeaderView,
|
||
QLabel,
|
||
QMainWindow,
|
||
QMenu,
|
||
QMessageBox,
|
||
QProgressBar,
|
||
QPushButton,
|
||
QSizePolicy,
|
||
QStackedWidget,
|
||
QToolButton,
|
||
QTreeWidget,
|
||
QTreeWidgetItem,
|
||
QVBoxLayout,
|
||
QWidget,
|
||
)
|
||
|
||
from .media import (
|
||
available_video_codecs,
|
||
classify_path,
|
||
find_ffmpeg_tools,
|
||
format_duration,
|
||
human_size,
|
||
probe_media,
|
||
)
|
||
from .models import (
|
||
IMAGE_EXTENSIONS,
|
||
VIDEO_EXTENSIONS,
|
||
ImageOptions,
|
||
MediaItem,
|
||
MediaKind,
|
||
VideoOptions,
|
||
)
|
||
from .processing import CompressionJob, CompressionWorker
|
||
from .settings_panel import SettingsPanel
|
||
from .theme import ACCENT, AMBER, GREEN, MUTED, RED, app_logo_pixmap, make_app_icon, media_icon
|
||
from .widgets import DropZone
|
||
|
||
APP_NAME = "Bolt Compress"
|
||
OUTPUT_SUFFIX = "-compressed"
|
||
|
||
_IMAGE_PATTERN = " ".join(f"*{suffix}" for suffix in sorted(IMAGE_EXTENSIONS))
|
||
_VIDEO_PATTERN = " ".join(f"*{suffix}" for suffix in sorted(VIDEO_EXTENSIONS))
|
||
_MEDIA_FILTER = (
|
||
f"Media files ({_IMAGE_PATTERN} {_VIDEO_PATTERN});;"
|
||
f"Images ({_IMAGE_PATTERN});;"
|
||
f"Videos ({_VIDEO_PATTERN});;"
|
||
"All files (*)"
|
||
)
|
||
|
||
|
||
def _open_path(path: Path) -> None:
|
||
QDesktopServices.openUrl(QUrl.fromLocalFile(str(path)))
|
||
|
||
|
||
def _reveal_path(path: Path) -> None:
|
||
if os.name == "nt":
|
||
try:
|
||
subprocess.Popen(
|
||
["explorer", f"/select,{os.path.normpath(str(path))}"],
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||
)
|
||
return
|
||
except OSError:
|
||
pass
|
||
_open_path(path.parent)
|
||
|
||
|
||
def _one_line(text: str, limit: int = 160) -> str:
|
||
cleaned = " ".join((text or "").split())
|
||
return cleaned if len(cleaned) <= limit else cleaned[: limit - 1] + "…"
|
||
|
||
|
||
class MainWindow(QMainWindow):
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.setWindowTitle(APP_NAME)
|
||
self.setWindowIcon(make_app_icon())
|
||
self.setMinimumSize(1040, 680)
|
||
self.resize(1200, 800)
|
||
self.setAcceptDrops(True)
|
||
|
||
self._settings = QSettings("Bolt", "Bolt Compress")
|
||
self._tools = find_ffmpeg_tools()
|
||
self._available_codecs = available_video_codecs(self._tools)
|
||
|
||
self._items: Dict[str, MediaItem] = {}
|
||
self._order: List[str] = []
|
||
self._rows: Dict[str, QTreeWidgetItem] = {}
|
||
self._seen: Set[str] = set()
|
||
self._output_dir: Optional[Path] = None
|
||
self._last_dir = str(Path.home())
|
||
self._worker: Optional[CompressionWorker] = None
|
||
self._running = False
|
||
self._current_id: Optional[str] = None
|
||
self._progress_visible = False
|
||
self._drag_active = False
|
||
self._last_run_summary: Optional[dict] = None
|
||
self._scanning = False
|
||
|
||
self._save_timer = QTimer(self)
|
||
self._save_timer.setSingleShot(True)
|
||
self._save_timer.setInterval(450)
|
||
self._save_timer.timeout.connect(self._save_preferences)
|
||
|
||
self._build_ui()
|
||
self._install_shortcuts()
|
||
self._load_preferences()
|
||
self._update_engine_status()
|
||
self._refresh_ui()
|
||
|
||
# ------------------------------------------------------------------ UI
|
||
|
||
def _build_ui(self) -> None:
|
||
root = QWidget()
|
||
root.setObjectName("root")
|
||
self.setCentralWidget(root)
|
||
outer = QVBoxLayout(root)
|
||
outer.setContentsMargins(24, 18, 24, 12)
|
||
outer.setSpacing(14)
|
||
|
||
outer.addLayout(self._build_header())
|
||
|
||
body = QHBoxLayout()
|
||
body.setSpacing(16)
|
||
self.queue_card = self._build_queue_card()
|
||
body.addWidget(self.queue_card, 1)
|
||
|
||
self.settings_panel = SettingsPanel(self._available_codecs)
|
||
self.settings_panel.setFixedWidth(372)
|
||
self.settings_panel.settingsChanged.connect(self._on_settings_changed)
|
||
body.addWidget(self.settings_panel)
|
||
outer.addLayout(body, 1)
|
||
|
||
outer.addLayout(self._build_status_footer())
|
||
|
||
def _build_header(self) -> QHBoxLayout:
|
||
layout = QHBoxLayout()
|
||
layout.setSpacing(12)
|
||
layout.setContentsMargins(2, 0, 2, 0)
|
||
|
||
logo = QLabel()
|
||
screen = QApplication.primaryScreen()
|
||
dpr = screen.devicePixelRatio() if screen else 1.0
|
||
logo_pixmap = app_logo_pixmap(max(46, int(46 * dpr)))
|
||
logo_pixmap.setDevicePixelRatio(dpr)
|
||
logo.setPixmap(logo_pixmap)
|
||
logo.setFixedSize(46, 46)
|
||
layout.addWidget(logo)
|
||
|
||
titles = QVBoxLayout()
|
||
titles.setSpacing(1)
|
||
title = QLabel(APP_NAME)
|
||
title.setObjectName("appTitle")
|
||
titles.addWidget(title)
|
||
subtitle = QLabel("Compress and resize video & images, privately on your machine.")
|
||
subtitle.setObjectName("appSubtitle")
|
||
titles.addWidget(subtitle)
|
||
layout.addLayout(titles)
|
||
layout.addStretch(1)
|
||
|
||
privacy_chip = QLabel("● 100% local")
|
||
privacy_chip.setObjectName("privacyChip")
|
||
privacy_chip.setToolTip("Files are processed on this computer. Nothing is uploaded.")
|
||
layout.addWidget(privacy_chip)
|
||
return layout
|
||
|
||
def _build_queue_card(self) -> QFrame:
|
||
card = QFrame()
|
||
card.setObjectName("queueCard")
|
||
card.setProperty("dragActive", False)
|
||
|
||
layout = QVBoxLayout(card)
|
||
layout.setContentsMargins(18, 16, 18, 16)
|
||
layout.setSpacing(12)
|
||
|
||
header = QHBoxLayout()
|
||
header.setSpacing(8)
|
||
title = QLabel("Queue")
|
||
title.setObjectName("cardTitle")
|
||
header.addWidget(title)
|
||
self.count_badge = QLabel("0")
|
||
self.count_badge.setObjectName("badge")
|
||
self.count_badge.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||
header.addWidget(self.count_badge)
|
||
header.addStretch(1)
|
||
|
||
self.add_files_button = QPushButton("Add files")
|
||
self.add_files_button.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
self.add_files_button.setToolTip("Add videos and images (Ctrl+O)")
|
||
self.add_files_button.clicked.connect(self._choose_files)
|
||
header.addWidget(self.add_files_button)
|
||
|
||
self.add_folder_button = QPushButton("Add folder")
|
||
self.add_folder_button.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
self.add_folder_button.setToolTip("Add supported files from a folder (Ctrl+Shift+O)")
|
||
self.add_folder_button.clicked.connect(self._choose_folder)
|
||
header.addWidget(self.add_folder_button)
|
||
|
||
self.clear_button = QPushButton("Clear")
|
||
self.clear_button.setObjectName("ghostButton")
|
||
self.clear_button.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
self.clear_button.setToolTip("Remove all files from the queue")
|
||
self.clear_button.clicked.connect(self._clear_queue)
|
||
header.addWidget(self.clear_button)
|
||
layout.addLayout(header)
|
||
|
||
self.queue_stack = QStackedWidget()
|
||
self.queue_stack.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
|
||
empty_page = QWidget()
|
||
empty_layout = QVBoxLayout(empty_page)
|
||
empty_layout.setContentsMargins(0, 0, 0, 0)
|
||
self.drop_zone = DropZone()
|
||
self.drop_zone.browseRequested.connect(self._choose_files)
|
||
empty_layout.addWidget(self.drop_zone)
|
||
self.queue_stack.addWidget(empty_page)
|
||
|
||
queue_page = QWidget()
|
||
queue_layout = QVBoxLayout(queue_page)
|
||
queue_layout.setContentsMargins(0, 0, 0, 0)
|
||
self.queue_view = QTreeWidget()
|
||
self.queue_view.setObjectName("mediaQueue")
|
||
self.queue_view.setColumnCount(5)
|
||
self.queue_view.setHeaderLabels(["FILE", "DETAILS", "BEFORE", "AFTER", "STATUS"])
|
||
self.queue_view.setRootIsDecorated(False)
|
||
self.queue_view.setUniformRowHeights(True)
|
||
self.queue_view.setAlternatingRowColors(False)
|
||
self.queue_view.setSelectionMode(QAbstractItemView.SelectionMode.ExtendedSelection)
|
||
self.queue_view.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
|
||
self.queue_view.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
|
||
self.queue_view.setIndentation(0)
|
||
self.queue_view.setIconSize(QSize(30, 30))
|
||
self.queue_view.setTextElideMode(Qt.TextElideMode.ElideMiddle)
|
||
self.queue_view.setDragDropMode(QAbstractItemView.DragDropMode.NoDragDrop)
|
||
self.queue_view.setAcceptDrops(False)
|
||
self.queue_view.viewport().setAcceptDrops(False)
|
||
self.queue_view.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu)
|
||
self.queue_view.customContextMenuRequested.connect(self._show_queue_menu)
|
||
self.queue_view.itemDoubleClicked.connect(self._open_queue_item)
|
||
header_view = self.queue_view.header()
|
||
header_view.setStretchLastSection(False)
|
||
header_view.setSectionsClickable(False)
|
||
header_view.setHighlightSections(False)
|
||
header_view.setDefaultAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter)
|
||
header_view.setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch)
|
||
for column, width in ((1, 155), (2, 76), (3, 76), (4, 90)):
|
||
header_view.setSectionResizeMode(column, QHeaderView.ResizeMode.Fixed)
|
||
self.queue_view.setColumnWidth(column, width)
|
||
queue_layout.addWidget(self.queue_view)
|
||
self.queue_stack.addWidget(queue_page)
|
||
layout.addWidget(self.queue_stack, 1)
|
||
|
||
divider = QFrame()
|
||
divider.setObjectName("divider")
|
||
divider.setFrameShape(QFrame.Shape.HLine)
|
||
layout.addWidget(divider)
|
||
|
||
self.stats_label = QLabel("No files yet")
|
||
self.stats_label.setObjectName("statsLabel")
|
||
layout.addWidget(self.stats_label)
|
||
|
||
output_row = QHBoxLayout()
|
||
output_row.setSpacing(8)
|
||
output_label = QLabel("OUTPUT")
|
||
output_label.setObjectName("outputLabel")
|
||
output_row.addWidget(output_label)
|
||
self.dest_button = QToolButton()
|
||
self.dest_button.setObjectName("destButton")
|
||
self.dest_button.setToolButtonStyle(Qt.ToolButtonStyle.ToolButtonTextOnly)
|
||
self.dest_button.setPopupMode(QToolButton.ToolButtonPopupMode.InstantPopup)
|
||
self.dest_button.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
self.dest_button.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
||
self._build_destination_menu()
|
||
output_row.addWidget(self.dest_button, 1)
|
||
layout.addLayout(output_row)
|
||
|
||
self.progress_wrap = QWidget()
|
||
progress_layout = QHBoxLayout(self.progress_wrap)
|
||
progress_layout.setContentsMargins(0, 4, 0, 0)
|
||
progress_layout.setSpacing(10)
|
||
self.progress_bar = QProgressBar()
|
||
self.progress_bar.setRange(0, 100)
|
||
self.progress_bar.setValue(0)
|
||
self.progress_bar.setTextVisible(False)
|
||
progress_layout.addWidget(self.progress_bar, 1)
|
||
self.percent_label = QLabel("0%")
|
||
self.percent_label.setObjectName("percentLabel")
|
||
self.percent_label.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
|
||
progress_layout.addWidget(self.percent_label)
|
||
self.progress_wrap.setVisible(False)
|
||
layout.addWidget(self.progress_wrap)
|
||
|
||
actions = QHBoxLayout()
|
||
actions.setSpacing(10)
|
||
self.compress_button = QPushButton("Compress")
|
||
self.compress_button.setObjectName("primaryAction")
|
||
self.compress_button.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
self.compress_button.clicked.connect(self._start_compression)
|
||
actions.addWidget(self.compress_button, 1)
|
||
self.cancel_button = QPushButton("Cancel")
|
||
self.cancel_button.setObjectName("cancelButton")
|
||
self.cancel_button.setCursor(Qt.CursorShape.PointingHandCursor)
|
||
self.cancel_button.clicked.connect(self._cancel_compression)
|
||
self.cancel_button.setVisible(False)
|
||
actions.addWidget(self.cancel_button)
|
||
layout.addLayout(actions)
|
||
return card
|
||
|
||
def _build_destination_menu(self) -> None:
|
||
menu = QMenu(self.dest_button)
|
||
self._dest_source_action = QAction("Next to original files", menu)
|
||
self._dest_source_action.setCheckable(True)
|
||
self._dest_source_action.triggered.connect(lambda: self._set_output_dir(None))
|
||
menu.addAction(self._dest_source_action)
|
||
|
||
self._dest_choose_action = QAction("Choose folder…", menu)
|
||
self._dest_choose_action.triggered.connect(self._choose_output_dir)
|
||
menu.addAction(self._dest_choose_action)
|
||
|
||
menu.addSeparator()
|
||
self._dest_open_action = QAction("Open destination folder", menu)
|
||
self._dest_open_action.triggered.connect(self._open_destination)
|
||
menu.addAction(self._dest_open_action)
|
||
self.dest_button.setMenu(menu)
|
||
|
||
def _build_status_footer(self) -> QHBoxLayout:
|
||
layout = QHBoxLayout()
|
||
layout.setContentsMargins(4, 0, 4, 0)
|
||
layout.setSpacing(12)
|
||
self.status_label = QLabel("Ready. Add media to begin.")
|
||
self.status_label.setObjectName("footerText")
|
||
self.status_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
|
||
layout.addWidget(self.status_label, 1)
|
||
self.engine_label = QLabel()
|
||
self.engine_label.setObjectName("engineLabel")
|
||
layout.addWidget(self.engine_label)
|
||
return layout
|
||
|
||
def _install_shortcuts(self) -> None:
|
||
self._shortcuts = []
|
||
self._shortcuts.append(QShortcut(QKeySequence("Ctrl+O"), self, activated=self._choose_files))
|
||
self._shortcuts.append(
|
||
QShortcut(QKeySequence("Ctrl+Shift+O"), self, activated=self._choose_folder)
|
||
)
|
||
delete_shortcut = QShortcut(QKeySequence("Delete"), self.queue_view)
|
||
delete_shortcut.setContext(Qt.ShortcutContext.WidgetWithChildrenShortcut)
|
||
delete_shortcut.activated.connect(self._remove_selected)
|
||
self._shortcuts.append(delete_shortcut)
|
||
|
||
# ---------------------------------------------------------- preferences
|
||
|
||
def _load_preferences(self) -> None:
|
||
geometry = self._settings.value("window/geometry")
|
||
if geometry is not None:
|
||
self.restoreGeometry(geometry)
|
||
|
||
stored_dir = self._settings.value("paths/last_dir", str(Path.home()))
|
||
if stored_dir:
|
||
self._last_dir = str(stored_dir)
|
||
|
||
output_folder = self._settings.value("output/folder", "")
|
||
if output_folder:
|
||
self._output_dir = Path(str(output_folder))
|
||
|
||
image_options = ImageOptions(
|
||
output_format=str(self._settings.value("image/format", "original")),
|
||
quality=self._setting_int("image/quality", 82),
|
||
resize_mode=str(self._settings.value("image/resize_mode", "original")),
|
||
max_width=self._setting_int("image/max_width", 1920),
|
||
max_height=self._setting_int("image/max_height", 1080),
|
||
scale_percent=self._setting_int("image/scale_percent", 80),
|
||
keep_metadata=self._setting_bool("image/keep_metadata", False),
|
||
)
|
||
video_options = VideoOptions(
|
||
codec=str(self._settings.value("video/codec", "h264")),
|
||
quality=self._setting_int("video/quality", 68),
|
||
speed=str(self._settings.value("video/speed", "balanced")),
|
||
resolution=self._setting_int("video/resolution", 0),
|
||
audio_bitrate=self._setting_int("video/audio_bitrate", 128),
|
||
keep_metadata=self._setting_bool("video/keep_metadata", False),
|
||
)
|
||
self.settings_panel.set_image_options(image_options)
|
||
self.settings_panel.set_video_options(video_options)
|
||
self.settings_panel.set_current_tab_index(self._setting_int("ui/tab", 0))
|
||
self._update_destination_button()
|
||
|
||
if not self._settings.contains("window/geometry"):
|
||
screen = QApplication.primaryScreen()
|
||
if screen is not None:
|
||
available = screen.availableGeometry()
|
||
width = min(1200, max(1040, available.width() - 160))
|
||
height = min(820, max(680, available.height() - 140))
|
||
self.resize(width, height)
|
||
frame = self.frameGeometry()
|
||
frame.moveCenter(available.center())
|
||
self.move(frame.topLeft())
|
||
|
||
def _setting_int(self, key: str, default: int) -> int:
|
||
try:
|
||
return int(self._settings.value(key, default))
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
def _setting_bool(self, key: str, default: bool) -> bool:
|
||
value = self._settings.value(key, default)
|
||
if isinstance(value, str):
|
||
return value.strip().lower() in {"1", "true", "yes", "on"}
|
||
return bool(value)
|
||
|
||
def _on_settings_changed(self) -> None:
|
||
self._save_timer.start()
|
||
|
||
def _schedule_save(self) -> None:
|
||
self._save_timer.start()
|
||
|
||
def _save_preferences(self) -> None:
|
||
self._settings.setValue("window/geometry", self.saveGeometry())
|
||
self._settings.setValue("paths/last_dir", self._last_dir)
|
||
self._settings.setValue("output/folder", str(self._output_dir) if self._output_dir else "")
|
||
|
||
image_options = self.settings_panel.image_options()
|
||
self._settings.setValue("image/format", image_options.output_format)
|
||
self._settings.setValue("image/quality", image_options.quality)
|
||
self._settings.setValue("image/resize_mode", image_options.resize_mode)
|
||
self._settings.setValue("image/max_width", image_options.max_width)
|
||
self._settings.setValue("image/max_height", image_options.max_height)
|
||
self._settings.setValue("image/scale_percent", image_options.scale_percent)
|
||
self._settings.setValue("image/keep_metadata", image_options.keep_metadata)
|
||
|
||
video_options = self.settings_panel.video_options()
|
||
self._settings.setValue("video/codec", video_options.codec)
|
||
self._settings.setValue("video/quality", video_options.quality)
|
||
self._settings.setValue("video/speed", video_options.speed)
|
||
self._settings.setValue("video/resolution", video_options.resolution)
|
||
self._settings.setValue("video/audio_bitrate", video_options.audio_bitrate)
|
||
self._settings.setValue("video/keep_metadata", video_options.keep_metadata)
|
||
self._settings.setValue("ui/tab", self.settings_panel.current_tab_index())
|
||
self._settings.sync()
|
||
|
||
# ---------------------------------------------------------------- files
|
||
|
||
def _choose_files(self) -> None:
|
||
if self._running or self._scanning:
|
||
return
|
||
start_dir = self._last_dir if Path(self._last_dir).exists() else str(Path.home())
|
||
paths, _selected_filter = QFileDialog.getOpenFileNames(
|
||
self,
|
||
"Add media",
|
||
start_dir,
|
||
_MEDIA_FILTER,
|
||
)
|
||
if paths:
|
||
self._last_dir = str(Path(paths[0]).parent)
|
||
self.add_paths(paths)
|
||
|
||
def _choose_folder(self) -> None:
|
||
if self._running or self._scanning:
|
||
return
|
||
start_dir = self._last_dir if Path(self._last_dir).exists() else str(Path.home())
|
||
folder = QFileDialog.getExistingDirectory(self, "Add a folder", start_dir)
|
||
if folder:
|
||
self._last_dir = folder
|
||
self.add_paths([folder])
|
||
|
||
def add_paths(self, paths) -> None:
|
||
if self._running or self._scanning:
|
||
return
|
||
entries = [Path(path) for path in paths]
|
||
if not entries:
|
||
return
|
||
|
||
self._scanning = True
|
||
added_ids: List[str] = []
|
||
ignored = 0
|
||
missing = 0
|
||
duplicates = 0
|
||
failures: List[tuple[str, str]] = []
|
||
already_seen = set(self._seen)
|
||
was_empty = not self._order
|
||
|
||
QApplication.setOverrideCursor(Qt.CursorShape.WaitCursor)
|
||
try:
|
||
candidates: List[Path] = []
|
||
for entry in entries:
|
||
if entry.is_dir():
|
||
for root, dirs, files in os.walk(entry):
|
||
dirs[:] = sorted(
|
||
(
|
||
name
|
||
for name in dirs
|
||
if not name.startswith(".")
|
||
and name.lower() not in {"$recycle.bin", "system volume information"}
|
||
),
|
||
key=str.casefold,
|
||
)
|
||
for filename in sorted(files, key=str.casefold):
|
||
candidate = Path(root) / filename
|
||
if classify_path(candidate):
|
||
candidates.append(candidate)
|
||
else:
|
||
ignored += 1
|
||
elif entry.is_file():
|
||
if classify_path(entry):
|
||
candidates.append(entry)
|
||
else:
|
||
ignored += 1
|
||
else:
|
||
missing += 1
|
||
|
||
candidates.sort(key=lambda p: (str(p.parent).casefold(), p.name.casefold()))
|
||
unique_candidates: List[Path] = []
|
||
new_keys: Set[str] = set()
|
||
for candidate in candidates:
|
||
absolute = Path(os.path.abspath(candidate))
|
||
key = os.path.normcase(str(absolute))
|
||
if key in already_seen or key in new_keys:
|
||
duplicates += 1
|
||
continue
|
||
new_keys.add(key)
|
||
unique_candidates.append(absolute)
|
||
|
||
if unique_candidates:
|
||
self._set_status(f"Inspecting {len(unique_candidates)} file(s)…", MUTED)
|
||
QApplication.processEvents(QEventLoop.ProcessEventsFlag.ExcludeUserInputEvents)
|
||
|
||
self.queue_view.setUpdatesEnabled(False)
|
||
try:
|
||
for path in unique_candidates:
|
||
try:
|
||
info = probe_media(path, self._tools)
|
||
except Exception as exc:
|
||
failures.append((path.name, _one_line(str(exc), 240)))
|
||
continue
|
||
item = MediaItem(item_id=uuid.uuid4().hex, info=info)
|
||
self._items[item.item_id] = item
|
||
self._order.append(item.item_id)
|
||
self._seen.add(os.path.normcase(str(path)))
|
||
self._add_queue_row(item)
|
||
added_ids.append(item.item_id)
|
||
finally:
|
||
self.queue_view.setUpdatesEnabled(True)
|
||
finally:
|
||
QApplication.restoreOverrideCursor()
|
||
self._scanning = False
|
||
|
||
if added_ids and was_empty:
|
||
kinds = {self._items[item_id].kind for item_id in added_ids}
|
||
if kinds == {MediaKind.VIDEO}:
|
||
self.settings_panel.set_current_tab_index(1)
|
||
elif kinds == {MediaKind.IMAGE}:
|
||
self.settings_panel.set_current_tab_index(0)
|
||
|
||
if added_ids:
|
||
self._progress_visible = False
|
||
self._set_status(
|
||
f"Added {len(added_ids)} file{'s' if len(added_ids) != 1 else ''}. "
|
||
"Adjust settings, then compress.",
|
||
MUTED,
|
||
)
|
||
elif failures:
|
||
self._set_status("No files could be added. See the message for details.", RED)
|
||
|
||
extras = []
|
||
if duplicates:
|
||
extras.append(f"{duplicates} duplicate{'s' if duplicates != 1 else ''} skipped")
|
||
if ignored:
|
||
extras.append(f"{ignored} unsupported file{'s' if ignored != 1 else ''} ignored")
|
||
if missing:
|
||
extras.append(f"{missing} path{'s' if missing != 1 else ''} not found")
|
||
if extras and added_ids:
|
||
self._set_status(
|
||
f"Added {len(added_ids)} file{'s' if len(added_ids) != 1 else ''} · "
|
||
+ " · ".join(extras),
|
||
MUTED,
|
||
)
|
||
elif not added_ids and not failures and (ignored or missing or duplicates):
|
||
detail = " · ".join(extras) if extras else "No supported media was found."
|
||
self._set_status(detail, AMBER)
|
||
|
||
if added_ids:
|
||
self._last_dir = str(unique_candidates[0].parent)
|
||
self._schedule_save()
|
||
self._refresh_ui()
|
||
|
||
if failures:
|
||
shown = failures[:6]
|
||
message = "\n".join(f"• {name}: {error}" for name, error in shown)
|
||
if len(failures) > len(shown):
|
||
message += f"\n…and {len(failures) - len(shown)} more."
|
||
QMessageBox.warning(
|
||
self,
|
||
"Some files couldn't be added",
|
||
message,
|
||
)
|
||
|
||
def _add_queue_row(self, item: MediaItem) -> None:
|
||
info = item.info
|
||
details = f"{info.width} × {info.height}"
|
||
if info.kind == MediaKind.IMAGE:
|
||
details += f" · {info.format_name}"
|
||
else:
|
||
if info.duration:
|
||
details += f" · {format_duration(info.duration)}"
|
||
|
||
row = QTreeWidgetItem(
|
||
[
|
||
item.path.name,
|
||
details,
|
||
human_size(item.size),
|
||
"—",
|
||
"Ready",
|
||
]
|
||
)
|
||
row.setIcon(0, media_icon(item.kind))
|
||
row.setData(0, Qt.ItemDataRole.UserRole, item.item_id)
|
||
row.setSizeHint(0, QSize(0, 50))
|
||
row.setToolTip(0, str(item.path))
|
||
row.setToolTip(1, self._full_details(item))
|
||
row.setToolTip(2, f"Original size: {human_size(item.size)}")
|
||
row.setToolTip(3, "Compressed size appears here.")
|
||
row.setToolTip(4, "Ready to compress")
|
||
muted = QBrush(QColor(MUTED))
|
||
row.setForeground(1, muted)
|
||
row.setForeground(2, muted)
|
||
row.setForeground(3, muted)
|
||
row.setForeground(4, muted)
|
||
self.queue_view.addTopLevelItem(row)
|
||
self._rows[item.item_id] = row
|
||
|
||
@staticmethod
|
||
def _full_details(item: MediaItem) -> str:
|
||
info = item.info
|
||
lines = [
|
||
str(item.path),
|
||
f"Dimensions: {info.width} × {info.height}",
|
||
f"Size: {human_size(info.size)}",
|
||
f"Format: {info.format_name}",
|
||
]
|
||
if info.kind == MediaKind.VIDEO:
|
||
if info.duration:
|
||
lines.append(f"Duration: {format_duration(info.duration)}")
|
||
if info.codec:
|
||
lines.append(f"Codec: {info.codec}")
|
||
return "\n".join(lines)
|
||
|
||
# --------------------------------------------------------------- queue
|
||
|
||
def _refresh_ui(self) -> None:
|
||
self.queue_stack.setCurrentIndex(1 if self._order else 0)
|
||
self.count_badge.setText(str(len(self._order)))
|
||
self._update_stats()
|
||
self._update_destination_button()
|
||
self._refresh_controls()
|
||
|
||
def _refresh_controls(self) -> None:
|
||
running = self._running
|
||
has_items = bool(self._order)
|
||
self.add_files_button.setEnabled(not running)
|
||
self.add_folder_button.setEnabled(not running)
|
||
self.clear_button.setEnabled(not running and has_items)
|
||
self.dest_button.setEnabled(not running)
|
||
self.settings_panel.setEnabled(not running)
|
||
self.progress_wrap.setVisible(self._progress_visible)
|
||
self.cancel_button.setVisible(running)
|
||
self.compress_button.setVisible(True)
|
||
if running:
|
||
self.compress_button.setEnabled(False)
|
||
self.compress_button.setText("Compressing…")
|
||
self.compress_button.setToolTip("Compression is in progress.")
|
||
self.cancel_button.setEnabled(True)
|
||
self.cancel_button.setText("Cancel")
|
||
else:
|
||
self.compress_button.setEnabled(has_items)
|
||
if has_items:
|
||
pending = sum(
|
||
1
|
||
for item_id in self._order
|
||
if not self._item_has_output(self._items[item_id])
|
||
)
|
||
if pending == 0:
|
||
self.compress_button.setText("Compress again")
|
||
self.compress_button.setToolTip(
|
||
"Re-encode every queued file and create new compressed copies."
|
||
)
|
||
else:
|
||
self.compress_button.setText(
|
||
f"Compress {pending} file{'s' if pending != 1 else ''}"
|
||
)
|
||
self.compress_button.setToolTip(
|
||
"Only files without a compressed result will be processed."
|
||
if pending != len(self._order)
|
||
else "Compress the files in the queue."
|
||
)
|
||
else:
|
||
self.compress_button.setText("Compress")
|
||
self.compress_button.setToolTip("Add media to the queue first.")
|
||
self._update_destination_menu_state()
|
||
|
||
@staticmethod
|
||
def _item_has_output(item: MediaItem) -> bool:
|
||
return item.output_path is not None and item.output_path.exists()
|
||
|
||
def _update_stats(self) -> None:
|
||
if not self._order:
|
||
self.stats_label.setText("No files yet")
|
||
self.stats_label.setToolTip("")
|
||
return
|
||
items = [self._items[item_id] for item_id in self._order]
|
||
total_input = sum(item.size for item in items)
|
||
completed = [item for item in items if self._item_has_output(item)]
|
||
if completed:
|
||
completed_input = sum(item.size for item in completed)
|
||
completed_output = sum(item.output_size for item in completed)
|
||
prefix = (
|
||
f"{len(completed)} of {len(items)} compressed"
|
||
if len(completed) != len(items)
|
||
else f"{len(items)} file{'s' if len(items) != 1 else ''}"
|
||
)
|
||
savings = 0.0
|
||
if completed_input > 0:
|
||
savings = (1 - completed_output / completed_input) * 100
|
||
if savings >= 0:
|
||
saving_text = f"{round(savings)}% smaller"
|
||
else:
|
||
saving_text = f"{round(abs(savings))}% larger"
|
||
text = (
|
||
f"{prefix} · {human_size(completed_input)} → {human_size(completed_output)}"
|
||
f" · {saving_text}"
|
||
)
|
||
full_tooltip = (
|
||
f"Queue: {human_size(total_input)} across {len(items)} file(s)\n"
|
||
f"Completed: {human_size(completed_input)} → {human_size(completed_output)}"
|
||
)
|
||
self.stats_label.setToolTip(full_tooltip)
|
||
else:
|
||
text = f"{len(items)} file{'s' if len(items) != 1 else ''} · {human_size(total_input)} total"
|
||
self.stats_label.setToolTip(f"Total input size: {human_size(total_input)}")
|
||
self.stats_label.setText(text)
|
||
|
||
def _set_row_status(self, item_id: str, text: str, color: str = MUTED, tooltip: Optional[str] = None) -> None:
|
||
row = self._rows.get(item_id)
|
||
if row is None:
|
||
return
|
||
row.setText(4, text)
|
||
row.setForeground(4, QBrush(QColor(color)))
|
||
if tooltip is not None:
|
||
row.setToolTip(4, tooltip)
|
||
|
||
def _set_row_result(self, item_id: str, text: str, color: str = MUTED, tooltip: Optional[str] = None) -> None:
|
||
row = self._rows.get(item_id)
|
||
if row is None:
|
||
return
|
||
row.setText(3, text)
|
||
row.setForeground(3, QBrush(QColor(color)))
|
||
if tooltip is not None:
|
||
row.setToolTip(3, tooltip)
|
||
|
||
def _show_queue_menu(self, position) -> None:
|
||
row = self.queue_view.itemAt(position)
|
||
if row is None:
|
||
return
|
||
if not row.isSelected():
|
||
self.queue_view.clearSelection()
|
||
row.setSelected(True)
|
||
item_id = row.data(0, Qt.ItemDataRole.UserRole)
|
||
item = self._items.get(item_id)
|
||
if item is None:
|
||
return
|
||
|
||
menu = QMenu(self)
|
||
open_original = menu.addAction("Open original")
|
||
reveal_original = menu.addAction("Show original in folder")
|
||
output_action = None
|
||
reveal_output_action = None
|
||
if item.output_path is not None and item.output_path.exists():
|
||
menu.addSeparator()
|
||
output_action = menu.addAction("Open compressed file")
|
||
reveal_output_action = menu.addAction("Show compressed file in folder")
|
||
error_action = None
|
||
if item.message and not self._running:
|
||
menu.addSeparator()
|
||
error_action = menu.addAction("Show error details…")
|
||
menu.addSeparator()
|
||
remove_action = menu.addAction("Remove selected from queue")
|
||
remove_action.setEnabled(not self._running)
|
||
|
||
chosen = menu.exec(self.queue_view.viewport().mapToGlobal(position))
|
||
if chosen is open_original:
|
||
_open_path(item.path)
|
||
elif chosen is reveal_original:
|
||
_reveal_path(item.path)
|
||
elif output_action is not None and chosen is output_action:
|
||
_open_path(item.output_path)
|
||
elif reveal_output_action is not None and chosen is reveal_output_action:
|
||
_reveal_path(item.output_path)
|
||
elif error_action is not None and chosen is error_action:
|
||
QMessageBox.warning(
|
||
self,
|
||
f"Could not compress {item.path.name}",
|
||
item.message or "No further details are available.",
|
||
)
|
||
elif chosen is remove_action:
|
||
self._remove_selected()
|
||
|
||
def _open_queue_item(self, row: QTreeWidgetItem, _column: int) -> None:
|
||
item_id = row.data(0, Qt.ItemDataRole.UserRole)
|
||
item = self._items.get(item_id)
|
||
if item is None:
|
||
return
|
||
target = item.output_path if item.output_path and item.output_path.exists() else item.path
|
||
_open_path(target)
|
||
|
||
def _remove_selected(self) -> None:
|
||
if self._running:
|
||
return
|
||
selected_ids = set()
|
||
for row in self.queue_view.selectedItems():
|
||
item_id = row.data(0, Qt.ItemDataRole.UserRole)
|
||
if item_id:
|
||
selected_ids.add(item_id)
|
||
if not selected_ids:
|
||
return
|
||
for index in reversed(range(self.queue_view.topLevelItemCount())):
|
||
row = self.queue_view.topLevelItem(index)
|
||
if row.data(0, Qt.ItemDataRole.UserRole) in selected_ids:
|
||
self.queue_view.takeTopLevelItem(index)
|
||
for item_id in selected_ids:
|
||
item = self._items.pop(item_id, None)
|
||
self._rows.pop(item_id, None)
|
||
if item is not None:
|
||
self._seen.discard(os.path.normcase(str(item.path)))
|
||
self._order = [item_id for item_id in self._order if item_id in self._items]
|
||
self._progress_visible = False
|
||
self._last_run_summary = None
|
||
self._set_status(
|
||
f"Removed {len(selected_ids)} file{'s' if len(selected_ids) != 1 else ''} from the queue.",
|
||
MUTED,
|
||
)
|
||
self._refresh_ui()
|
||
|
||
def _clear_queue(self) -> None:
|
||
if self._running or not self._order:
|
||
return
|
||
self.queue_view.clear()
|
||
self._items.clear()
|
||
self._rows.clear()
|
||
self._order.clear()
|
||
self._seen.clear()
|
||
self._progress_visible = False
|
||
self._last_run_summary = None
|
||
self.progress_bar.setValue(0)
|
||
self.percent_label.setText("0%")
|
||
self._set_status("Queue cleared.", MUTED)
|
||
self._refresh_ui()
|
||
|
||
# ------------------------------------------------------------- outputs
|
||
|
||
def _set_output_dir(self, folder: Optional[Path]) -> None:
|
||
if self._running:
|
||
return
|
||
self._output_dir = Path(folder) if folder else None
|
||
self._update_destination_button()
|
||
self._schedule_save()
|
||
if folder:
|
||
self._set_status(f"Output folder set to {folder}.", MUTED)
|
||
else:
|
||
self._set_status("Output will be saved next to each original file.", MUTED)
|
||
|
||
def _choose_output_dir(self) -> None:
|
||
if self._running:
|
||
return
|
||
if self._output_dir and self._output_dir.exists():
|
||
start_dir = str(self._output_dir)
|
||
elif self._order:
|
||
start_dir = str(self._items[self._order[0]].path.parent)
|
||
else:
|
||
start_dir = self._last_dir if Path(self._last_dir).exists() else str(Path.home())
|
||
folder = QFileDialog.getExistingDirectory(self, "Choose output folder", start_dir)
|
||
if folder:
|
||
self._set_output_dir(Path(folder))
|
||
|
||
def _update_destination_button(self) -> None:
|
||
if self._output_dir:
|
||
name = self._output_dir.name or str(self._output_dir)
|
||
self.dest_button.setText(f"Save to {name}")
|
||
self.dest_button.setToolTip(f"Compressed files will be saved to:\n{self._output_dir}")
|
||
else:
|
||
self.dest_button.setText("Next to originals")
|
||
self.dest_button.setToolTip(
|
||
"Compressed files are saved beside their originals with a -compressed suffix."
|
||
)
|
||
self._dest_source_action.setChecked(self._output_dir is None)
|
||
self._update_destination_menu_state()
|
||
|
||
def _update_destination_menu_state(self) -> None:
|
||
target = self._destination_folder()
|
||
exists = target is not None and target.exists()
|
||
if hasattr(self, "_dest_open_action"):
|
||
self._dest_open_action.setEnabled(exists)
|
||
if hasattr(self, "_dest_choose_action"):
|
||
self._dest_choose_action.setEnabled(not self._running)
|
||
if hasattr(self, "_dest_source_action"):
|
||
self._dest_source_action.setEnabled(not self._running)
|
||
|
||
def _destination_folder(self) -> Optional[Path]:
|
||
if self._output_dir is not None:
|
||
return self._output_dir
|
||
if self._order:
|
||
return self._items[self._order[0]].path.parent
|
||
return None
|
||
|
||
def _open_destination(self) -> None:
|
||
target = self._destination_folder()
|
||
if target is not None and target.exists():
|
||
_open_path(target)
|
||
|
||
# ----------------------------------------------------------- processing
|
||
|
||
def _start_compression(self) -> None:
|
||
if self._running or not self._order:
|
||
return
|
||
all_items = [self._items[item_id] for item_id in self._order]
|
||
pending_items = [
|
||
item for item in all_items if not self._item_has_output(item)
|
||
]
|
||
# If everything already has a result, "Compress again" re-encodes the whole queue.
|
||
items = pending_items or all_items
|
||
video_items = [item for item in items if item.kind == MediaKind.VIDEO]
|
||
if video_items and not self._tools.can_encode_video:
|
||
QMessageBox.warning(
|
||
self,
|
||
"FFmpeg is required for video",
|
||
"FFmpeg (and ffprobe) could not be found.\n\n"
|
||
"Install FFmpeg and make sure both tools are on PATH, then restart Bolt Compress. "
|
||
"Image compression will still work without FFmpeg.",
|
||
)
|
||
return
|
||
|
||
if self._output_dir is not None:
|
||
if self._output_dir.exists() and not self._output_dir.is_dir():
|
||
QMessageBox.warning(
|
||
self,
|
||
"Invalid output folder",
|
||
"The chosen output destination is a file, not a folder.",
|
||
)
|
||
return
|
||
|
||
self._last_run_summary = None
|
||
self._progress_visible = True
|
||
self.progress_bar.setValue(0)
|
||
self.percent_label.setText("0%")
|
||
self._set_status("Preparing compression…", MUTED)
|
||
for item in items:
|
||
item.output_path = None
|
||
item.output_size = 0
|
||
item.message = ""
|
||
self._set_row_result(item.item_id, "—", MUTED, "Compressed size appears here.")
|
||
self._set_row_status(item.item_id, "Queued", MUTED, "Waiting to compress")
|
||
self._update_stats()
|
||
|
||
self._running = True
|
||
self._refresh_controls()
|
||
self.queue_view.clearSelection()
|
||
|
||
jobs = [
|
||
CompressionJob(
|
||
item_id=item.item_id,
|
||
source=item.path,
|
||
kind=item.kind,
|
||
width=item.info.width,
|
||
height=item.info.height,
|
||
duration=item.info.duration,
|
||
)
|
||
for item in items
|
||
]
|
||
worker = CompressionWorker(
|
||
jobs,
|
||
image_options=self.settings_panel.image_options(),
|
||
video_options=self.settings_panel.video_options(),
|
||
output_dir=self._output_dir,
|
||
suffix=OUTPUT_SUFFIX,
|
||
parent=self,
|
||
)
|
||
worker.itemStarted.connect(self._on_item_started)
|
||
worker.itemProgress.connect(self._on_item_progress)
|
||
worker.itemFinished.connect(self._on_item_finished)
|
||
worker.overallProgress.connect(self._on_overall_progress)
|
||
worker.summary.connect(self._on_worker_summary)
|
||
worker.finished.connect(self._on_worker_finished)
|
||
self._worker = worker
|
||
worker.start()
|
||
|
||
def _cancel_compression(self) -> None:
|
||
if self._worker is None or not self._running:
|
||
return
|
||
self._worker.cancel()
|
||
self.cancel_button.setEnabled(False)
|
||
self.cancel_button.setText("Stopping…")
|
||
self._set_status("Stopping the current encode…", AMBER)
|
||
|
||
def _on_item_started(self, item_id: str) -> None:
|
||
item = self._items.get(item_id)
|
||
if item is None:
|
||
return
|
||
self._current_id = item_id
|
||
self._set_row_status(item_id, "Working", ACCENT, "Compressing this file")
|
||
row = self._rows.get(item_id)
|
||
if row is not None:
|
||
self.queue_view.scrollToItem(row, QAbstractItemView.ScrollHint.PositionAtCenter)
|
||
self._set_status(f"Compressing {item.path.name}…", MUTED)
|
||
|
||
def _on_item_progress(self, item_id: str, percent: int) -> None:
|
||
self._set_row_status(item_id, f"{percent}%", ACCENT, "Compressing this file")
|
||
|
||
def _on_overall_progress(self, percent: int) -> None:
|
||
self.progress_bar.setValue(percent)
|
||
self.percent_label.setText(f"{percent}%")
|
||
|
||
def _on_item_finished(
|
||
self,
|
||
item_id: str,
|
||
status: str,
|
||
message: str,
|
||
output_path: str,
|
||
output_size: int,
|
||
) -> None:
|
||
item = self._items.get(item_id)
|
||
if item is None:
|
||
return
|
||
if status == "done":
|
||
item.output_path = Path(output_path) if output_path else None
|
||
item.output_size = output_size
|
||
self._set_row_result(
|
||
item_id,
|
||
human_size(output_size),
|
||
"#D7E0EA",
|
||
f"Compressed file: {output_path}",
|
||
)
|
||
if item.size > 0 and output_size < item.size:
|
||
saved_percent = max(1, round((1 - output_size / item.size) * 100))
|
||
status_text = f"Saved {saved_percent}%"
|
||
status_color = GREEN
|
||
detail = f"{human_size(item.size - output_size)} saved ({saved_percent}%)"
|
||
elif item.size > 0:
|
||
growth = round((output_size / item.size - 1) * 100)
|
||
status_text = "No gain" if growth < 1 else f"Grew {growth}%"
|
||
status_color = AMBER
|
||
detail = "no size gain"
|
||
else:
|
||
status_text = "Done"
|
||
status_color = GREEN
|
||
detail = "compression finished"
|
||
self._set_row_status(
|
||
item_id,
|
||
status_text,
|
||
status_color,
|
||
f"{detail}\n{output_path}",
|
||
)
|
||
self._set_status(
|
||
f"Compressed {item.path.name} · {detail}.",
|
||
status_color,
|
||
)
|
||
elif status == "cancelled":
|
||
item.output_path = None
|
||
item.output_size = 0
|
||
self._set_row_result(item_id, "—", MUTED, "No output was created.")
|
||
self._set_row_status(item_id, "Cancelled", AMBER, "Compression was cancelled.")
|
||
else:
|
||
item.output_path = None
|
||
item.output_size = 0
|
||
item.message = message
|
||
self._set_row_result(item_id, "—", MUTED, "No output was created.")
|
||
self._set_row_status(item_id, "Failed", RED, message or "Compression failed.")
|
||
self._set_status(
|
||
f"Could not compress {item.path.name}: {_one_line(message, 120)}",
|
||
RED,
|
||
)
|
||
self._update_stats()
|
||
|
||
def _on_worker_summary(self, summary: dict) -> None:
|
||
self._last_run_summary = summary
|
||
|
||
def _on_worker_finished(self) -> None:
|
||
worker = self._worker
|
||
self._worker = None
|
||
if worker is not None:
|
||
worker.deleteLater()
|
||
self._running = False
|
||
self._current_id = None
|
||
self._refresh_controls()
|
||
self._update_stats()
|
||
|
||
summary = self._last_run_summary or {
|
||
"total": len(self._order),
|
||
"done": 0,
|
||
"errors": 0,
|
||
"cancelled": 0,
|
||
"output_bytes": 0,
|
||
"input_bytes": 0,
|
||
}
|
||
total = summary.get("total", 0)
|
||
done = summary.get("done", 0)
|
||
errors = summary.get("errors", 0)
|
||
cancelled = summary.get("cancelled", 0)
|
||
if cancelled:
|
||
self._set_status(
|
||
f"Cancelled. {done} file{'s' if done != 1 else ''} completed"
|
||
+ (f", {errors} failed." if errors else "."),
|
||
AMBER,
|
||
)
|
||
elif errors:
|
||
self._set_status(
|
||
f"Finished with {errors} failure{'s' if errors != 1 else ''}. "
|
||
f"{done} of {total} compressed. Hover a failed row for details.",
|
||
RED,
|
||
)
|
||
elif done:
|
||
input_bytes = summary.get("input_bytes", 0)
|
||
output_bytes = summary.get("output_bytes", 0)
|
||
if input_bytes > 0:
|
||
savings = (1 - output_bytes / input_bytes) * 100
|
||
if savings >= 0:
|
||
savings_text = f"{human_size(max(0, input_bytes - output_bytes))} saved ({round(savings)}%)"
|
||
else:
|
||
savings_text = f"output is {human_size(max(0, output_bytes - input_bytes))} larger"
|
||
else:
|
||
savings_text = "done"
|
||
self._set_status(
|
||
f"Done — {done} file{'s' if done != 1 else ''} compressed · {savings_text}.",
|
||
GREEN,
|
||
)
|
||
else:
|
||
self._set_status("Queue is ready.", MUTED)
|
||
self.progress_wrap.setVisible(self._progress_visible)
|
||
self._refresh_ui()
|
||
|
||
# ------------------------------------------------------------- drag/drop
|
||
|
||
def dragEnterEvent(self, event) -> None:
|
||
if self._running or self._scanning or not event.mimeData().hasUrls():
|
||
event.ignore()
|
||
return
|
||
if any(url.isLocalFile() for url in event.mimeData().urls()):
|
||
self._set_drag_active(True)
|
||
event.acceptProposedAction()
|
||
else:
|
||
event.ignore()
|
||
|
||
def dragMoveEvent(self, event) -> None:
|
||
if self._running or self._scanning:
|
||
event.ignore()
|
||
return
|
||
if event.mimeData().hasUrls():
|
||
event.acceptProposedAction()
|
||
else:
|
||
event.ignore()
|
||
|
||
def dragLeaveEvent(self, event) -> None:
|
||
self._set_drag_active(False)
|
||
super().dragLeaveEvent(event)
|
||
|
||
def dropEvent(self, event) -> None:
|
||
self._set_drag_active(False)
|
||
if self._running or self._scanning:
|
||
event.ignore()
|
||
return
|
||
local_paths = [
|
||
url.toLocalFile()
|
||
for url in event.mimeData().urls()
|
||
if url.isLocalFile() and url.toLocalFile()
|
||
]
|
||
if local_paths:
|
||
event.acceptProposedAction()
|
||
self.add_paths(local_paths)
|
||
else:
|
||
event.ignore()
|
||
|
||
def _set_drag_active(self, active: bool) -> None:
|
||
if self._drag_active == active:
|
||
return
|
||
self._drag_active = active
|
||
self.queue_card.setProperty("dragActive", active)
|
||
self.queue_card.style().unpolish(self.queue_card)
|
||
self.queue_card.style().polish(self.queue_card)
|
||
self.queue_card.update()
|
||
self.drop_zone.set_drag_active(active)
|
||
|
||
# --------------------------------------------------------------- status
|
||
|
||
def _set_status(self, text: str, color: Optional[str] = None) -> None:
|
||
self.status_label.setText(text)
|
||
if color:
|
||
self.status_label.setStyleSheet(f"color: {color};")
|
||
else:
|
||
self.status_label.setStyleSheet("")
|
||
|
||
def _update_engine_status(self) -> None:
|
||
tools = self._tools
|
||
if tools.can_encode_video:
|
||
self.engine_label.setText("● Engines ready · nothing uploaded")
|
||
self.engine_label.setStyleSheet(f"color: {GREEN};")
|
||
tooltip = f"FFmpeg: {tools.ffmpeg}\nffprobe: {tools.ffprobe}\nImages: Pillow"
|
||
elif tools.ffmpeg is not None:
|
||
self.engine_label.setText("● ffprobe missing — video unavailable")
|
||
self.engine_label.setStyleSheet(f"color: {AMBER};")
|
||
tooltip = "ffprobe was not found. Add it to PATH to enable video compression."
|
||
else:
|
||
self.engine_label.setText("● FFmpeg missing — video unavailable")
|
||
self.engine_label.setStyleSheet(f"color: {RED};")
|
||
tooltip = (
|
||
"FFmpeg was not found. Install it and add ffmpeg + ffprobe to PATH "
|
||
"to enable video compression. Images work without it."
|
||
)
|
||
self.engine_label.setToolTip(tooltip)
|
||
|
||
# --------------------------------------------------------------- events
|
||
|
||
def closeEvent(self, event) -> None:
|
||
if self._worker is not None and self._worker.isRunning():
|
||
reply = QMessageBox.question(
|
||
self,
|
||
"Stop compression and quit?",
|
||
"A compression is still running. Stop it and close Bolt Compress?",
|
||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||
QMessageBox.StandardButton.No,
|
||
)
|
||
if reply != QMessageBox.StandardButton.Yes:
|
||
event.ignore()
|
||
return
|
||
self._worker.cancel()
|
||
if not self._worker.wait(15000):
|
||
QMessageBox.warning(
|
||
self,
|
||
"Still stopping",
|
||
"The current file is still finishing. Please wait a moment and close again.",
|
||
)
|
||
event.ignore()
|
||
return
|
||
self._save_preferences()
|
||
super().closeEvent(event)
|