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