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)