Pierre-Olivier Mercier
8c27498632
Replace the directory-scan queue with a real ingestion pipeline: provider -> fetcher -> cache -> ready queue, driven by a background prefetch thread. - subsonic.py: minimal OpenSubsonic client (salted-token auth, getPlaylists/getPlaylist, raw streaming download). - providers/navidrome.py: pick tracks from a playlist (by name or id), with anti-repeat and periodic playlist reload. - fetchers/subsonic.py: atomic download into the shared cache. - db.py: SQLite state — append-only play history (anti-repeat + stats) and cache_files LRU retention (keep the N most recently played). - queue.py: prefetch buffer + retention on play; graceful degradation to the stream's local-cache fallback when no source is configured. - api.py: GET /next now carries real title/artist metadata. - Config via .env (Navidrome credentials), persistent state/ volume, httpx dependency. Verified end-to-end against a live Navidrome: playlist resolved, tracks downloaded and broadcast, retention and history correct. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""SQLite state: play history (anti-repeat + stats) and cache-file retention.
|
|
|
|
Two concerns, two tables:
|
|
|
|
- ``history`` is append-only. It drives anti-repeat (recently played track
|
|
keys) and survives cache eviction, so a track can stay "recently played" even
|
|
after its file is deleted.
|
|
- ``cache_files`` tracks downloaded files so we can keep only the N most
|
|
recently *played* ones (LRU retention). Files not yet played are never
|
|
evicted.
|
|
"""
|
|
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from .models import Track
|
|
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS history (
|
|
id INTEGER PRIMARY KEY,
|
|
track_key TEXT NOT NULL,
|
|
artist TEXT,
|
|
title TEXT,
|
|
origin TEXT,
|
|
played_at REAL NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_history_played_at ON history(played_at);
|
|
|
|
CREATE TABLE IF NOT EXISTS cache_files (
|
|
path TEXT PRIMARY KEY,
|
|
track_key TEXT,
|
|
played_at REAL -- NULL until the file has been played
|
|
);
|
|
"""
|
|
|
|
|
|
class Database:
|
|
def __init__(self, path: Path):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._lock = threading.Lock()
|
|
self._conn = sqlite3.connect(
|
|
path, check_same_thread=False, isolation_level=None
|
|
)
|
|
self._conn.row_factory = sqlite3.Row
|
|
self._conn.executescript(_SCHEMA)
|
|
|
|
# --- anti-repeat / history -------------------------------------------
|
|
|
|
def recent_keys(self, limit: int) -> set[str]:
|
|
with self._lock:
|
|
rows = self._conn.execute(
|
|
"SELECT track_key FROM history ORDER BY played_at DESC LIMIT ?",
|
|
(limit,),
|
|
).fetchall()
|
|
return {r["track_key"] for r in rows}
|
|
|
|
def record_play(self, track: Track) -> None:
|
|
with self._lock:
|
|
self._conn.execute(
|
|
"INSERT INTO history (track_key, artist, title, origin, played_at)"
|
|
" VALUES (?, ?, ?, ?, ?)",
|
|
(track.key, track.artist, track.title, track.origin, time.time()),
|
|
)
|
|
|
|
# --- cache-file retention --------------------------------------------
|
|
|
|
def register_download(self, path: str, track_key: str) -> None:
|
|
with self._lock:
|
|
self._conn.execute(
|
|
"INSERT OR REPLACE INTO cache_files (path, track_key, played_at)"
|
|
" VALUES (?, ?, NULL)",
|
|
(path, track_key),
|
|
)
|
|
|
|
def mark_played(self, path: str) -> None:
|
|
with self._lock:
|
|
self._conn.execute(
|
|
"UPDATE cache_files SET played_at = ? WHERE path = ?",
|
|
(time.time(), path),
|
|
)
|
|
|
|
def evict(self, keep: int) -> list[str]:
|
|
"""Return (and forget) played files beyond the ``keep`` most recent.
|
|
|
|
The caller is responsible for deleting the returned files from disk.
|
|
Never touches files that have not been played yet.
|
|
"""
|
|
with self._lock:
|
|
rows = self._conn.execute(
|
|
"SELECT path FROM cache_files WHERE played_at IS NOT NULL"
|
|
" AND path NOT IN ("
|
|
" SELECT path FROM cache_files WHERE played_at IS NOT NULL"
|
|
" ORDER BY played_at DESC LIMIT ?"
|
|
")",
|
|
(keep,),
|
|
).fetchall()
|
|
paths = [r["path"] for r in rows]
|
|
if paths:
|
|
self._conn.executemany(
|
|
"DELETE FROM cache_files WHERE path = ?",
|
|
[(p,) for p in paths],
|
|
)
|
|
return paths
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
self._conn.close()
|