feat: migrate LibraryCache to SQLite, add SessionStore + splash overlay
- Rewrite LibraryCache from JSON (cache dir) to SQLite (data dir with WAL) - Each track gets a stable UUID so play stats survive file renames/re-encodes - Thread-safe via PRAGMA busy_timeout + RLock - Automatic backup rotation (.backup, .bak) on every write - One-shot migration from old ~/.cache/library_cache.json - Add SessionStore (~/.local/share/session.json) for playback state - Add SplashOverlay with spinning animation on first (empty-DB) launch - _scan_library() now accepts on_progress callback for the overlay - Fix _mtime → mtime SQLite column name lookup in recently-added filter - PlaylistManager path migrated to ~/.local/share/ with auto-migration - Metadata sync: remove dead consume_play_deltas(), use library API
This commit is contained in:
Maksim Totmin
parent
2d77b3ea3e
commit
ba7144c88f
50
src/app.py
50
src/app.py
@ -22,11 +22,14 @@ from config_loader import ConfigLoader
|
||||
from device_monitor import DeviceMonitor
|
||||
from hotkeys import HotkeyManager
|
||||
from playlist_manager import PlaylistManager
|
||||
from session_store import SessionStore
|
||||
from library_cache import get_library_cache
|
||||
from ui.library_tab import LibraryTab
|
||||
from ui.ipod_tab import iPodTab
|
||||
from ui.settings_tab import SettingsTab
|
||||
from ui.sidebar_widget import SIDEBAR_DEFAULT_WIDTH
|
||||
from ui.player_header import PlayerHeader
|
||||
from ui.splash_overlay import SplashOverlay
|
||||
from artwork.cache import cover_cache
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
@ -47,6 +50,7 @@ class MainWindow(QMainWindow):
|
||||
self.device_monitor: Optional[DeviceMonitor] = None
|
||||
|
||||
self.playlist_manager = PlaylistManager()
|
||||
self.session_store = SessionStore()
|
||||
self._sidebar_visible = True
|
||||
self._pending_playlist_id: str = ""
|
||||
self._pending_playlist_tracks: list = []
|
||||
@ -68,7 +72,16 @@ class MainWindow(QMainWindow):
|
||||
self._setup_hotkeys()
|
||||
self._wire_signals()
|
||||
|
||||
self.library_tab._scan_library()
|
||||
cache = get_library_cache()
|
||||
splash = SplashOverlay(self)
|
||||
if len(cache) == 0:
|
||||
splash.show()
|
||||
QApplication.processEvents()
|
||||
|
||||
self.library_tab._scan_library(on_progress=splash.set_progress)
|
||||
|
||||
if splash.isVisible():
|
||||
splash.close()
|
||||
self._resume_playback_if_saved()
|
||||
self._restore_library_state()
|
||||
self._start_device_monitor()
|
||||
@ -294,12 +307,15 @@ class MainWindow(QMainWindow):
|
||||
hide_buttons = config.get_boolean("Advanced", "hide_library_buttons", fallback=False)
|
||||
self.library_tab.set_buttons_visible(not hide_buttons)
|
||||
|
||||
self._saved_volume = config.get_int("Playback", "last_volume", fallback=80)
|
||||
self._saved_track_path = config.get("Playback", "last_track_path", fallback="")
|
||||
self._saved_position_ms = config.get_int("Playback", "last_position_ms", fallback=0)
|
||||
self._saved_playing = config.get_boolean("Playback", "last_playing", fallback=False)
|
||||
self._shuffle_active = config.get_boolean("Playback", "shuffle_active", fallback=False)
|
||||
self._repeat_mode = config.get("Playback", "repeat_mode", fallback="no_repeat")
|
||||
# Migrate Playback section → SessionStore if needed
|
||||
self.session_store.migrate_from_config(config)
|
||||
|
||||
self._saved_volume = self.session_store.get_int("last_volume", 80)
|
||||
self._saved_track_path = self.session_store.get_str("last_track_path", "")
|
||||
self._saved_position_ms = self.session_store.get_int("last_position_ms", 0)
|
||||
self._saved_playing = self.session_store.get_bool("last_playing", False)
|
||||
self._shuffle_active = self.session_store.get_bool("shuffle_active", False)
|
||||
self._repeat_mode = self.session_store.get_str("repeat_mode", "no_repeat")
|
||||
|
||||
self.player_header.set_shuffle_active(self._shuffle_active)
|
||||
self.player_header.set_repeat_mode(self._repeat_mode)
|
||||
@ -309,21 +325,25 @@ class MainWindow(QMainWindow):
|
||||
|
||||
def _save_settings(self):
|
||||
config = self.config_loader
|
||||
session = self.session_store
|
||||
self.settings_tab.save_settings(config)
|
||||
config.set("Playback", "last_volume", str(self.player_header.volume_slider.value()))
|
||||
config.set("Playback", "shuffle_active", str(self._shuffle_active).lower())
|
||||
config.set("Playback", "repeat_mode", self._repeat_mode)
|
||||
|
||||
session.set("last_volume", self.player_header.volume_slider.value())
|
||||
session.set("shuffle_active", self._shuffle_active)
|
||||
session.set("repeat_mode", self._repeat_mode)
|
||||
if self.current_track_path and os.path.exists(self.current_track_path):
|
||||
config.set("Playback", "last_track_path", self.current_track_path)
|
||||
session.set("last_track_path", self.current_track_path)
|
||||
position = self.player.position()
|
||||
duration = self.player.duration()
|
||||
if duration > 0 and position > duration - 2000:
|
||||
position = 0
|
||||
config.set("Playback", "last_position_ms", str(position))
|
||||
config.set("Playback", "last_playing", str(self._is_playing_now).lower())
|
||||
session.set("last_position_ms", position)
|
||||
session.set("last_playing", self._is_playing_now)
|
||||
else:
|
||||
config.set("Playback", "last_track_path", "")
|
||||
config.set("Playback", "last_position_ms", "0")
|
||||
session.set("last_track_path", "")
|
||||
session.set("last_position_ms", 0)
|
||||
session.save()
|
||||
|
||||
state = self.library_tab.get_state()
|
||||
config.set("Library", "view_mode", state.get("view_mode", "table"))
|
||||
config.set("Library", "filtered_album", state.get("filtered_album", ""))
|
||||
|
||||
@ -10,22 +10,11 @@ import hashlib
|
||||
import logging
|
||||
import os
|
||||
|
||||
from xdg_base import xdg_cache_path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def xdg_cache_path(subdir: str) -> str:
|
||||
"""Return path inside XDG cache directory for neo-pod-desktop.
|
||||
Creates the directory structure if missing.
|
||||
"""
|
||||
base = os.environ.get(
|
||||
"XDG_CACHE_HOME",
|
||||
os.path.join(os.path.expanduser("~"), ".cache"),
|
||||
)
|
||||
path = os.path.join(base, "neo-pod-desktop", subdir)
|
||||
os.makedirs(path, exist_ok=True)
|
||||
return path
|
||||
|
||||
|
||||
class CoverCache:
|
||||
"""Manages a local cache of cover art images for the player UI.
|
||||
|
||||
|
||||
@ -163,21 +163,12 @@ class ConfigLoader:
|
||||
if not self.config.has_option('Hotkeys', key):
|
||||
self.config.set('Hotkeys', key, value)
|
||||
|
||||
# Playback section
|
||||
# Playback section is DEPRECATED — session state moved to
|
||||
# session.json in ~/.local/share/. The defaults below are
|
||||
# kept only for backward-compatible reads during migration.
|
||||
# All new code should use SessionStore instead.
|
||||
if not self.config.has_section('Playback'):
|
||||
self.config.add_section('Playback')
|
||||
if not self.config.has_option('Playback', 'last_track_path'):
|
||||
self.config.set('Playback', 'last_track_path', '')
|
||||
if not self.config.has_option('Playback', 'last_position_ms'):
|
||||
self.config.set('Playback', 'last_position_ms', '0')
|
||||
if not self.config.has_option('Playback', 'last_volume'):
|
||||
self.config.set('Playback', 'last_volume', '80')
|
||||
if not self.config.has_option('Playback', 'last_playing'):
|
||||
self.config.set('Playback', 'last_playing', 'false')
|
||||
if not self.config.has_option('Playback', 'shuffle_active'):
|
||||
self.config.set('Playback', 'shuffle_active', 'false')
|
||||
if not self.config.has_option('Playback', 'repeat_mode'):
|
||||
self.config.set('Playback', 'repeat_mode', 'no_repeat')
|
||||
|
||||
def get(self, section: str, option: str, fallback: Any = None) -> Any:
|
||||
"""
|
||||
|
||||
@ -1,25 +1,79 @@
|
||||
#!/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.
|
||||
|
||||
Stored as SQLite-format database in ``~/.local/share/neo-pod-desktop/library_cache.db``.
|
||||
|
||||
Each row is keyed by its **file path** and also carries a stable **UUID** so
|
||||
play-count / rating / last-played data survives file renames, moves, and
|
||||
re-encodings. The database is thread-safe (WAL + busy_timeout) and
|
||||
automatically rotated to a backup file on every write.
|
||||
|
||||
Migration from the old JSON file (``~/.cache/neo-pod-desktop/library_cache.json``)
|
||||
happens automatically on the first access.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Any, Dict, Optional
|
||||
import sqlite3
|
||||
import threading
|
||||
import time
|
||||
import uuid
|
||||
from typing import Any, Iterator
|
||||
|
||||
from artwork.cache import xdg_cache_path
|
||||
from xdg_base import xdg_data_path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_CACHE_FILE = "library_cache.json"
|
||||
_DB_FILE = "library_cache.db"
|
||||
_BACKUP_FILE = "library_cache.db.backup"
|
||||
_OLD_JSON_FILE = "library_cache.json"
|
||||
|
||||
_SCHEMA_VERSION = 1
|
||||
|
||||
_SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS library_cache (
|
||||
path TEXT PRIMARY KEY,
|
||||
uuid TEXT UNIQUE NOT NULL,
|
||||
title TEXT DEFAULT '',
|
||||
artist TEXT DEFAULT '',
|
||||
album TEXT DEFAULT '',
|
||||
album_artist TEXT DEFAULT '',
|
||||
genre TEXT DEFAULT '',
|
||||
duration_s REAL DEFAULT 0,
|
||||
size INTEGER DEFAULT 0,
|
||||
track_num INTEGER DEFAULT 0,
|
||||
source_type TEXT DEFAULT '',
|
||||
bitrate INTEGER DEFAULT 0,
|
||||
sample_rate INTEGER DEFAULT 44100,
|
||||
cover_path TEXT DEFAULT '',
|
||||
thumbnail_url TEXT DEFAULT '',
|
||||
play_count INTEGER DEFAULT 0,
|
||||
rating INTEGER DEFAULT 0,
|
||||
last_played INTEGER DEFAULT 0,
|
||||
skip_count INTEGER DEFAULT 0,
|
||||
mtime REAL NOT NULL DEFAULT 0,
|
||||
content_hash TEXT NOT NULL DEFAULT '',
|
||||
updated_at INTEGER DEFAULT (cast(strftime('%s','now') as int))
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_lc_uuid ON library_cache(uuid);
|
||||
CREATE INDEX IF NOT EXISTS idx_lc_hash ON library_cache(content_hash);
|
||||
CREATE INDEX IF NOT EXISTS idx_lc_artist ON library_cache(artist);
|
||||
CREATE INDEX IF NOT EXISTS idx_lc_album ON library_cache(album);
|
||||
|
||||
PRAGMA user_version = {version};
|
||||
"""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fast content fingerprint
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _content_hash(filepath: str, max_bytes: int = 65536) -> str:
|
||||
"""SHA1 of first max_bytes of audio data (fast content fingerprint)."""
|
||||
"""SHA1 of first *max_bytes* of audio data."""
|
||||
h = hashlib.sha1()
|
||||
try:
|
||||
with open(filepath, "rb") as f:
|
||||
@ -29,96 +83,19 @@ def _content_hash(filepath: str, max_bytes: int = 65536) -> str:
|
||||
return h.hexdigest()
|
||||
|
||||
|
||||
class LibraryCache:
|
||||
"""JSON-based cache of audio file metadata.
|
||||
# ---------------------------------------------------------------------------
|
||||
# Singleton helpers (kept for backward compatibility)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Keys are absolute file paths. Values are dicts of metadata
|
||||
plus internal ``_mtime`` and ``_content_hash`` fields for
|
||||
change detection and track matching.
|
||||
|
||||
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
|
||||
result = {}
|
||||
for k, v in entry.items():
|
||||
if k == "_mtime":
|
||||
continue
|
||||
if k == "_content_hash":
|
||||
result["content_hash"] = v
|
||||
else:
|
||||
result[k] = v
|
||||
return result
|
||||
|
||||
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
|
||||
entry["_content_hash"] = _content_hash(file_path)
|
||||
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: "LibraryCache | None" = None
|
||||
_migrated_flag = False
|
||||
|
||||
|
||||
_library_cache: Optional[LibraryCache] = None
|
||||
|
||||
|
||||
def get_library_cache() -> LibraryCache:
|
||||
def get_library_cache() -> "LibraryCache":
|
||||
global _library_cache
|
||||
if _library_cache is None:
|
||||
_library_cache = LibraryCache()
|
||||
_library_cache.load()
|
||||
_library_cache._migrate_old()
|
||||
return _library_cache
|
||||
|
||||
|
||||
@ -131,3 +108,347 @@ def save_library_cache() -> None:
|
||||
def invalidate_cache_for_path(file_path: str) -> None:
|
||||
cache = get_library_cache()
|
||||
cache.remove(file_path)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SQLite-backed cache
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class LibraryCache:
|
||||
"""Persistent cache backed by a single SQLite database.
|
||||
|
||||
Public API (unchanged from the JSON version):
|
||||
get(path) → dict | None
|
||||
put(path, data) → None
|
||||
remove(path) → None
|
||||
clear() → None
|
||||
save() → None
|
||||
__len__() → int
|
||||
|
||||
New public API:
|
||||
iter_entries() → Iterator[(path, dict)]
|
||||
get_by_content_hash(chash) → dict | None
|
||||
get_by_uuid(uid) → dict | None
|
||||
update_play_count(path) → int (increment + return)
|
||||
"""
|
||||
|
||||
def __init__(self, db_path: str | None = None):
|
||||
if db_path is None:
|
||||
db_path = os.path.join(xdg_data_path(), _DB_FILE)
|
||||
self._path = db_path
|
||||
self._lock = threading.RLock()
|
||||
|
||||
self.conn = sqlite3.connect(
|
||||
db_path, check_same_thread=False,
|
||||
detect_types=sqlite3.PARSE_DECLTYPES,
|
||||
)
|
||||
self.conn.row_factory = sqlite3.Row
|
||||
self.conn.execute("PRAGMA journal_mode=WAL")
|
||||
self.conn.execute("PRAGMA busy_timeout=5000")
|
||||
self.conn.execute("PRAGMA synchronous=NORMAL")
|
||||
self.conn.execute("PRAGMA foreign_keys=OFF")
|
||||
self._init_db()
|
||||
|
||||
def _init_db(self) -> None:
|
||||
self.conn.executescript(
|
||||
_SCHEMA.format(version=_SCHEMA_VERSION)
|
||||
)
|
||||
self.conn.commit()
|
||||
|
||||
# ── public read API ────────────────────────────────────────────────
|
||||
|
||||
def get(self, file_path: str) -> dict[str, Any] | None:
|
||||
"""Return cached data for *file_path*, or None if missing or stale."""
|
||||
with self._lock:
|
||||
row = self.conn.execute(
|
||||
"SELECT * FROM library_cache WHERE path=?",
|
||||
(file_path,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return None
|
||||
if not os.path.isfile(file_path):
|
||||
return None
|
||||
actual_mtime = os.path.getmtime(file_path)
|
||||
if actual_mtime != row["mtime"]:
|
||||
return None
|
||||
return self._row_to_dict(row)
|
||||
|
||||
def get_by_content_hash(self, content_hash: str) -> dict[str, Any] | None:
|
||||
"""Look up a track by its content fingerprint."""
|
||||
with self._lock:
|
||||
row = self.conn.execute(
|
||||
"SELECT * FROM library_cache WHERE content_hash=?",
|
||||
(content_hash,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return None
|
||||
return self._row_to_dict(row)
|
||||
|
||||
def get_by_uuid(self, track_uuid: str) -> dict[str, Any] | None:
|
||||
"""Look up a track by its stable UUID."""
|
||||
with self._lock:
|
||||
row = self.conn.execute(
|
||||
"SELECT * FROM library_cache WHERE uuid=?",
|
||||
(track_uuid,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
return None
|
||||
return self._row_to_dict(row)
|
||||
|
||||
def iter_entries(self) -> Iterator[tuple[str, dict[str, Any]]]:
|
||||
"""Yield (file_path, data_dict) for every row in the database."""
|
||||
with self._lock:
|
||||
rows = self.conn.execute(
|
||||
"SELECT * FROM library_cache"
|
||||
).fetchall()
|
||||
for row in rows:
|
||||
yield (row["path"], self._row_to_dict(row))
|
||||
|
||||
def __len__(self) -> int:
|
||||
with self._lock:
|
||||
return self.conn.execute(
|
||||
"SELECT COUNT(*) FROM library_cache"
|
||||
).fetchone()[0]
|
||||
|
||||
# ── public write API ───────────────────────────────────────────────
|
||||
|
||||
def put(self, file_path: str, data: dict[str, Any]) -> None:
|
||||
"""Insert or replace a cache entry for *file_path*."""
|
||||
try:
|
||||
mtime = os.path.getmtime(file_path)
|
||||
except OSError:
|
||||
return
|
||||
|
||||
track_uuid = data.get("_track_uuid") or data.get("uuid") or str(uuid.uuid4())
|
||||
ch = data.get("_content_hash") or data.get("content_hash") or _content_hash(file_path)
|
||||
|
||||
with self._lock:
|
||||
self.conn.execute(
|
||||
"""INSERT OR REPLACE INTO library_cache
|
||||
(path, uuid, title, artist, album, album_artist,
|
||||
genre, duration_s, size, track_num, source_type,
|
||||
bitrate, sample_rate, cover_path, thumbnail_url,
|
||||
play_count, rating, last_played, skip_count,
|
||||
mtime, content_hash, updated_at)
|
||||
VALUES (?,?,?,?,?,?,
|
||||
?,?,?,?,?,
|
||||
?,?,?,?,
|
||||
?,?,?,?,
|
||||
?,?,cast(strftime('%s','now') as int))""",
|
||||
(
|
||||
file_path,
|
||||
track_uuid,
|
||||
data.get("title", ""),
|
||||
data.get("artist", ""),
|
||||
data.get("album", ""),
|
||||
data.get("album_artist", ""),
|
||||
data.get("genre", ""),
|
||||
data.get("duration_s", 0) or data.get("duration", 0),
|
||||
data.get("size", 0),
|
||||
data.get("track_num", 0) or data.get("track_number", 0),
|
||||
data.get("source_type", ""),
|
||||
data.get("bitrate", 0),
|
||||
data.get("sample_rate", 44100),
|
||||
data.get("cover_path", ""),
|
||||
data.get("thumbnail_url", ""),
|
||||
data.get("play_count", 0),
|
||||
data.get("rating", 0),
|
||||
data.get("last_played", 0),
|
||||
data.get("skip_count", 0),
|
||||
mtime,
|
||||
ch,
|
||||
),
|
||||
)
|
||||
self._backup()
|
||||
|
||||
def update_play_stats(self, file_path: str,
|
||||
play_count: int | None = None,
|
||||
rating: int | None = None,
|
||||
last_played: int | None = None,
|
||||
skip_count: int | None = None) -> None:
|
||||
"""Update only the play-stat fields for *file_path*."""
|
||||
fields = []
|
||||
vals: list[Any] = []
|
||||
for k, v in [("play_count", play_count), ("rating", rating),
|
||||
("last_played", last_played), ("skip_count", skip_count)]:
|
||||
if v is not None:
|
||||
fields.append(f"{k}=?")
|
||||
vals.append(v)
|
||||
if not fields:
|
||||
return
|
||||
fields.append("updated_at=cast(strftime('%s','now') as int)")
|
||||
with self._lock:
|
||||
self.conn.execute(
|
||||
f"UPDATE library_cache SET {', '.join(fields)} WHERE path=?",
|
||||
(*vals, file_path),
|
||||
)
|
||||
self._backup()
|
||||
|
||||
def remove(self, file_path: str) -> None:
|
||||
with self._lock:
|
||||
self.conn.execute(
|
||||
"DELETE FROM library_cache WHERE path=?", (file_path,),
|
||||
)
|
||||
self._backup()
|
||||
|
||||
def clear(self) -> None:
|
||||
with self._lock:
|
||||
self.conn.execute("DELETE FROM library_cache")
|
||||
self._backup()
|
||||
|
||||
def save(self) -> None:
|
||||
"""Explicit commit + backup."""
|
||||
with self._lock:
|
||||
self.conn.commit()
|
||||
self._backup()
|
||||
|
||||
# ── backup rotation ────────────────────────────────────────────────
|
||||
|
||||
def _backup(self) -> None:
|
||||
"""Rotate database backup files after each write.
|
||||
|
||||
Commits the current transaction first so the backup connection
|
||||
can checkpoint the WAL without blocking.
|
||||
|
||||
Keeps two historical copies:
|
||||
library_cache.db ← current
|
||||
library_cache.db.backup ← previous version
|
||||
library_cache.db.bak ← version before that
|
||||
"""
|
||||
# Commit so the second connection can safely checkpoint.
|
||||
self.conn.commit()
|
||||
|
||||
db = self._path
|
||||
b1 = db + ".backup"
|
||||
b2 = db + ".bak"
|
||||
|
||||
try:
|
||||
bak_conn = sqlite3.connect(db)
|
||||
bak_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
||||
bak_conn.close()
|
||||
except Exception as exc:
|
||||
logger.debug("Checkpoint failed: %s", exc)
|
||||
|
||||
if os.path.exists(b1):
|
||||
try:
|
||||
os.replace(b1, b2)
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
import shutil
|
||||
if os.path.exists(db):
|
||||
shutil.copy2(db, b1)
|
||||
except OSError as exc:
|
||||
logger.debug("Backup rotation failed: %s", exc)
|
||||
|
||||
# ── internal helpers ───────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _row_to_dict(row: sqlite3.Row) -> dict[str, Any]:
|
||||
keys = row.keys()
|
||||
d = dict(zip(keys, row))
|
||||
d.pop("updated_at", None)
|
||||
alias = {
|
||||
"duration_s": "duration",
|
||||
"track_num": "track_number",
|
||||
}
|
||||
for old_k, new_k in alias.items():
|
||||
if old_k in d and new_k not in d:
|
||||
d[new_k] = d[old_k]
|
||||
return d
|
||||
|
||||
# ── JSON migration legacy→SQLite ──────────────────────────────────
|
||||
|
||||
def _migrate_old(self) -> None:
|
||||
"""One-shot migration from the old JSON cache file."""
|
||||
global _migrated_flag
|
||||
if _migrated_flag:
|
||||
return
|
||||
|
||||
old_dir = xdg_data_path()
|
||||
old_path = os.path.join(
|
||||
os.path.dirname(old_dir), # parent: neo-pod-desktop
|
||||
".cache", "neo-pod-desktop", _OLD_JSON_FILE,
|
||||
)
|
||||
# also try the actual XDG cache dir
|
||||
from xdg_base import xdg_cache_path
|
||||
alt_path = os.path.join(xdg_cache_path(), _OLD_JSON_FILE)
|
||||
|
||||
for candidate in (old_path, alt_path):
|
||||
if os.path.exists(candidate):
|
||||
self._do_migrate_json(candidate)
|
||||
break
|
||||
|
||||
_migrated_flag = True
|
||||
|
||||
def _do_migrate_json(self, json_path: str) -> None:
|
||||
"""Read old JSON and insert all entries."""
|
||||
try:
|
||||
with open(json_path, "r", encoding="utf-8") as f:
|
||||
old_data = json.load(f)
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning("JSON migration source corrupt: %s", exc)
|
||||
return
|
||||
|
||||
count = 0
|
||||
for file_path, entry in old_data.items():
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
mtime = entry.get("_mtime", 0)
|
||||
if not mtime:
|
||||
try:
|
||||
mtime = os.path.getmtime(file_path)
|
||||
except OSError:
|
||||
continue
|
||||
entry["_track_uuid"] = entry.get("_track_uuid") or str(uuid.uuid4())
|
||||
entry["_content_hash"] = entry.get("_content_hash") or _content_hash(file_path)
|
||||
with self._lock:
|
||||
self.conn.execute(
|
||||
"""INSERT OR IGNORE INTO library_cache
|
||||
(path, uuid, title, artist, album, album_artist,
|
||||
genre, duration_s, size, track_num, source_type,
|
||||
bitrate, sample_rate, cover_path, thumbnail_url,
|
||||
play_count, rating, last_played, skip_count,
|
||||
mtime, content_hash, updated_at)
|
||||
VALUES (?,?,?,?,?,?,
|
||||
?,?,?,?,?,
|
||||
?,?,?,?,
|
||||
?,?,?,?,
|
||||
?,?,cast(strftime('%s','now') as int))""",
|
||||
(
|
||||
file_path,
|
||||
entry.get("_track_uuid", ""),
|
||||
entry.get("title", ""),
|
||||
entry.get("artist", ""),
|
||||
entry.get("album", ""),
|
||||
entry.get("album_artist", ""),
|
||||
entry.get("genre", ""),
|
||||
entry.get("duration_s", 0) or entry.get("duration", 0),
|
||||
entry.get("size", 0),
|
||||
entry.get("track_num", 0) or entry.get("track_number", 0),
|
||||
entry.get("source_type", ""),
|
||||
entry.get("bitrate", 0),
|
||||
entry.get("sample_rate", 44100),
|
||||
entry.get("cover_path", ""),
|
||||
entry.get("thumbnail_url", ""),
|
||||
entry.get("play_count", 0),
|
||||
entry.get("rating", 0),
|
||||
entry.get("last_played", 0),
|
||||
entry.get("skip_count", 0),
|
||||
mtime,
|
||||
entry.get("_content_hash", ""),
|
||||
),
|
||||
)
|
||||
count += 1
|
||||
self.conn.commit()
|
||||
self._backup()
|
||||
|
||||
# Rename old file so we don't re-migrate
|
||||
try:
|
||||
os.rename(json_path, json_path + ".migrated")
|
||||
logger.info(
|
||||
"Migrated %d entries from %s → SQLite",
|
||||
count, json_path,
|
||||
)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
@ -5,11 +5,11 @@ import time
|
||||
import logging
|
||||
from typing import List, Dict, Optional
|
||||
|
||||
from xdg_base import xdg_data_path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_PLAYLIST_FILE = os.path.join(
|
||||
os.path.expanduser("~"), ".config", "neo-pod-desktop", "playlists.json"
|
||||
)
|
||||
_PLAYLIST_FILE = os.path.join(xdg_data_path(), "playlists.json")
|
||||
|
||||
|
||||
class Playlist:
|
||||
@ -47,8 +47,22 @@ class PlaylistManager:
|
||||
def __init__(self, filepath: str = _PLAYLIST_FILE):
|
||||
self._filepath = filepath
|
||||
self._playlists: List[Playlist] = []
|
||||
self._migrate_old()
|
||||
self._load()
|
||||
|
||||
def _migrate_old(self):
|
||||
"""Migrate from old ~/.config/ location to ~/.local/share/."""
|
||||
old = os.path.join(
|
||||
os.path.expanduser("~"), ".config", "neo-pod-desktop", "playlists.json",
|
||||
)
|
||||
if os.path.exists(old) and not os.path.exists(self._filepath):
|
||||
try:
|
||||
os.makedirs(os.path.dirname(self._filepath), exist_ok=True)
|
||||
os.rename(old, self._filepath)
|
||||
logger.info("Migrated playlists from %s → %s", old, self._filepath)
|
||||
except OSError as exc:
|
||||
logger.warning("Playlist migration failed: %s", exc)
|
||||
|
||||
def _load(self):
|
||||
try:
|
||||
os.makedirs(os.path.dirname(self._filepath), exist_ok=True)
|
||||
|
||||
88
src/session_store.py
Normal file
88
src/session_store.py
Normal file
@ -0,0 +1,88 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Session state store — persistent playback state.
|
||||
|
||||
Stored as JSON in ``~/.local/share/neo-pod-desktop/session.json``.
|
||||
Replaces the ``[Playback]`` section of config.ini — session state
|
||||
(user's last track, position, volume, playback mode) is user data,
|
||||
not configuration.
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
from xdg_base import xdg_data_path
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_SESSION_FILE = "session.json"
|
||||
|
||||
|
||||
class SessionStore:
|
||||
def __init__(self, path: str | None = None):
|
||||
self._path = path or os.path.join(xdg_data_path(), _SESSION_FILE)
|
||||
self._data: dict[str, Any] = {}
|
||||
self._load()
|
||||
|
||||
# ── load / save ────────────────────────────────────────────────────
|
||||
|
||||
def _load(self) -> None:
|
||||
if not os.path.exists(self._path):
|
||||
return
|
||||
try:
|
||||
with open(self._path, "r", encoding="utf-8") as f:
|
||||
self._data = json.load(f)
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning("Failed to load session: %s", exc)
|
||||
self._data = {}
|
||||
|
||||
def save(self) -> None:
|
||||
tmp = self._path + ".tmp"
|
||||
try:
|
||||
os.makedirs(os.path.dirname(self._path), exist_ok=True)
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
json.dump(self._data, f, ensure_ascii=False)
|
||||
os.replace(tmp, self._path)
|
||||
except OSError as exc:
|
||||
logger.warning("Failed to save session: %s", exc)
|
||||
|
||||
# ── typed accessors ────────────────────────────────────────────────
|
||||
|
||||
def get_str(self, key: str, fallback: str = "") -> str:
|
||||
return str(self._data.get(key, fallback))
|
||||
|
||||
def get_int(self, key: str, fallback: int = 0) -> int:
|
||||
return int(self._data.get(key, fallback))
|
||||
|
||||
def get_bool(self, key: str, fallback: bool = False) -> bool:
|
||||
v = self._data.get(key, fallback)
|
||||
if isinstance(v, bool):
|
||||
return v
|
||||
return str(v).lower() in ("true", "1", "yes")
|
||||
|
||||
def set(self, key: str, value: Any) -> None:
|
||||
self._data[key] = value
|
||||
|
||||
# ── migrations ─────────────────────────────────────────────────────
|
||||
|
||||
def migrate_from_config(self, config_loader: Any) -> None:
|
||||
"""Import ``[Playback]`` section from the old config loader."""
|
||||
keys = [
|
||||
"last_track_path", "last_position_ms", "last_volume",
|
||||
"last_playing", "shuffle_active", "repeat_mode",
|
||||
]
|
||||
changed = False
|
||||
for k in keys:
|
||||
if k not in self._data:
|
||||
v = config_loader.get("Playback", k, fallback="")
|
||||
self.set(k, v)
|
||||
changed = True
|
||||
if changed:
|
||||
self.save()
|
||||
# Wipe old Playback section so it isn't read again.
|
||||
if config_loader.config.has_section("Playback"):
|
||||
config_loader.config.remove_section("Playback")
|
||||
config_loader.save()
|
||||
logger.info("Migrated playback state from config.ini to session.json")
|
||||
@ -9,7 +9,7 @@ import sys
|
||||
import time
|
||||
import logging
|
||||
import shutil
|
||||
from typing import List, Dict, Optional, Tuple, Any
|
||||
from typing import List, Dict, Optional, Tuple, Any, Callable
|
||||
from pathlib import Path
|
||||
|
||||
from PyQt6.QtWidgets import (
|
||||
@ -450,7 +450,8 @@ class LibraryTab(QWidget):
|
||||
return f"{size_bytes / 1024:.0f} KB"
|
||||
return f"{size_bytes / (1024 * 1024):.1f} MB"
|
||||
|
||||
def _scan_library(self, force: bool = False):
|
||||
def _scan_library(self, force: bool = False,
|
||||
on_progress: Callable[[int, int], None] | None = None):
|
||||
output_dir = self.get_output_dir()
|
||||
if not output_dir or not os.path.isdir(output_dir):
|
||||
self.library_ready.clear()
|
||||
@ -458,10 +459,23 @@ class LibraryTab(QWidget):
|
||||
self._refresh_library_ui()
|
||||
return
|
||||
|
||||
# Pre-count total scannable files so the overlay can show progress.
|
||||
total = 0
|
||||
if on_progress:
|
||||
for root, _dirs, files in os.walk(output_dir):
|
||||
for fname in files:
|
||||
fpath = os.path.join(root, fname)
|
||||
ext = os.path.splitext(fname)[1].lower()
|
||||
if os.path.isfile(fpath) and (ext in AUDIO_EXTS or ext in SOURCE_EXTS):
|
||||
total += 1
|
||||
if total == 0:
|
||||
total = 1
|
||||
|
||||
ready = []
|
||||
sources = []
|
||||
cache_hits = 0
|
||||
cache_misses = 0
|
||||
processed = 0
|
||||
|
||||
cache = get_library_cache()
|
||||
handler = MetadataHandler()
|
||||
@ -473,6 +487,7 @@ class LibraryTab(QWidget):
|
||||
if not os.path.isfile(fpath):
|
||||
continue
|
||||
|
||||
processed += 1
|
||||
cached = None if force else cache.get(fpath)
|
||||
|
||||
if cached is not None:
|
||||
@ -483,10 +498,8 @@ class LibraryTab(QWidget):
|
||||
elif ext in SOURCE_EXTS:
|
||||
cached.pop("_mtime", None)
|
||||
sources.append(cached)
|
||||
continue
|
||||
|
||||
else:
|
||||
cache_misses += 1
|
||||
|
||||
if ext in AUDIO_EXTS:
|
||||
track_data = self._extract_ready_track(fpath, fname, handler)
|
||||
ready.append(track_data)
|
||||
@ -496,6 +509,9 @@ class LibraryTab(QWidget):
|
||||
sources.append(track_data)
|
||||
cache.put(fpath, track_data)
|
||||
|
||||
if on_progress:
|
||||
on_progress(processed, total)
|
||||
|
||||
cache.save()
|
||||
|
||||
if cache_misses:
|
||||
@ -1384,7 +1400,7 @@ class LibraryTab(QWidget):
|
||||
entry = cache.get(path) if path else None
|
||||
visible = False
|
||||
if entry and isinstance(entry, dict):
|
||||
added = entry.get("_mtime", 0) or entry.get("added_at", 0) or 0
|
||||
added = entry.get("mtime", 0) or entry.get("added_at", 0) or 0
|
||||
visible = added >= cutoff
|
||||
self.library_table.setRowHidden(row, not visible)
|
||||
|
||||
|
||||
118
src/ui/splash_overlay.py
Normal file
118
src/ui/splash_overlay.py
Normal file
@ -0,0 +1,118 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Splash overlay shown while scanning the music library for the first time."""
|
||||
|
||||
from PyQt6.QtWidgets import QDialog, QVBoxLayout, QLabel, QWidget, QApplication
|
||||
from PyQt6.QtCore import Qt, QTimer
|
||||
from PyQt6.QtGui import QPainter, QPen, QColor, QFont
|
||||
|
||||
|
||||
class _Spinner(QWidget):
|
||||
"""A spinning arc — like a circular progress indicator."""
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setFixedSize(64, 64)
|
||||
self._angle = 0.0
|
||||
self._timer = QTimer(self)
|
||||
self._timer.timeout.connect(self._tick)
|
||||
self._timer.start(30)
|
||||
|
||||
def _tick(self):
|
||||
self._angle = (self._angle + 6) % 360
|
||||
self.update()
|
||||
|
||||
def paintEvent(self, event):
|
||||
p = QPainter(self)
|
||||
p.setRenderHint(QPainter.RenderHint.Antialiasing)
|
||||
r = self.rect().adjusted(4, 4, -4, -4)
|
||||
|
||||
p.setPen(QPen(QColor("#e0e0e0"), 3, Qt.PenStyle.SolidLine, Qt.PenCapStyle.RoundCap))
|
||||
p.drawEllipse(r)
|
||||
|
||||
p.setPen(QPen(QColor("#3b82f6"), 4, Qt.PenStyle.SolidLine, Qt.PenCapStyle.RoundCap))
|
||||
p.drawArc(r, int(self._angle * 16), 60 * 16)
|
||||
|
||||
|
||||
class SplashOverlay(QDialog):
|
||||
"""Modal overlay with a spinning animation + progress counter.
|
||||
|
||||
Usage::
|
||||
|
||||
splash = SplashOverlay(window)
|
||||
splash.show()
|
||||
# … scan …
|
||||
splash.set_progress(current, total)
|
||||
splash.close()
|
||||
"""
|
||||
|
||||
def __init__(self, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setWindowFlags(
|
||||
Qt.WindowType.FramelessWindowHint | Qt.WindowType.Dialog
|
||||
)
|
||||
self.setAttribute(Qt.WidgetAttribute.WA_TranslucentBackground)
|
||||
self.setAttribute(Qt.WidgetAttribute.WA_ShowWithoutActivating)
|
||||
self.setModal(True)
|
||||
|
||||
self._card = QWidget(self)
|
||||
self._card.setObjectName("card")
|
||||
self._card.setFixedSize(340, 240)
|
||||
|
||||
layout = QVBoxLayout(self._card)
|
||||
layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
layout.setSpacing(16)
|
||||
|
||||
self._spinner = _Spinner()
|
||||
layout.addWidget(self._spinner, alignment=Qt.AlignmentFlag.AlignCenter)
|
||||
|
||||
self._title = QLabel("Создаём музыкальную\nмедиатеку…")
|
||||
self._title.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
f = self._title.font()
|
||||
f.setPointSize(16)
|
||||
f.setBold(True)
|
||||
self._title.setFont(f)
|
||||
layout.addWidget(self._title)
|
||||
|
||||
self._count = QLabel("")
|
||||
self._count.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
f2 = self._count.font()
|
||||
f2.setPointSize(12)
|
||||
self._count.setFont(f2)
|
||||
layout.addWidget(self._count)
|
||||
|
||||
self._card.setStyleSheet("""
|
||||
#card {
|
||||
background: white;
|
||||
border-radius: 16px;
|
||||
}
|
||||
QLabel {
|
||||
color: #333;
|
||||
background: transparent;
|
||||
}
|
||||
""")
|
||||
|
||||
def _recenter(self):
|
||||
if not self.parent():
|
||||
return
|
||||
pr = self.parent().rect()
|
||||
self.setGeometry(pr)
|
||||
cx = (pr.width() - self._card.width()) // 2
|
||||
cy = (pr.height() - self._card.height()) // 2
|
||||
self._card.move(cx, cy)
|
||||
|
||||
def showEvent(self, event):
|
||||
self._recenter()
|
||||
super().showEvent(event)
|
||||
QApplication.processEvents()
|
||||
|
||||
def resizeEvent(self, event):
|
||||
self._recenter()
|
||||
super().resizeEvent(event)
|
||||
|
||||
def set_progress(self, current: int, total: int) -> None:
|
||||
"""Update the file counter shown on the overlay."""
|
||||
if total > 0:
|
||||
self._count.setText(f"Файл {current} из {total}")
|
||||
else:
|
||||
self._count.setText(f"Файл {current}")
|
||||
QApplication.processEvents()
|
||||
@ -205,9 +205,12 @@ class WorkerThread(QThread):
|
||||
|
||||
self.progress_signal.emit(60, "Syncing play counts from iPod...")
|
||||
|
||||
from library_cache import get_library_cache
|
||||
from library_cache import get_library_cache, save_library_cache
|
||||
cache = get_library_cache()
|
||||
|
||||
# Read deltas from iPod Play Counts file without deleting it —
|
||||
# the file is cleaned up by _cleanup_play_counts_file() only after a
|
||||
# successful DB write in _sync_itunescdb_impl / sync_metadata_for.
|
||||
ipod_base, ipod_delta = db.read_play_stats()
|
||||
|
||||
hash_to_location: dict[str, str] = {}
|
||||
@ -223,37 +226,41 @@ class WorkerThread(QThread):
|
||||
hash_to_location[ch] = loc
|
||||
|
||||
updated_cache = 0
|
||||
for file_path, entry in list(cache._data.items()):
|
||||
for file_path, entry in cache.iter_entries():
|
||||
if not os.path.exists(file_path):
|
||||
continue
|
||||
ch = entry.get("_content_hash", "")
|
||||
ch = entry.get("content_hash", "")
|
||||
if not ch or ch not in hash_to_location:
|
||||
continue
|
||||
loc = hash_to_location[ch]
|
||||
bs = ipod_base.get(loc, {})
|
||||
dd = ipod_delta.get(loc, {})
|
||||
|
||||
entry["play_count"] = max(
|
||||
new_play_count = max(
|
||||
entry.get("play_count", 0), bs.get("play_count", 0),
|
||||
) + dd.get("play_count", 0)
|
||||
|
||||
entry["skip_count"] = max(
|
||||
new_skip_count = max(
|
||||
entry.get("skip_count", 0), bs.get("skip_count", 0),
|
||||
) + dd.get("skip_count", 0)
|
||||
|
||||
dd_rating = dd.get("rating", 0)
|
||||
entry["rating"] = dd_rating if dd_rating >= 0 else bs.get("rating", 0) or entry.get("rating", 0)
|
||||
|
||||
entry["last_played"] = max(
|
||||
new_rating = dd_rating if dd_rating >= 0 else bs.get("rating", 0) or entry.get("rating", 0)
|
||||
new_last_played = max(
|
||||
entry.get("last_played", 0), bs.get("last_played", 0),
|
||||
)
|
||||
|
||||
cache.update_play_stats(
|
||||
file_path,
|
||||
play_count=new_play_count,
|
||||
rating=new_rating,
|
||||
last_played=new_last_played,
|
||||
skip_count=new_skip_count,
|
||||
)
|
||||
updated_cache += 1
|
||||
|
||||
if updated_cache:
|
||||
cache.save()
|
||||
save_library_cache()
|
||||
logger.info("Updated play counts for %d local tracks from iPod", updated_cache)
|
||||
|
||||
db.consume_play_deltas()
|
||||
self.progress_signal.emit(80, "Loading complete")
|
||||
|
||||
result = {
|
||||
|
||||
54
src/xdg_base.py
Normal file
54
src/xdg_base.py
Normal file
@ -0,0 +1,54 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
XDG Base Directory helpers.
|
||||
|
||||
Follows the XDG Base Directory Specification:
|
||||
~/.config/ — configuration files
|
||||
~/.local/share/ — persistent user data
|
||||
~/.cache/ — temporary cache data
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
_DATA_DIR: str | None = None
|
||||
_CONFIG_DIR: str | None = None
|
||||
_CACHE_DIR: str | None = None
|
||||
|
||||
|
||||
def xdg_config_path(subdir: str = "") -> str:
|
||||
global _CONFIG_DIR
|
||||
if _CONFIG_DIR is None:
|
||||
base = os.environ.get(
|
||||
"XDG_CONFIG_HOME",
|
||||
os.path.join(os.path.expanduser("~"), ".config"),
|
||||
)
|
||||
_CONFIG_DIR = os.path.join(base, "neo-pod-desktop")
|
||||
path = os.path.join(_CONFIG_DIR, subdir) if subdir else _CONFIG_DIR
|
||||
os.makedirs(path, exist_ok=True)
|
||||
return path
|
||||
|
||||
|
||||
def xdg_data_path(subdir: str = "") -> str:
|
||||
global _DATA_DIR
|
||||
if _DATA_DIR is None:
|
||||
base = os.environ.get(
|
||||
"XDG_DATA_HOME",
|
||||
os.path.join(os.path.expanduser("~"), ".local", "share"),
|
||||
)
|
||||
_DATA_DIR = os.path.join(base, "neo-pod-desktop")
|
||||
path = os.path.join(_DATA_DIR, subdir) if subdir else _DATA_DIR
|
||||
os.makedirs(path, exist_ok=True)
|
||||
return path
|
||||
|
||||
|
||||
def xdg_cache_path(subdir: str = "") -> str:
|
||||
global _CACHE_DIR
|
||||
if _CACHE_DIR is None:
|
||||
base = os.environ.get(
|
||||
"XDG_CACHE_HOME",
|
||||
os.path.join(os.path.expanduser("~"), ".cache"),
|
||||
)
|
||||
_CACHE_DIR = os.path.join(base, "neo-pod-desktop")
|
||||
path = os.path.join(_CACHE_DIR, subdir) if subdir else _CACHE_DIR
|
||||
os.makedirs(path, exist_ok=True)
|
||||
return path
|
||||
Loading…
x
Reference in New Issue
Block a user