Houmeres/SageTube
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

Refactor history dialog to use virtualized list and SQLite

Browse Source 
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.
This commit is contained in:
oop7
2026-01-21 17:56:56 +02:00
parent 29f713cf17
commit 5f17fabc50
2 changed files with 752 additions and 782 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gui/ytsage_gui_dialogs/ytsage_dialogs_history.py
+416 -508
View File
File diff suppressed because it is too large Load Diff
src/utils/ytsage_history_manager.py
+311 -249
View File
@@ -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,140 +273,98 @@ 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 = {}
with cls._lock:
# Generate unique ID based on timestamp
entry_id = f"{int(time.time() * 1000)}"
timestamp = time.time()
# Ensure unique ID
unique_id = f"{int(timestamp * 1000)}"
download_date = datetime.fromtimestamp(timestamp).isoformat()
# 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
# 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
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 {},
}
# Allow None for optional strings
channel = channel or "Unknown"
duration = duration or ""
thumbnail_url = thumbnail_url or ""
# Add to beginning of list (most recent first)
cls._history.insert(0, entry)
cls._save()
try:
with cls._lock:
with cls._get_connection() as conn:
cursor = conn.cursor()
logger.info(f"Added entry to history: {title}")
return entry_id
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()
@classmethod
def get_all_entries(cls, limit: Optional[int] = None) -> List[Dict[str, Any]]:
"""
Retrieve all history entries.
logger.info(f"Added history entry: {title}")
return unique_id
Args:
limit: Optional limit on number of entries to return (most recent first)
Returns:
List of history entry dictionaries
"""
cls._ensure_loaded()
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
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()
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()
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()
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)
if query_lower in title or query_lower in channel or query_lower in url:
results.append(entry.copy())
except Exception as e:
logger.error(f"Error searching history: {e}")
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))
return {
"total_downloads": total_downloads,
"total_size": total_size,
"video_count": video_count,
"audio_count": audio_count,
}
return entries