From 5f17fabc50755675e92cdbea82e88e6e6646542d Mon Sep 17 00:00:00 2001 From: oop7 <110548351+oop7@users.noreply.github.com> Date: Wed, 21 Jan 2026 17:56:56 +0200 Subject: [PATCH] Refactor history dialog to use virtualized list and SQLite Replaces the history dialog's widget-per-entry approach with a virtualized QListView using a custom model and delegate for improved performance. Switches the HistoryManager backend from JSON file storage to SQLite, including automatic migration of legacy data. Adds support for efficient search, thumbnail caching, and context menu actions in the new UI. --- .../ytsage_dialogs_history.py | 960 ++++++++---------- src/utils/ytsage_history_manager.py | 574 ++++++----- 2 files changed, 752 insertions(+), 782 deletions(-) diff --git a/src/gui/ytsage_gui_dialogs/ytsage_dialogs_history.py b/src/gui/ytsage_gui_dialogs/ytsage_dialogs_history.py index 6345ff5..58308db 100644 --- a/src/gui/ytsage_gui_dialogs/ytsage_dialogs_history.py +++ b/src/gui/ytsage_gui_dialogs/ytsage_dialogs_history.py @@ -1,19 +1,26 @@ """ History Dialog for YTSage application. -Displays download history with thumbnails and provides options to redownload or remove entries. +Displays download history with thumbnails using a virtualized list for performance. """ import os import subprocess +import json from datetime import datetime from io import BytesIO from pathlib import Path -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Optional, List, Any, Dict import requests from PIL import Image -from PySide6.QtCore import Qt, QSize, Signal, QThread, QTimer -from PySide6.QtGui import QPixmap, QIcon +from PySide6.QtCore import ( + Qt, QSize, Signal, QThread, QTimer, QAbstractListModel, + QModelIndex, QRect, QPoint, QEvent +) +from PySide6.QtGui import ( + QPixmap, QIcon, QPainter, QColor, QFont, QBrush, QPen, + QMouseEvent, QDesktopServices, QAction, QCursor, QPainterPath +) from PySide6.QtWidgets import ( QDialog, QVBoxLayout, @@ -21,13 +28,13 @@ from PySide6.QtWidgets import ( QLabel, QLineEdit, QPushButton, - QScrollArea, + QListView, QWidget, - QFrame, QMenu, QMessageBox, - QSizePolicy, - QProgressBar, + QStyledItemDelegate, + QStyle, + QApplication ) from src.utils.ytsage_history_manager import HistoryManager @@ -43,18 +50,19 @@ class HistoryLoaderThread(QThread): """Thread to load history and pre-fetch thumbnails.""" entries_loaded = Signal(list) - thumbnail_loaded = Signal(str) + thumbnail_loaded = Signal(str, bytes) # entry_id, image_bytes def run(self): try: - # HistoryManager uses get_all_entries, not get_all + # Load entries from DB entries = HistoryManager.get_all_entries() - - # Emit entries immediately so UI shows up self.entries_loaded.emit(entries) - # Pre-fetch thumbnails in background + # Background thumbnail loader for entry in entries: + if self.isInterruptionRequested(): + break + thumbnail_url = entry.get("thumbnail_url") entry_id = entry.get("id", "") @@ -64,594 +72,494 @@ class HistoryLoaderThread(QThread): thumbnail_filename = f"{entry_id}.jpg" thumbnail_path = APP_THUMBNAILS_DIR / thumbnail_filename - # If not exists, download it if not thumbnail_path.exists(): try: response = requests.get(thumbnail_url, timeout=5) if response.status_code == 200: APP_THUMBNAILS_DIR.mkdir(parents=True, exist_ok=True) - image = Image.open(BytesIO(response.content)) - image.save(thumbnail_path, "JPEG", quality=95, optimize=True) - self.thumbnail_loaded.emit(entry_id) + # Optimize image before saving + img_io = BytesIO(response.content) + image = Image.open(img_io) + + # Save to disk + image.save(thumbnail_path, "JPEG", quality=90, optimize=True) + + # Emit bytes for memory cache + self.thumbnail_loaded.emit(entry_id, response.content) except Exception as e: - logger.debug(f"Error caching thumbnail in background: {e}") - + logger.debug(f"Error caching thumbnail: {e}") + except Exception as e: logger.error(f"Error loading history: {e}") self.entries_loaded.emit([]) -class HistoryEntryWidget(QFrame): - """Widget representing a single history entry.""" +class HistoryModel(QAbstractListModel): + """List Model for History Entries.""" - remove_requested = Signal(str) # Emit entry ID when remove is requested - redownload_requested = Signal(dict) # Emit entry data when redownload is requested - - def __init__(self, entry: dict, parent=None): + EntryRole = Qt.ItemDataRole.UserRole + 1 + IdRole = Qt.ItemDataRole.UserRole + 2 + ThumbnailRole = Qt.ItemDataRole.UserRole + 3 + + def __init__(self, entries=None, parent=None): super().__init__(parent) - self.entry = entry - self.entry_id = entry.get("id", "") + self._entries = entries or [] + self.thumbnail_cache = {} # Map entry_id -> QPixmap + + def rowCount(self, parent=QModelIndex()): + return len(self._entries) + + def data(self, index, role=Qt.ItemDataRole.DisplayRole): + if not index.isValid() or not (0 <= index.row() < len(self._entries)): + return None - self.setup_ui() + entry = self._entries[index.row()] + entry_id = entry.get("id") + + if role == self.EntryRole: + return entry + + elif role == self.IdRole: + return entry_id + + elif role == self.ThumbnailRole: + return self.thumbnail_cache.get(entry_id) + + elif role == Qt.ItemDataRole.DisplayRole: + return entry.get("title", "") + + return None + + def update_entries(self, entries): + self.beginResetModel() + self._entries = entries + self.endResetModel() + + def remove_item(self, row): + if 0 <= row < len(self._entries): + self.beginRemoveRows(QModelIndex(), row, row) + del self._entries[row] + self.endRemoveRows() + + def update_thumbnail(self, entry_id, pixmap): + """Update cache and notify view.""" + self.thumbnail_cache[entry_id] = pixmap + # Find index for this ID + for i, entry in enumerate(self._entries): + if entry.get("id") == entry_id: + idx = self.index(i) + self.dataChanged.emit(idx, idx, [self.ThumbnailRole]) + break + + +class HistoryDelegate(QStyledItemDelegate): + """Delegate to render history cards similar to the widgets.""" - def setup_ui(self): - """Setup the UI for this history entry.""" - self.setFrameStyle(QFrame.Shape.StyledPanel | QFrame.Shadow.Raised) - self.setStyleSheet(""" - QFrame { - background-color: #1d1e22; - border: 1px solid #2a2d36; - border-radius: 8px; - padding: 4px; - margin: 2px; - } - QFrame:hover { - background-color: #252830; - border-color: #3a3d46; - } - """) + menu_clicked = Signal(QModelIndex, QPoint) # Signal for menu click + + def __init__(self, parent=None): + super().__init__(parent) + self.padding = 10 + self.thumb_width = 240 + self.thumb_height = 135 + # Increased card height to accommodate spacing + self.card_height = 175 + # Define margins for spacing between cards + self.h_margin = 10 + self.v_margin = 8 + + def sizeHint(self, option, index): + return QSize(option.rect.width(), self.card_height) + + def paint(self, painter, option, index): + entry = index.data(HistoryModel.EntryRole) + if not entry: + return + + painter.save() + painter.setRenderHint(QPainter.RenderHint.Antialiasing) - main_layout = QHBoxLayout(self) - main_layout.setSpacing(12) - main_layout.setContentsMargins(8, 8, 8, 8) + rect = option.rect + # Apply margins for spacing + card_rect = rect.adjusted(self.h_margin, self.v_margin, -self.h_margin, -self.v_margin) - # Thumbnail - Standard YouTube 16:9 ratio (e.g., 240x135) - self.thumbnail_label = QLabel() - self.thumbnail_label.setFixedSize(240, 135) - self.thumbnail_label.setStyleSheet(""" - QLabel { - border: 2px solid #3d3d3d; - border-radius: 6px; - background-color: #15181b; - } - """) - self.thumbnail_label.setAlignment(Qt.AlignmentFlag.AlignCenter) - self.thumbnail_label.setScaledContents(True) + is_hover = option.state & QStyle.StateFlag.State_MouseOver + bg_color = QColor("#252830") if is_hover else QColor("#1d1e22") + border_color = QColor("#3a3d46") if is_hover else QColor("#2a2d36") - # Load thumbnail - self.load_thumbnail() + # Draw Card + path = QPainterPath() + path.addRoundedRect(card_rect, 8, 8) - main_layout.addWidget(self.thumbnail_label, alignment=Qt.AlignmentFlag.AlignTop) + painter.fillPath(path, QBrush(bg_color)) + painter.setPen(QPen(border_color, 1)) + painter.drawPath(path) - # Info section - info_layout = QVBoxLayout() - info_layout.setSpacing(5) + # Draw Thumbnail + thumb_rect = QRect( + card_rect.left() + 10, + card_rect.top() + 10, + self.thumb_width, + self.thumb_height + ) + + pixmap = index.data(HistoryModel.ThumbnailRole) + if pixmap and not pixmap.isNull(): + scaled = pixmap.scaled( + thumb_rect.size(), + Qt.AspectRatioMode.KeepAspectRatioByExpanding, + Qt.TransformationMode.SmoothTransformation + ) + # Clip to rect + painter.setClipRect(thumb_rect) + painter.drawPixmap(thumb_rect.topLeft(), scaled) + painter.setClipping(False) + else: + painter.fillRect(thumb_rect, QColor("#15181b")) + painter.setPen(QPen(QColor("#666666"))) + icon_char = "🎵" if entry.get("is_audio_only") else "📹" + painter.setFont(QFont("Segoe UI Emoji", 24)) + painter.drawText(thumb_rect, Qt.AlignmentFlag.AlignCenter, icon_char) + + # Draw Border around thumb + painter.setPen(QPen(QColor("#3d3d3d"), 2)) + painter.drawRect(thumb_rect) + + # Text Area + text_x = thumb_rect.right() + 12 + text_width = card_rect.right() - text_x - 50 # Title - title = self.entry.get("title", _("video_info.unknown_title")) - self.title_label = QLabel(title) - self.title_label.setWordWrap(True) - self.title_label.setStyleSheet(""" - QLabel { - font-size: 14px; - font-weight: bold; - color: #ffffff; - } - """) - info_layout.addWidget(self.title_label) + title_rect = QRect(text_x, thumb_rect.top(), text_width, 50) + painter.setPen(QColor("#ffffff")) + font_title = QFont() + font_title.setBold(True) + font_title.setPixelSize(14) + painter.setFont(font_title) - # Channel (if available) - channel = self.entry.get("channel") + # Use simple alignment flags + painter.drawText(title_rect, Qt.AlignmentFlag.AlignLeft | Qt.TextFlag.TextWordWrap, entry.get("title", "")) + + current_y = title_rect.bottom() + 5 + + # Channel + channel = entry.get("channel") if channel: - channel_label = QLabel(f"{_('video_info.channel')}: {channel}") - channel_label.setStyleSheet("color: #cccccc; font-size: 12px;") - info_layout.addWidget(channel_label) + painter.setPen(QColor("#cccccc")) + font_meta = QFont() + font_meta.setPixelSize(12) + painter.setFont(font_meta) + painter.drawText(text_x, current_y, f"{_('video_info.channel')}: {channel}") + current_y += 18 + + # Date + date_str = entry.get("download_date", "")[:16].replace('T', ' ') + if date_str: + painter.setPen(QColor("#aaaaaa")) + painter.setFont(QFont("Arial", 11)) + painter.drawText(text_x, current_y, f"{_('history.downloaded_on', date=date_str)}") + current_y += 25 + + # Badge + is_audio = entry.get("is_audio_only", False) + badge_text = _("history.audio_download") if is_audio else _("history.video_download") + badge_color = QColor("#0066cc") if is_audio else QColor("#c90000") - # Download date - download_date = self.entry.get("download_date", "") - if download_date: - try: - dt = datetime.fromisoformat(download_date) - date_str = dt.strftime("%Y-%m-%d %H:%M") - date_label = QLabel(_("history.downloaded_on", date=date_str)) - date_label.setStyleSheet("color: #aaaaaa; font-size: 11px;") - info_layout.addWidget(date_label) - except Exception as e: - logger.debug(f"Error parsing date: {e}") + badge_rect = QRect(text_x, current_y, 80, 20) + painter.setBrush(QBrush(badge_color)) + painter.setPen(Qt.PenStyle.NoPen) + painter.drawRoundedRect(badge_rect, 3, 3) - # File size and type - file_size = self.entry.get("file_size", 0) - is_audio = self.entry.get("is_audio_only", False) + painter.setPen(QColor("white")) + font_badge = QFont() + font_badge.setBold(True) + font_badge.setPixelSize(10) + painter.setFont(font_badge) + painter.drawText(badge_rect, Qt.AlignmentFlag.AlignCenter, badge_text) - size_type_layout = QHBoxLayout() - - # File type badge - type_badge = QLabel(_("history.audio_download") if is_audio else _("history.video_download")) - type_badge.setStyleSheet(f""" - QLabel {{ - background-color: {'#c90000' if not is_audio else '#0066cc'}; - color: white; - padding: 2px 8px; - border-radius: 3px; - font-size: 10px; - font-weight: bold; - }} - """) - size_type_layout.addWidget(type_badge) - - # File size + # File Size + file_size = entry.get("file_size", 0) if file_size > 0: size_str = self.format_file_size(file_size) - size_label = QLabel(_("history.file_size", size=size_str)) - size_label.setStyleSheet("color: #aaaaaa; font-size: 11px;") - size_type_layout.addWidget(size_label) - - size_type_layout.addStretch() - info_layout.addLayout(size_type_layout) - - info_layout.addStretch() - - main_layout.addLayout(info_layout, 1) - - # Three-dot menu button - self.menu_button = QPushButton("⋮") - self.menu_button.setFixedSize(32, 32) - self.menu_button.setStyleSheet(""" - QPushButton { - background-color: #2a2d36; - border: none; - border-radius: 16px; - color: white; - font-size: 18px; - font-weight: bold; - padding-bottom: 5px; /* Adjust vertical alignment of dots */ - } - QPushButton:hover { - background-color: #3a3d46; - } - QPushButton:pressed { - background-color: #c90000; - } - """) - self.menu_button.clicked.connect(self.show_menu) - - main_layout.addWidget(self.menu_button, alignment=Qt.AlignmentFlag.AlignTop) - - def load_thumbnail(self): - """Load and display the thumbnail.""" - thumbnail_url = self.entry.get("thumbnail_url") - - if not thumbnail_url: - self.set_placeholder_thumbnail() - return - - # Check if thumbnail is cached - thumbnail_filename = f"{self.entry_id}.jpg" - thumbnail_path = APP_THUMBNAILS_DIR / thumbnail_filename - - if thumbnail_path.exists(): - try: - pixmap = QPixmap(str(thumbnail_path)) - if not pixmap.isNull(): - # Don't scale here, let setScaledContents handle it - self.thumbnail_label.setPixmap(pixmap) - return - except Exception as e: - logger.debug(f"Error loading cached thumbnail: {e}") - - # If not in cache, just set placeholder. - # Background thread will download it and notify parent to reload. - self.set_placeholder_thumbnail() + painter.setPen(QColor("#aaaaaa")) + painter.setFont(QFont("Arial", 11)) + painter.drawText(badge_rect.right() + 10, current_y + 14, size_str) + + # Menu Button + menu_rect = self.get_menu_rect(card_rect) + + # Check hover on menu button specifically + mouse_pos = QCursor.pos() + if option.widget: + mouse_pos = option.widget.mapFromGlobal(mouse_pos) + + if menu_rect.contains(mouse_pos): + painter.setPen(QColor("#c90000")) + else: + painter.setPen(QColor("#ffffff")) + + painter.setFont(QFont("Arial", 18, QFont.Weight.Bold)) + painter.drawText(menu_rect, Qt.AlignmentFlag.AlignCenter, "⋮") + + painter.restore() + + def get_menu_rect(self, card_rect): + return QRect(card_rect.right() - 40, card_rect.top() + 10, 30, 30) + + def editorEvent(self, event, model, option, index): + """Handle mouse clicks.""" + if event.type() == QEvent.Type.MouseButtonRelease: + if event.button() == Qt.MouseButton.LeftButton: + # Use same margins as paint to ensure hit consistency + card_rect = option.rect.adjusted(self.h_margin, self.v_margin, -self.h_margin, -self.v_margin) + menu_rect = self.get_menu_rect(card_rect) + + if menu_rect.contains(event.pos()): + self.menu_clicked.emit(index, event.globalPos()) + return True + + return super().editorEvent(event, model, option, index) - def reload_thumbnail(self): - """Reload thumbnail from cache (called when background download finishes).""" - self.load_thumbnail() - - def set_placeholder_thumbnail(self): - """Set a placeholder when thumbnail is not available.""" - self.thumbnail_label.setText("📹" if not self.entry.get("is_audio_only") else "🎵") - self.thumbnail_label.setStyleSheet(""" - QLabel { - border: 1px solid #3d3d3d; - border-radius: 4px; - background-color: #15181b; - color: #666666; - font-size: 48px; - } - """) - def format_file_size(self, size_bytes: int) -> str: - """Format file size in human-readable format.""" for unit in ['B', 'KB', 'MB', 'GB']: if size_bytes < 1024.0: return f"{size_bytes:.1f} {unit}" size_bytes /= 1024.0 return f"{size_bytes:.1f} TB" + + +class HistoryDialog(QDialog): + """Dialog to display and manage download history.""" - def show_menu(self): - """Show the context menu with options.""" + redownload_requested = Signal(dict) + + def __init__(self, parent: Optional["YTSageApp"] = None): + super().__init__(parent) + self.parent_app = parent + + self.setup_ui() + self.show_loading_state() + + QTimer.singleShot(100, self.start_loading_history) + + def setup_ui(self): + self.setWindowTitle(_("history.title")) + self.resize(850, 600) + self.setWindowFlags(Qt.WindowType.Dialog | Qt.WindowType.WindowCloseButtonHint) + self.setStyleSheet(""" + QDialog { background-color: #15181b; } + QLabel { color: #ffffff; } + """) + + layout = QVBoxLayout(self) + + # --- Header --- + header = QHBoxLayout() + title = QLabel(_("history.title")) + title.setStyleSheet("font-size: 18px; font-weight: bold;") + header.addWidget(title) + header.addStretch() + + self.clear_btn = QPushButton(_("history.clear_all")) + self.clear_btn.setStyleSheet(""" + QPushButton { + background-color: #c90000; color: white; padding: 6px 12px; + border: none; border-radius: 4px; font-weight: bold; + } + QPushButton:hover { background-color: #a50000; } + QPushButton:disabled { background-color: #555555; color: #aaaaaa; } + """) + self.clear_btn.clicked.connect(self.clear_all_history) + header.addWidget(self.clear_btn) + layout.addLayout(header) + + # --- Search --- + self.search_input = QLineEdit() + self.search_input.setPlaceholderText(_("history.search_placeholder")) + self.search_input.setStyleSheet(""" + QLineEdit { + padding: 8px; border: 2px solid #2a2d36; border-radius: 4px; + background-color: #1b2021; color: white; + } + """) + self.search_input.textChanged.connect(self.filter_history) + layout.addWidget(self.search_input) + + # --- List View --- + self.list_view = QListView() + self.list_view.setStyleSheet(""" + QListView { + background-color: transparent; + border: none; + outline: none; + } + QListView::item { + border: none; + background: transparent; + } + """) + self.list_view.setVerticalScrollMode(QListView.ScrollMode.ScrollPerPixel) + self.list_view.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff) + self.list_view.setUniformItemSizes(True) + self.list_view.setSelectionMode(QListView.SelectionMode.NoSelection) + self.list_view.setMouseTracking(True) + self.list_view.setResizeMode(QListView.ResizeMode.Adjust) + + self.model = HistoryModel([], self) + self.list_view.setModel(self.model) + + self.delegate = HistoryDelegate(self.list_view) + self.delegate.menu_clicked.connect(self.show_context_menu) + self.list_view.setItemDelegate(self.delegate) + + self.list_view.setContextMenuPolicy(Qt.ContextMenuPolicy.CustomContextMenu) + self.list_view.customContextMenuRequested.connect(self.on_context_menu_requested) + + layout.addWidget(self.list_view) + + # --- Status --- + self.status_label = QLabel() + self.status_label.setStyleSheet("color: #aaaaaa; font-size: 12px;") + layout.addWidget(self.status_label) + + def show_loading_state(self): + self.status_label.setText("Loading history...") + self.clear_btn.setEnabled(False) + + def start_loading_history(self): + self.loader_thread = HistoryLoaderThread() + self.loader_thread.entries_loaded.connect(self.on_entries_loaded) + self.loader_thread.thumbnail_loaded.connect(self.on_thumbnail_loaded) + self.loader_thread.start() + + def on_entries_loaded(self, entries): + self.model.update_entries(entries) + + count = len(entries) + if count == 0: + self.status_label.setText(_("history.no_history")) + self.clear_btn.setEnabled(False) + else: + self.status_label.setText(_("history.entries_count", count=count)) + self.clear_btn.setEnabled(True) + + self.load_cached_thumbnails(entries) + + def load_cached_thumbnails(self, entries): + for entry in entries: + eid = entry.get("id") + if not eid: continue + + p = APP_THUMBNAILS_DIR / f"{eid}.jpg" + if p.exists(): + pix = QPixmap(str(p)) + if not pix.isNull(): + self.model.thumbnail_cache[eid] = pix + + def on_thumbnail_loaded(self, entry_id, data_bytes): + pixmap = QPixmap() + pixmap.loadFromData(data_bytes) + if not pixmap.isNull(): + self.model.update_thumbnail(entry_id, pixmap) + + def on_context_menu_requested(self, pos): + index = self.list_view.indexAt(pos) + if index.isValid(): + global_pos = self.list_view.mapToGlobal(pos) + self.show_context_menu(index, global_pos) + + def show_context_menu(self, index, global_pos): + entry = index.data(HistoryModel.EntryRole) + if not entry: return + menu = QMenu(self) menu.setStyleSheet(""" QMenu { - background-color: #2a2d36; - border: 1px solid #3a3d46; - color: white; - padding: 5px; + background-color: #2a2d36; border: 1px solid #3a3d46; color: white; } QMenu::item { padding: 8px 20px; - border-radius: 4px; } QMenu::item:selected { background-color: #c90000; } """) - # Open file location - open_action = menu.addAction("📁 " + _("history.open_location")) - open_action.triggered.connect(self.open_file_location) - - # Redownload - redownload_action = menu.addAction("⬇️ " + _("history.redownload")) - redownload_action.triggered.connect(self.redownload) - + act_open = menu.addAction("📁 " + _("history.open_location")) + act_redownload = menu.addAction("⬇️ " + _("history.redownload")) menu.addSeparator() + act_remove = menu.addAction("🗑️ " + _("history.remove")) - # Remove from history - remove_action = menu.addAction("🗑️ " + _("history.remove")) - remove_action.triggered.connect(self.remove_from_history) + action = menu.exec(global_pos) - # Show menu at button position - menu.exec(self.menu_button.mapToGlobal(self.menu_button.rect().bottomLeft())) - - def open_file_location(self): - """Open the file location in the system file explorer.""" - file_path = Path(self.entry.get("file_path", "")) + if action == act_open: + self.open_file_location(entry) + elif action == act_redownload: + self.redownload_entry(entry) + elif action == act_remove: + self.remove_entry(index) + + def open_file_location(self, entry): + path_str = entry.get("file_path", "") + if not path_str: return - if not file_path.exists(): - QMessageBox.warning( - self, - _("history.file_not_found"), - _("history.file_not_found_message", path=str(file_path)) - ) + path = Path(path_str) + if not path.exists(): + QMessageBox.warning(self, "Error", f"File not found: {path}") return - - try: - # On Windows, use explorer with /select to highlight the file - if os.name == "nt": - subprocess.run(['explorer', '/select,', str(file_path)], creationflags=SUBPROCESS_CREATIONFLAGS) - # On macOS, use open with -R to reveal in Finder - elif subprocess.sys.platform == "darwin": - subprocess.run(['open', '-R', str(file_path)]) - # On Linux, try to open the folder - else: - folder_path = file_path.parent - subprocess.run(['xdg-open', str(folder_path)]) - logger.info(f"Opened file location: {file_path}") + try: + if os.name == "nt": + subprocess.run(['explorer', '/select,', str(path)], creationflags=SUBPROCESS_CREATIONFLAGS) + elif subprocess.sys.platform == "darwin": + subprocess.run(['open', '-R', str(path)]) + else: + folder_path = path.parent + subprocess.run(['xdg-open', str(folder_path)]) except Exception as e: - logger.exception(f"Error opening file location: {e}") - QMessageBox.warning(self, "Error", f"Could not open file location: {str(e)}") - - def redownload(self): - """Request redownload of this entry.""" + logger.error(f"Failed to open file: {e}") + + def redownload_entry(self, entry): reply = QMessageBox.question( self, _("history.redownload_confirm_title"), - _("history.redownload_confirm_message", title=self.entry.get("title", "")), + _("history.redownload_confirm_message", title=entry.get("title")), QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No ) - if reply == QMessageBox.StandardButton.Yes: - self.redownload_requested.emit(self.entry) - - def remove_from_history(self): - """Request removal of this entry from history.""" + self.redownload_requested.emit(entry) + self.accept() + + def remove_entry(self, index): + entry = index.data(HistoryModel.EntryRole) reply = QMessageBox.question( self, _("history.remove_confirm_title"), - _("history.remove_confirm_message", title=self.entry.get("title", "")), + _("history.remove_confirm_message", title=entry.get("title")), QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No ) - if reply == QMessageBox.StandardButton.Yes: - self.remove_requested.emit(self.entry_id) + if HistoryManager.remove_entry(entry.get("id")): + self.model.remove_item(index.row()) + self.status_label.setText( + _("history.entries_count", count=self.model.rowCount()) + ) - -class HistoryDialog(QDialog): - """Dialog to display and manage download history.""" - - redownload_requested = Signal(dict) # Signal to request redownload in main window - - def __init__(self, parent: Optional["YTSageApp"] = None): - super().__init__(parent) - self.parent_app = parent - self.entry_widgets = [] - self.entry_widgets_map = {} # Map entry_id -> widget - - self.setup_ui() - - # Show loading state initially - self.show_loading_state() - - # Start loading history in background after a short delay - # to ensure the dialog is shown first - QTimer.singleShot(100, self.start_loading_history) - - def start_loading_history(self): - """Start the background thread to load history.""" - self.loader_thread = HistoryLoaderThread() - self.loader_thread.entries_loaded.connect(self.on_history_loaded) - self.loader_thread.thumbnail_loaded.connect(self.update_entry_thumbnail) - self.loader_thread.start() - - def on_history_loaded(self, entries): - """Called when history is loaded from background thread.""" - self.load_history_entries(entries) - - def update_entry_thumbnail(self, entry_id: str): - """Called when a thumbnail is downloaded in the background.""" - if entry_id in self.entry_widgets_map: - self.entry_widgets_map[entry_id].reload_thumbnail() - - def setup_ui(self): - """Setup the dialog UI.""" - self.setWindowTitle(_("history.title")) - self.setMinimumSize(700, 500) - self.resize(850, 600) - - # Set window flags - self.setWindowFlags(Qt.WindowType.Dialog | Qt.WindowType.WindowCloseButtonHint) - - layout = QVBoxLayout(self) - layout.setSpacing(10) - layout.setContentsMargins(20, 20, 20, 20) - - # Header with title and buttons - header_layout = QHBoxLayout() - - title_label = QLabel(_("history.title")) - title_label.setStyleSheet(""" - QLabel { - font-size: 18px; - font-weight: bold; - color: white; - } - """) - header_layout.addWidget(title_label) - - header_layout.addStretch() - - # Clear all button - self.clear_all_btn = QPushButton(_("history.clear_all")) - self.clear_all_btn.setStyleSheet(""" - QPushButton { - background-color: #c90000; - color: white; - padding: 8px 16px; - border: none; - border-radius: 4px; - font-weight: bold; - } - QPushButton:hover { - background-color: #a50000; - } - QPushButton:pressed { - background-color: #800000; - } - """) - self.clear_all_btn.clicked.connect(self.clear_all_history) - header_layout.addWidget(self.clear_all_btn) - - layout.addLayout(header_layout) - - # Search bar - self.search_input = QLineEdit() - self.search_input.setPlaceholderText(_("history.search_placeholder")) - self.search_input.setStyleSheet(""" - QLineEdit { - padding: 10px; - border: 2px solid #1b2021; - border-radius: 4px; - background-color: #1b2021; - color: #ffffff; - font-size: 13px; - } - """) - self.search_input.textChanged.connect(self.filter_history) - layout.addWidget(self.search_input) - - # Scroll area for history entries - scroll_area = QScrollArea() - scroll_area.setWidgetResizable(True) - scroll_area.setStyleSheet(""" - QScrollArea { - border: none; - background-color: transparent; - } - """) - - # Container for history entries - self.history_container = QWidget() - self.history_layout = QVBoxLayout(self.history_container) - self.history_layout.setSpacing(10) - self.history_layout.setContentsMargins(0, 0, 0, 0) - self.history_layout.setAlignment(Qt.AlignmentFlag.AlignTop) - - scroll_area.setWidget(self.history_container) - layout.addWidget(scroll_area) - - # Status label - self.status_label = QLabel() - self.status_label.setStyleSheet("color: #aaaaaa; font-size: 12px;") - layout.addWidget(self.status_label) - - # Apply dark theme - self.setStyleSheet(""" - QDialog { - background-color: #15181b; - } - QLabel { - color: #ffffff; - } - """) - - def clear_history_list(self): - """Clear existing history widgets.""" - for widget in self.entry_widgets: - widget.deleteLater() - self.entry_widgets.clear() - self.entry_widgets_map.clear() - - def show_loading_state(self): - """Show loading indicator.""" - # Clear existing content - self.clear_history_list() - - loading_widget = QWidget() - loading_layout = QVBoxLayout(loading_widget) - loading_layout.setAlignment(Qt.AlignmentFlag.AlignCenter) - - spinner_label = QLabel("⏳") # Simple spinner icon - spinner_label.setStyleSheet("font-size: 48px; color: #c90000;") - spinner_label.setAlignment(Qt.AlignmentFlag.AlignCenter) - loading_layout.addWidget(spinner_label) - - # text_label removed as per user request - - self.history_layout.addWidget(loading_widget) - self.entry_widgets.append(loading_widget) - - self.status_label.setText("Loading...") - self.clear_all_btn.setEnabled(False) - - def load_history(self): - """Load and display history entries.""" - if hasattr(self, 'history_container'): - self.show_loading_state() - self.start_loading_history() - - def load_history_entries(self, entries): - """Populate the history list with entries.""" - self.clear_history_list() - - if not entries: - self.show_empty_state() - return - - # Create widgets for each entry - for entry in entries: - widget = HistoryEntryWidget(entry, self.history_container) - widget.remove_requested.connect(self.remove_entry) - widget.redownload_requested.connect(self.handle_redownload) - - self.history_layout.addWidget(widget) - - # Map widget by ID for updates - entry_id = entry.get("id") - if entry_id: - self.entry_widgets_map[entry_id] = widget - self.entry_widgets.append(widget) - - # Update status - count = len(entries) - if count == 1: - status_text = _("history.one_entry") - else: - status_text = _("history.entries_count", count=count) - self.status_label.setText(status_text) - self.clear_all_btn.setEnabled(True) - - def show_empty_state(self): - """Show empty state when there's no history.""" - empty_widget = QWidget() - empty_layout = QVBoxLayout(empty_widget) - empty_layout.setAlignment(Qt.AlignmentFlag.AlignCenter) - - icon_label = QLabel("📂") - icon_label.setStyleSheet("font-size: 64px; color: #555555;") - icon_label.setAlignment(Qt.AlignmentFlag.AlignCenter) - empty_layout.addWidget(icon_label) - - title_label = QLabel(_("history.no_history")) - title_label.setStyleSheet("font-size: 16px; color: #888888; font-weight: bold;") - title_label.setAlignment(Qt.AlignmentFlag.AlignCenter) - empty_layout.addWidget(title_label) - - desc_label = QLabel(_("history.no_history_description")) - desc_label.setStyleSheet("font-size: 13px; color: #666666;") - desc_label.setAlignment(Qt.AlignmentFlag.AlignCenter) - empty_layout.addWidget(desc_label) - - self.history_layout.addWidget(empty_widget) - self.entry_widgets.append(empty_widget) - - self.status_label.setText("") - self.clear_all_btn.setEnabled(False) - - def filter_history(self, query: str): - """Filter history entries based on search query.""" - if not query: - # Show all entries - for widget in self.entry_widgets: - widget.show() - return - - # Hide/show based on query - query_lower = query.lower() - visible_count = 0 - - for widget in self.entry_widgets: - if isinstance(widget, HistoryEntryWidget): - title = (widget.entry.get("title") or "").lower() - channel = (widget.entry.get("channel") or "").lower() - - if query_lower in title or query_lower in channel: - widget.show() - visible_count += 1 - else: - widget.hide() - - def remove_entry(self, entry_id: str): - """Remove an entry from history.""" - success = HistoryManager.remove_entry(entry_id) - - if success: - # Reload history - self.load_history() - logger.info(f"Removed entry from history: {entry_id}") - else: - QMessageBox.warning(self, "Error", "Failed to remove entry from history") - def clear_all_history(self): - """Clear all history entries.""" reply = QMessageBox.question( self, _("history.clear_confirm_title"), _("history.clear_confirm_message"), QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No ) - if reply == QMessageBox.StandardButton.Yes: - count = HistoryManager.clear_history() - self.load_history() - logger.info(f"Cleared all history: {count} entries") - - def handle_redownload(self, entry: dict): - """Handle redownload request.""" - # Emit signal to parent window - self.redownload_requested.emit(entry) - - # Close dialog - self.accept() + HistoryManager.clear_history() + self.model.update_entries([]) + self.clear_btn.setEnabled(False) + self.status_label.setText(_("history.no_history")) + + def filter_history(self, query): + results = HistoryManager.search_entries(query) + self.model.update_entries(results) + self.load_cached_thumbnails(results) diff --git a/src/utils/ytsage_history_manager.py b/src/utils/ytsage_history_manager.py index b915f00..90fdbbe 100644 --- a/src/utils/ytsage_history_manager.py +++ b/src/utils/ytsage_history_manager.py @@ -3,144 +3,259 @@ History Manager Module ====================== This module provides **thread-safe** centralized management for download -history in YTSage. It handles reading, writing, and managing download history -stored in a JSON file. - -Thread safety is ensured using a reentrant lock (`RLock`), so multiple threads -can safely access or modify history concurrently. +history in YTSage using SQLite for high performance and scalability. Features -------- -- Thread-safe operations for getting, adding, and removing history entries. -- Loads history from a JSON file (`APP_HISTORY_FILE`). -- Creates the history file if missing or corrupt. -- Manages download history with metadata including thumbnails, file paths, and download options. -- Provides safe error handling with logging instead of raising exceptions. -- Persists updates back to disk automatically. +- Scalable: Uses SQLite instead of parsing potentially large JSON files. +- Thread-safe: Handles database connections safely. +- Migration: Automatically migrates legacy JSON history to SQLite. +- CRUD: Create, Read, Delete, Clear operations for history entries. Usage ----- from src.utils.ytsage_history_manager import HistoryManager # Add a download to history -HistoryManager.add_entry( - title="Video Title", - url="https://youtube.com/watch?v=...", - thumbnail_url="https://...", - file_path="/path/to/file.mp4", - format_id="137+140", - is_audio_only=False, - resolution="1080p", - download_options={...} -) +HistoryManager.add_entry(...) # Get all history entries history = HistoryManager.get_all_entries() +# Get recent entries (limit + offset support planned) # Remove an entry HistoryManager.remove_entry(entry_id) # Clear all history HistoryManager.clear_history() - -Design Notes ------------- -- History entries are stored in `HistoryManager._history` (a list of dicts). -- Each entry has a unique ID based on timestamp. -- All modifications trigger a save (`_save`) to keep JSON in sync. -- Logs actions and errors using the app's central logger. -- Uses `RLock` to allow safe concurrent access from multiple threads. - -Exceptions ----------- -- Any issues during file I/O (permissions, disk errors, JSON corruption) - are caught and logged. The application continues running with an empty - history when possible. """ import json +import sqlite3 import threading import time from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional -from src.utils.ytsage_constants import APP_HISTORY_FILE +from src.utils.ytsage_constants import APP_HISTORY_FILE, APP_DATA_DIR from src.utils.ytsage_logger import logger class HistoryManager: """ - Thread-safe history manager for YTSage. - - Provides methods to load, save, get, add, and remove download history entries. - Automatically persists changes to disk. + Thread-safe history manager for YTSage using SQLite. """ _lock = threading.RLock() - _history_file = APP_HISTORY_FILE - _history: List[Dict[str, Any]] = [] - _loaded = False + # Define DB file next to the old JSON file + _db_file = APP_DATA_DIR / "ytsage_history.db" + _initialized = False @classmethod - def _load(cls) -> None: - """ - Loads download history from a JSON file if it exists and is valid. - If the file is missing or corrupt, initializes with an empty history. - Logs actions and errors during the process. - """ - with cls._lock: - if cls._history_file.exists(): - try: - with open(cls._history_file, "r", encoding="utf-8") as f: - data = json.load(f) - # Ensure it's a list - if isinstance(data, list): - cls._history = data - else: - cls._history = [] - logger.warning("History file format invalid, initialized empty history.") - logger.info(f"History loaded from file: {len(cls._history)} entries.") - except json.JSONDecodeError: - cls._history = [] - logger.warning("History file corrupt, initialized empty history.") - except Exception as e: - cls._history = [] - logger.error(f"Error loading history file: {e}") - else: - cls._history = [] - cls._save() - logger.info("History file not found, created empty history.") - cls._loaded = True + def _init_db(cls): + """Initialize the database: create table and migrate if needed.""" + if cls._initialized: + return - @classmethod - def _save(cls) -> None: - """ - Save current history to JSON file. - - Note: - May raise exceptions if the file cannot be written due to permission issues, - disk errors, or other I/O problems. - """ with cls._lock: + # Check if file exists to know if we need to migrate or just create schema + db_exists = cls._db_file.exists() + legacy_json_exists = APP_HISTORY_FILE.exists() + try: - # Ensure parent directory exists - cls._history_file.parent.mkdir(parents=True, exist_ok=True) + # Ensure directory exists + cls._db_file.parent.mkdir(parents=True, exist_ok=True) - with open(cls._history_file, "w", encoding="utf-8") as f: - json.dump(cls._history, f, indent=2, ensure_ascii=False) - logger.debug(f"History saved to file: {len(cls._history)} entries.") - except (OSError, PermissionError) as e: - logger.exception(f"Failed to save history: {e}") - except Exception as e: - logger.exception(f"Unexpected error while saving history: {e}") + with sqlite3.connect(cls._db_file, check_same_thread=False) as conn: + cursor = conn.cursor() + + # Create table + cursor.execute(""" + CREATE TABLE IF NOT EXISTS history ( + id TEXT PRIMARY KEY, + title TEXT, + url TEXT, + channel TEXT, + file_path TEXT, + download_date TEXT, + file_size INTEGER, + thumbnail_url TEXT, + format_id TEXT, + resolution TEXT, + is_audio_only INTEGER, + duration TEXT, + options TEXT, + timestamp REAL + ) + """) + + # Index for faster sorting by date + cursor.execute(""" + CREATE INDEX IF NOT EXISTS idx_timestamp + ON history (timestamp DESC) + """) + + conn.commit() + + # If we just created the DB and have a JSON file, migrate + if not db_exists and legacy_json_exists: + cls._migrate_legacy_json() + + cls._initialized = True + + except sqlite3.Error as e: + logger.error(f"Failed to initialize history database: {e}") @classmethod - def _ensure_loaded(cls) -> None: - """Ensure history is loaded before any operation.""" - with cls._lock: - if not cls._loaded: - cls._load() + def _migrate_legacy_json(cls): + """Migrate legacy JSON history to SQLite.""" + logger.info("Migrating legacy history JSON to SQLite...") + try: + with open(APP_HISTORY_FILE, "r", encoding="utf-8") as f: + data = json.load(f) + + if isinstance(data, list): + count = 0 + with sqlite3.connect(cls._db_file, check_same_thread=False) as conn: + cursor = conn.cursor() + for entry in data: + try: + # Safely extract download_options logic if complex + options_json = json.dumps(entry.get("download_options", {})) + + # Construct timestamp from isoformat if missing + ts = entry.get("timestamp") + if not ts and "download_date" in entry: + try: + dt = datetime.fromisoformat(entry["download_date"]) + ts = dt.timestamp() + except Exception: + ts = time.time() + + cursor.execute(""" + INSERT OR IGNORE INTO history ( + id, title, url, channel, file_path, download_date, + file_size, thumbnail_url, format_id, resolution, + is_audio_only, duration, options, timestamp + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, ( + entry.get("id", str(int(time.time()*1000))), + entry.get("title", ""), + entry.get("url", ""), + entry.get("channel", "Unknown"), + entry.get("file_path", ""), + entry.get("download_date", ""), + entry.get("file_size", 0), + entry.get("thumbnail_url", ""), + entry.get("format_id", ""), + entry.get("resolution", ""), + 1 if entry.get("is_audio_only") else 0, + entry.get("duration", ""), + options_json, + ts or time.time() + )) + count += 1 + except Exception as e: + logger.error(f"Skipped invalid entry during migration: {e}") + + conn.commit() + + logger.info(f"Successfully migrated {count} history entries.") + + # Rename old JSON to .bak to avoid re-migration, or keep as backup + try: + APP_HISTORY_FILE.rename(APP_HISTORY_FILE.with_suffix(".json.bak")) + except Exception as e: + logger.warning(f"Could not rename legacy history file: {e}") + + except Exception as e: + logger.error(f"Migration failed: {e}") + + @classmethod + def _get_connection(cls): + """Get a database connection.""" + cls._init_db() + return sqlite3.connect(cls._db_file, check_same_thread=False) + + @classmethod + def get_all_entries(cls, limit: Optional[int] = None) -> List[Dict[str, Any]]: + """ + Get all history entries, sorted by most recent first. + + Args: + limit: Optional limit on number of entries to return (most recent first) + + Returns: + List of dictionary entries. + """ + entries = [] + try: + with cls._lock: # Lock for simple concurrency safety + with cls._get_connection() as conn: + # Return dict-like rows + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + + query = "SELECT * FROM history ORDER BY timestamp DESC" + params = () + + if limit is not None: + query += " LIMIT ?" + params = (limit,) + + cursor.execute(query, params) + rows = cursor.fetchall() + + for row in rows: + entry = dict(row) + # Convert boolean back + entry["is_audio_only"] = bool(entry["is_audio_only"]) + # Parse options JSON + try: + entry["download_options"] = json.loads(entry["options"]) if entry["options"] else {} + except json.JSONDecodeError: + entry["download_options"] = {} + del entry["options"] # Remove internal column + entries.append(entry) + + except Exception as e: + logger.error(f"Error fetching history: {e}") + + return entries + + @classmethod + def get_entry(cls, entry_id: str) -> Optional[Dict[str, Any]]: + """ + Get a specific history entry by ID. + + Args: + entry_id: The unique entry ID + + Returns: + Entry dictionary or None if not found + """ + try: + with cls._lock: + with cls._get_connection() as conn: + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + cursor.execute("SELECT * FROM history WHERE id = ?", (entry_id,)) + row = cursor.fetchone() + + if row: + entry = dict(row) + entry["is_audio_only"] = bool(entry["is_audio_only"]) + try: + entry["download_options"] = json.loads(entry["options"]) if entry["options"] else {} + except json.JSONDecodeError: + entry["download_options"] = {} + del entry["options"] + return entry + return None + except Exception as e: + logger.error(f"Error fetching entry {entry_id}: {e}") + return None @classmethod def add_entry( @@ -158,141 +273,99 @@ class HistoryManager: download_options: Optional[Dict[str, Any]] = None, ) -> str: """ - Add a new download entry to history. - - Args: - title: Video/audio title - url: Original URL - thumbnail_url: Thumbnail URL (can be None) - file_path: Path to downloaded file - format_id: Format ID used for download - is_audio_only: Whether it's audio-only download - resolution: Resolution string (e.g., "1080p", "best audio") - file_size: File size in bytes (optional) - channel: Channel name (optional) - duration: Duration string (optional) - download_options: Dictionary of all download options used (optional) - - Returns: - str: The unique ID of the created entry + Add a new entry to the download history. """ - cls._ensure_loaded() + if download_options is None: + download_options = {} + + timestamp = time.time() + # Ensure unique ID + unique_id = f"{int(timestamp * 1000)}" + download_date = datetime.fromtimestamp(timestamp).isoformat() - with cls._lock: - # Generate unique ID based on timestamp - entry_id = f"{int(time.time() * 1000)}" - - # Get file size if not provided - if file_size is None: - try: - file_path_obj = Path(file_path) - if file_path_obj.exists(): - file_size = file_path_obj.stat().st_size - except Exception as e: - logger.debug(f"Could not get file size: {e}") - file_size = 0 - - entry = { - "id": entry_id, - "title": title, - "url": url, - "thumbnail_url": thumbnail_url, - "file_path": file_path, - "download_date": datetime.now().isoformat(), - "format_id": format_id, - "is_audio_only": is_audio_only, - "resolution": resolution, - "file_size": file_size or 0, - "channel": channel, - "duration": duration, - "download_options": download_options or {}, - } - - # Add to beginning of list (most recent first) - cls._history.insert(0, entry) - cls._save() - - logger.info(f"Added entry to history: {title}") - return entry_id + # Determine file size if not provided + if file_size is None: + try: + p = Path(file_path) + if p.exists(): + file_size = p.stat().st_size + except Exception: + file_size = 0 - @classmethod - def get_all_entries(cls, limit: Optional[int] = None) -> List[Dict[str, Any]]: - """ - Retrieve all history entries. - - Args: - limit: Optional limit on number of entries to return (most recent first) - - Returns: - List of history entry dictionaries - """ - cls._ensure_loaded() + # Allow None for optional strings + channel = channel or "Unknown" + duration = duration or "" + thumbnail_url = thumbnail_url or "" - with cls._lock: - if limit is not None: - return cls._history[:limit] - return cls._history.copy() - - @classmethod - def get_entry(cls, entry_id: str) -> Optional[Dict[str, Any]]: - """ - Get a specific history entry by ID. - - Args: - entry_id: The unique entry ID - - Returns: - Entry dictionary or None if not found - """ - cls._ensure_loaded() - - with cls._lock: - for entry in cls._history: - if entry.get("id") == entry_id: - return entry.copy() - return None + try: + with cls._lock: + with cls._get_connection() as conn: + cursor = conn.cursor() + + cursor.execute(""" + INSERT INTO history ( + id, title, url, channel, file_path, download_date, + file_size, thumbnail_url, format_id, resolution, + is_audio_only, duration, options, timestamp + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, ( + unique_id, + title, + url, + channel, + str(file_path), + download_date, + file_size, + thumbnail_url, + format_id, + resolution, + 1 if is_audio_only else 0, + duration, + json.dumps(download_options), + timestamp + )) + conn.commit() + + logger.info(f"Added history entry: {title}") + return unique_id + + except Exception as e: + logger.error(f"Error adding history entry: {e}") + return "" @classmethod def remove_entry(cls, entry_id: str) -> bool: - """ - Remove a specific entry from history. - - Args: - entry_id: The unique entry ID to remove - - Returns: - bool: True if entry was found and removed, False otherwise - """ - cls._ensure_loaded() - - with cls._lock: - for i, entry in enumerate(cls._history): - if entry.get("id") == entry_id: - removed = cls._history.pop(i) - cls._save() - logger.info(f"Removed entry from history: {removed.get('title', 'Unknown')}") - return True - - logger.debug(f"Entry ID '{entry_id}' not found in history.") + """Remove an entry from history by ID.""" + try: + with cls._lock: + with cls._get_connection() as conn: + cursor = conn.cursor() + cursor.execute("DELETE FROM history WHERE id = ?", (entry_id,)) + if cursor.rowcount > 0: + conn.commit() + logger.info(f"Removed history entry: {entry_id}") + return True + return False + except Exception as e: + logger.error(f"Error removing history entry: {e}") return False @classmethod def clear_history(cls) -> int: - """ - Clear all history entries. - - Returns: - int: Number of entries that were cleared - """ - cls._ensure_loaded() - - with cls._lock: - count = len(cls._history) - cls._history = [] - cls._save() - logger.info(f"Cleared all history: {count} entries removed.") + """Clear all history entries.""" + try: + with cls._lock: + with cls._get_connection() as conn: + cursor = conn.cursor() + cursor.execute("DELETE FROM history") + count = cursor.rowcount + conn.commit() + logger.info("History cleared") return count - + except Exception as e: + logger.error(f"Error clearing history: {e}") + return 0 + @classmethod def search_entries(cls, query: str) -> List[Dict[str, Any]]: """ @@ -304,45 +377,34 @@ class HistoryManager: Returns: List of matching history entries """ - cls._ensure_loaded() - if not query: return cls.get_all_entries() - - query_lower = query.lower() - - with cls._lock: - results = [] - for entry in cls._history: - # Search in title, channel, and URL - title = (entry.get("title") or "").lower() - channel = (entry.get("channel") or "").lower() - url = (entry.get("url") or "").lower() - - if query_lower in title or query_lower in channel or query_lower in url: - results.append(entry.copy()) - return results - - @classmethod - def get_statistics(cls) -> Dict[str, Any]: - """ - Get statistics about download history. - - Returns: - Dictionary with statistics (total_downloads, total_size, etc.) - """ - cls._ensure_loaded() - - with cls._lock: - total_downloads = len(cls._history) - total_size = sum(entry.get("file_size", 0) for entry in cls._history) - video_count = sum(1 for entry in cls._history if not entry.get("is_audio_only", False)) - audio_count = sum(1 for entry in cls._history if entry.get("is_audio_only", False)) + entries = [] + try: + search_pattern = f"%{query}%" + with cls._lock: + with cls._get_connection() as conn: + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + cursor.execute(""" + SELECT * FROM history + WHERE title LIKE ? OR channel LIKE ? OR url LIKE ? + ORDER BY timestamp DESC + """, (search_pattern, search_pattern, search_pattern)) + rows = cursor.fetchall() + + for row in rows: + entry = dict(row) + entry["is_audio_only"] = bool(entry["is_audio_only"]) + try: + entry["download_options"] = json.loads(entry["options"]) if entry["options"] else {} + except json.JSONDecodeError: + entry["download_options"] = {} + del entry["options"] + entries.append(entry) + + except Exception as e: + logger.error(f"Error searching history: {e}") - return { - "total_downloads": total_downloads, - "total_size": total_size, - "video_count": video_count, - "audio_count": audio_count, - } + return entries