fix: hotkey capture compatibility, add library metadata caching
- Fix KeyCaptureDialog for both old (int) and new (enum) PyQt6 enum types using universal _to_int() helper - New library_cache.py: JSON cache (~/.cache/neo-pod-desktop/library_cache.json) with mtime-based invalidation — second launch skips tag extraction for unchanged files (instant load for large libraries) - _scan_library(force=False) uses cache by default; Refresh/F5 uses force=True - _incremental_update_after_edit() updates cache + refreshes UI without full filesystem walk - Cache entries invalidated on file deletion
This commit is contained in:
parent
ba6935420d
commit
cbc888660f
@ -144,8 +144,11 @@ class KeyCaptureDialog(QDialog):
|
||||
if key in ignore:
|
||||
return
|
||||
|
||||
# Build the sequence
|
||||
keys = mod_mask.value | key.value
|
||||
# Build the sequence — handle both old (int) and new (enum) PyQt6 types
|
||||
def _to_int(val):
|
||||
return val if isinstance(val, int) else val.value
|
||||
|
||||
keys = _to_int(mod_mask) | _to_int(key)
|
||||
self._captured_seq = QSeq(keys)
|
||||
self._label.setText(f"Captured: {self._captured_seq.toString()}\nPress Close to accept")
|
||||
self.accept()
|
||||
|
||||
111
src/library_cache.py
Normal file
111
src/library_cache.py
Normal file
@ -0,0 +1,111 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Library metadata cache for neo-pod-desktop.
|
||||
Caches extracted tag data per file path with mtime-based invalidation.
|
||||
Stored as JSON in XDG cache directory.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from artwork.cache import xdg_cache_path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_CACHE_FILE = "library_cache.json"
|
||||
|
||||
|
||||
class LibraryCache:
|
||||
"""JSON-based cache of audio file metadata.
|
||||
|
||||
Keys are absolute file paths. Values are dicts of metadata
|
||||
plus an internal ``_mtime`` field for change detection.
|
||||
|
||||
Cached entries are returned only when the file still exists
|
||||
and its ``os.path.getmtime`` matches the stored value.
|
||||
"""
|
||||
|
||||
def __init__(self, cache_path: Optional[str] = None):
|
||||
if cache_path is None:
|
||||
cache_path = os.path.join(xdg_cache_path(""), _CACHE_FILE)
|
||||
self._path = cache_path
|
||||
self._data: Dict[str, Dict[str, Any]] = {}
|
||||
|
||||
def load(self) -> None:
|
||||
self._data = {}
|
||||
if not os.path.exists(self._path):
|
||||
return
|
||||
try:
|
||||
with open(self._path, "r", encoding="utf-8") as fh:
|
||||
self._data = json.load(fh)
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning("Failed to load library cache: %s", exc)
|
||||
self._data = {}
|
||||
|
||||
def save(self) -> None:
|
||||
self._data = {k: v for k, v in self._data.items()
|
||||
if os.path.exists(k)}
|
||||
os.makedirs(os.path.dirname(self._path), exist_ok=True)
|
||||
try:
|
||||
with open(self._path, "w", encoding="utf-8") as fh:
|
||||
json.dump(self._data, fh, ensure_ascii=False)
|
||||
except OSError as exc:
|
||||
logger.warning("Failed to save library cache: %s", exc)
|
||||
|
||||
def get(self, file_path: str) -> Optional[Dict[str, Any]]:
|
||||
entry = self._data.get(file_path)
|
||||
if entry is None:
|
||||
return None
|
||||
if not os.path.isfile(file_path):
|
||||
self._data.pop(file_path, None)
|
||||
return None
|
||||
actual_mtime = os.path.getmtime(file_path)
|
||||
if actual_mtime != entry.get("_mtime", 0):
|
||||
return None
|
||||
return {k: v for k, v in entry.items() if k != "_mtime"}
|
||||
|
||||
def put(self, file_path: str, data: Dict[str, Any]) -> None:
|
||||
entry = dict(data)
|
||||
try:
|
||||
entry["_mtime"] = os.path.getmtime(file_path)
|
||||
except OSError:
|
||||
return
|
||||
self._data[file_path] = entry
|
||||
|
||||
def remove(self, file_path: str) -> None:
|
||||
self._data.pop(file_path, None)
|
||||
|
||||
def clear(self) -> None:
|
||||
self._data.clear()
|
||||
if os.path.exists(self._path):
|
||||
try:
|
||||
os.remove(self._path)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._data)
|
||||
|
||||
|
||||
_library_cache: Optional[LibraryCache] = None
|
||||
|
||||
|
||||
def get_library_cache() -> LibraryCache:
|
||||
global _library_cache
|
||||
if _library_cache is None:
|
||||
_library_cache = LibraryCache()
|
||||
_library_cache.load()
|
||||
return _library_cache
|
||||
|
||||
|
||||
def save_library_cache() -> None:
|
||||
global _library_cache
|
||||
if _library_cache is not None:
|
||||
_library_cache.save()
|
||||
|
||||
|
||||
def invalidate_cache_for_path(file_path: str) -> None:
|
||||
cache = get_library_cache()
|
||||
cache.remove(file_path)
|
||||
172
src/main.py
172
src/main.py
@ -26,6 +26,7 @@ from audio_converter import AudioConverter
|
||||
from metadata_handler import MetadataHandler
|
||||
from metadata_editor import MetadataEditorDialog
|
||||
from ipod_device import IPodDevice
|
||||
from library_cache import get_library_cache, save_library_cache, invalidate_cache_for_path
|
||||
from artwork.cache import cover_cache
|
||||
from config_loader import ConfigLoader
|
||||
from hotkeys import HotkeyManager, KeyCaptureDialog
|
||||
@ -938,8 +939,13 @@ class MainWindow(QMainWindow):
|
||||
return f"{size_bytes / 1024:.0f} KB"
|
||||
return f"{size_bytes / (1024 * 1024):.1f} MB"
|
||||
|
||||
def _scan_library(self):
|
||||
"""Scan Output Directory for ready-to-transfer and source files."""
|
||||
def _scan_library(self, force: bool = False):
|
||||
"""Scan Output Directory for ready-to-transfer and source files.
|
||||
|
||||
Uses a JSON cache to skip metadata extraction for files whose mtime
|
||||
hasn't changed since they were last cached. Pass ``force=True`` to
|
||||
bypass the cache and re-read every file (refresh / F5).
|
||||
"""
|
||||
output_dir = self.output_dir_input.text().strip()
|
||||
if not output_dir or not os.path.isdir(output_dir):
|
||||
self.library_ready.clear()
|
||||
@ -949,7 +955,10 @@ class MainWindow(QMainWindow):
|
||||
|
||||
ready = []
|
||||
sources = []
|
||||
cache_hits = 0
|
||||
cache_misses = 0
|
||||
|
||||
cache = get_library_cache()
|
||||
handler = MetadataHandler()
|
||||
|
||||
for root, _dirs, files in os.walk(output_dir):
|
||||
@ -959,60 +968,35 @@ class MainWindow(QMainWindow):
|
||||
if not os.path.isfile(fpath):
|
||||
continue
|
||||
|
||||
cached = None if force else cache.get(fpath)
|
||||
|
||||
if cached is not None:
|
||||
cache_hits += 1
|
||||
if ext in AUDIO_EXTS:
|
||||
cached.pop("_mtime", None)
|
||||
ready.append(cached)
|
||||
elif ext in SOURCE_EXTS:
|
||||
cached.pop("_mtime", None)
|
||||
sources.append(cached)
|
||||
continue
|
||||
|
||||
cache_misses += 1
|
||||
|
||||
if ext in AUDIO_EXTS:
|
||||
size = os.path.getsize(fpath)
|
||||
cover_path = None
|
||||
try:
|
||||
info = handler.extract_tags(fpath)
|
||||
duration = info.duration if info else 0
|
||||
artist = info.artist if info else "Unknown"
|
||||
title = info.title if info else fname
|
||||
album = info.album if info else ""
|
||||
genre = info.genre if info else ""
|
||||
track_num = info.track_number if info else 0
|
||||
cover_path = info.cover_path if info else None
|
||||
except Exception:
|
||||
duration = 0
|
||||
artist = "Unknown"
|
||||
title = fname
|
||||
album = ""
|
||||
genre = ""
|
||||
track_num = 0
|
||||
|
||||
ready.append({
|
||||
"path": fpath,
|
||||
"title": title,
|
||||
"artist": artist,
|
||||
"album": album,
|
||||
"duration_s": duration,
|
||||
"size": size,
|
||||
"genre": genre,
|
||||
"track_num": track_num,
|
||||
"cover_path": cover_path,
|
||||
})
|
||||
track_data = self._extract_ready_track(fpath, fname, handler)
|
||||
ready.append(track_data)
|
||||
cache.put(fpath, track_data)
|
||||
elif ext in SOURCE_EXTS:
|
||||
size = os.path.getsize(fpath)
|
||||
try:
|
||||
info = handler.extract_tags(fpath)
|
||||
title = info.title if info else fname
|
||||
artist = info.artist if info else "Unknown"
|
||||
track_num = info.track_number if info else 0
|
||||
cover_path = info.cover_path if info else None
|
||||
except Exception:
|
||||
title = fname
|
||||
artist = "Unknown"
|
||||
track_num = 0
|
||||
cover_path = None
|
||||
sources.append({
|
||||
"path": fpath,
|
||||
"title": title,
|
||||
"artist": artist,
|
||||
"size": size,
|
||||
"track_num": track_num,
|
||||
"cover_path": cover_path,
|
||||
})
|
||||
track_data = self._extract_source_track(fpath, fname, handler)
|
||||
sources.append(track_data)
|
||||
cache.put(fpath, track_data)
|
||||
|
||||
cache.save()
|
||||
|
||||
if cache_misses:
|
||||
self.library_status.setText(
|
||||
f"Scanned {cache_misses} file(s) — {cache_hits} from cache")
|
||||
|
||||
# Deduplicate by file path (defensive — os.walk shouldn't produce dupes)
|
||||
seen_paths = set()
|
||||
deduped_ready = []
|
||||
for track in ready:
|
||||
@ -1025,6 +1009,49 @@ class MainWindow(QMainWindow):
|
||||
self.library_scanned_sources = sources
|
||||
self._refresh_library_ui()
|
||||
|
||||
def _extract_ready_track(self, fpath: str, fname: str, handler) -> Dict[str, Any]:
|
||||
size = os.path.getsize(fpath)
|
||||
cover_path = None
|
||||
try:
|
||||
info = handler.extract_tags(fpath)
|
||||
duration = info.duration if info else 0
|
||||
artist = info.artist if info else "Unknown"
|
||||
title = info.title if info else fname
|
||||
album = info.album if info else ""
|
||||
genre = info.genre if info else ""
|
||||
track_num = info.track_number if info else 0
|
||||
cover_path = info.cover_path if info else None
|
||||
except Exception:
|
||||
duration = 0
|
||||
artist = "Unknown"
|
||||
title = fname
|
||||
album = ""
|
||||
genre = ""
|
||||
track_num = 0
|
||||
return {
|
||||
"path": fpath, "title": title, "artist": artist,
|
||||
"album": album, "duration_s": duration, "size": size,
|
||||
"genre": genre, "track_num": track_num, "cover_path": cover_path,
|
||||
}
|
||||
|
||||
def _extract_source_track(self, fpath: str, fname: str, handler) -> Dict[str, Any]:
|
||||
size = os.path.getsize(fpath)
|
||||
try:
|
||||
info = handler.extract_tags(fpath)
|
||||
title = info.title if info else fname
|
||||
artist = info.artist if info else "Unknown"
|
||||
track_num = info.track_number if info else 0
|
||||
cover_path = info.cover_path if info else None
|
||||
except Exception:
|
||||
title = fname
|
||||
artist = "Unknown"
|
||||
track_num = 0
|
||||
cover_path = None
|
||||
return {
|
||||
"path": fpath, "title": title, "artist": artist,
|
||||
"size": size, "track_num": track_num, "cover_path": cover_path,
|
||||
}
|
||||
|
||||
def _refresh_library_ui(self):
|
||||
self.library_table.setSortingEnabled(False)
|
||||
self.library_table.setRowCount(0)
|
||||
@ -1193,7 +1220,7 @@ class MainWindow(QMainWindow):
|
||||
|
||||
def _on_refresh_library_clicked(self):
|
||||
self.library_status.setText("Scanning library...")
|
||||
self._scan_library()
|
||||
self._scan_library(force=True)
|
||||
|
||||
def _on_library_add_files(self):
|
||||
source_filter = "Audio Files (*.flac *.wav *.ogg *.opus *.wma *.ape *.aiff *.alac)"
|
||||
@ -1210,6 +1237,38 @@ class MainWindow(QMainWindow):
|
||||
if new_files:
|
||||
self._scan_library()
|
||||
|
||||
def _incremental_update_after_edit(self, file_paths: list):
|
||||
"""Re-extract metadata for edited files, update cache, refresh UI
|
||||
without walking the entire library directory."""
|
||||
cache = get_library_cache()
|
||||
handler = MetadataHandler()
|
||||
|
||||
paths_set = set(file_paths)
|
||||
|
||||
for fpath in file_paths:
|
||||
ext = os.path.splitext(fpath)[1].lower()
|
||||
if ext in AUDIO_EXTS:
|
||||
data = self._extract_ready_track(fpath, os.path.basename(fpath), handler)
|
||||
cache.put(fpath, data)
|
||||
for item in self.library_ready:
|
||||
if item["path"] == fpath:
|
||||
item.update(data)
|
||||
break
|
||||
else:
|
||||
self.library_ready.append(data)
|
||||
elif ext in SOURCE_EXTS:
|
||||
data = self._extract_source_track(fpath, os.path.basename(fpath), handler)
|
||||
cache.put(fpath, data)
|
||||
for item in self.library_scanned_sources:
|
||||
if item["path"] == fpath:
|
||||
item.update(data)
|
||||
break
|
||||
else:
|
||||
self.library_scanned_sources.append(data)
|
||||
|
||||
cache.save()
|
||||
self._refresh_library_ui()
|
||||
|
||||
def _on_edit_metadata(self):
|
||||
"""Open metadata editor for selected track(s)."""
|
||||
selected_rows = set()
|
||||
@ -1236,7 +1295,7 @@ class MainWindow(QMainWindow):
|
||||
|
||||
dialog = MetadataEditorDialog(file_paths, self)
|
||||
if dialog.exec() == MetadataEditorDialog.DialogCode.Accepted:
|
||||
self._scan_library()
|
||||
self._incremental_update_after_edit(file_paths)
|
||||
|
||||
def _get_first_selected_row(self):
|
||||
for item in self.library_table.selectedItems():
|
||||
@ -1297,6 +1356,9 @@ class MainWindow(QMainWindow):
|
||||
except OSError as e:
|
||||
logger.error(f"Failed to delete {path}: {e}")
|
||||
|
||||
for f in files_to_delete:
|
||||
invalidate_cache_for_path(f["path"])
|
||||
|
||||
self._scan_library()
|
||||
QMessageBox.information(self, "Done", f"Deleted {deleted} file(s)")
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user