"""FeedBack — FastAPI backend serving highway viewer + library."""
import asyncio
import bisect
import hashlib
import json
import logging
import math
import os
import secrets
import sys
import tempfile
import shutil
from pathlib import Path
from typing import Any, ClassVar
from logging_setup import configure_logging
from env_compat import getenv_compat
configure_logging()
log = logging.getLogger("feedBack.server")
from fastapi import Body, FastAPI, WebSocket, WebSocketDisconnect, UploadFile, File, HTTPException, Query
from fastapi.concurrency import run_in_threadpool
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response, StreamingResponse
from safepath import safe_join
from song import (
anchor_to_wire,
arrangement_string_count,
base_open_string_midis,
compute_smart_names,
chord_template_to_wire,
chord_to_wire,
hand_shape_to_wire,
key_to_tonic_pc,
load_song,
note_to_wire,
phrase_to_wire,
pitch_from_base,
scale_degree_for_pitch,
)
from audio import find_wem_files, convert_wem
from tunings import tuning_name, DEFAULT_TUNINGS, DEFAULT_REFERENCE_PITCH, apply_reference_pitch
import sloppak as sloppak_mod
import drums as drums_mod
import notation as notation_mod
import loosefolder as loosefolder_mod
# Metadata extraction lives in a side-effect-free module so ProcessPool
# scan workers can import + unpickle _scan_one without re-running this
# module's import-time side effects (see lib/scan_worker.py).
from scan_worker import _extract_meta_for_file, _relpath, _scan_one
import concurrent.futures
import contextvars
import inspect
import ipaddress
import multiprocessing
import re
import sqlite3
import threading
import time
import uuid
import warnings
import xml.etree.ElementTree as ET
import structlog
from fastapi import Request
app = FastAPI(title="FeedBack")
# Plugins that maintain session stores can register a cleanup callback here.
# The demo-mode janitor calls every registered hook once per hour so stale
# sessions are swept without the core needing to know plugin internals.
_DEMO_JANITOR_HOOKS: list = []
_DEMO_JANITOR_HOOKS_LOCK = threading.Lock()
_DEMO_JANITOR_STARTED = False
_DEMO_JANITOR_STOP = threading.Event()
_DEMO_JANITOR_THREAD: threading.Thread | None = None
def register_demo_janitor_hook(fn) -> None:
"""Register a zero-argument callable to be invoked hourly by the demo
janitor. Plugins call this from their ``setup(app, context)`` when they
want to participate in session cleanup under demo mode.
The callable must accept no required arguments. Async (coroutine)
functions are rejected: the janitor runs in a plain thread and cannot
await coroutines.
"""
if not callable(fn):
raise TypeError(
f"register_demo_janitor_hook expects a callable, got {type(fn).__name__!r}"
)
# Reject coroutine functions — check both the callable itself and its
# __call__ method so objects with an async __call__ (e.g. class instances,
# functools.partial wrappers around async functions) are also caught.
_call = getattr(fn, "__call__", None)
if inspect.iscoroutinefunction(fn) or (
_call is not None and inspect.iscoroutinefunction(_call)
):
raise TypeError(
"register_demo_janitor_hook does not accept async functions; "
"the janitor runs in a plain thread and cannot await coroutines"
)
# Validate that the callable accepts zero required arguments so it won't
# crash at sweep time (hourly, far from the registration site).
try:
sig = inspect.signature(fn)
except ValueError:
# inspect.signature() raises ValueError for built-in C callables whose
# signature cannot be determined. Accept them as-is; if they fail at
# runtime the janitor will catch and log the exception.
pass
else:
required = [
p for p in sig.parameters.values()
if p.default is inspect.Parameter.empty
and p.kind not in (
inspect.Parameter.VAR_POSITIONAL,
inspect.Parameter.VAR_KEYWORD,
)
]
if required:
raise TypeError(
f"register_demo_janitor_hook expects a zero-argument callable; "
f"{fn!r} has {len(required)} required parameter(s): "
+ ", ".join(p.name for p in required)
)
with _DEMO_JANITOR_HOOKS_LOCK:
_DEMO_JANITOR_HOOKS.append(fn)
def _run_janitor_hook(hook) -> None:
"""Run a single janitor hook inline, swallowing and logging any exception.
If the hook returns an awaitable (e.g. a coroutine slipped through the
async-function guard), the coroutine is closed immediately to avoid
``RuntimeWarning: coroutine was never awaited`` noise, and a warning is
emitted so the plugin author knows to fix their hook.
"""
try:
result = hook()
except Exception:
log.exception("janitor hook %r raised", hook)
return
if inspect.iscoroutine(result):
# A coroutine slipped through the async-function guard (e.g. via a
# wrapper/partial). Close it to suppress "coroutine never awaited",
# then warn so the plugin author knows to fix their hook.
try:
result.close()
except Exception:
log.exception("error closing coroutine from janitor hook %r", hook)
warnings.warn(
f"janitor hook {hook!r} returned a coroutine; "
"hooks must be plain synchronous callables — "
"register_demo_janitor_hook does not accept async functions",
RuntimeWarning,
stacklevel=1,
)
elif inspect.isawaitable(result):
# Future/Task: no .close() method; just warn and leave it alone.
warnings.warn(
f"janitor hook {hook!r} returned an awaitable (Future/Task); "
"hooks must be plain synchronous callables",
RuntimeWarning,
stacklevel=1,
)
_DEMO_BLOCKED: list[tuple[str, re.Pattern]] = [
("POST", re.compile(r"^/api/settings$")),
("POST", re.compile(r"^/api/settings/import$")),
("POST", re.compile(r"^/api/settings/reset$")),
("POST", re.compile(r"^/api/rescan$")),
("POST", re.compile(r"^/api/rescan/full$")),
("POST", re.compile(r"^/api/songs/upload$")),
("DELETE", re.compile(r"^/api/song/.+$")),
("POST", re.compile(r"^/api/favorites/toggle$")),
("POST", re.compile(r"^/api/loops$")),
("DELETE", re.compile(r"^/api/loops/[^/]+$")),
("POST", re.compile(r"^/api/audio-effects/mappings$")),
("DELETE", re.compile(r"^/api/audio-effects/mappings/[^/]+$")),
("POST", re.compile(r"^/api/audio-effects/mappings/[^/]+/activate$")),
("DELETE", re.compile(r"^/api/audio-effects/active-mapping$")),
("POST", re.compile(r"^/api/song/.*/meta$")),
("POST", re.compile(r"^/api/song/.*/art/upload$")),
("GET", re.compile(r"^/api/plugins/updates$")),
("POST", re.compile(r"^/api/plugins/[^/]+/update$")),
("POST", re.compile(r"^/api/plugins/editor/save$")),
("POST", re.compile(r"^/api/plugins/editor/build$")),
("POST", re.compile(r"^/api/plugins/editor/upload-art$")),
("POST", re.compile(r"^/api/plugins/editor/upload-audio$")),
("POST", re.compile(r"^/api/plugins/editor/youtube-audio$")),
("POST", re.compile(r"^/api/plugins/editor/import-gp$")),
("POST", re.compile(r"^/api/plugins/editor/import-midi$")),
("POST", re.compile(r"^/api/plugins/lyrics_karaoke/align$")),
("POST", re.compile(r"^/api/plugins/lyrics_karaoke/generate-pitch$")),
("POST", re.compile(r"^/api/plugins/lyrics_karaoke/save-lyrics$")),
("POST", re.compile(r"^/api/plugins/lyrics_sync/align$")),
("POST", re.compile(r"^/api/plugins/lyrics_sync/save$")),
("POST", re.compile(r"^/api/plugins/studio/sessions/[^/]+/extract-drums$")),
("POST", re.compile(r"^/api/diagnostics/export$")),
("GET", re.compile(r"^/api/diagnostics/preview$")),
("GET", re.compile(r"^/api/diagnostics/hardware$")),
# Bundled core plugin — video background upload/delete
("POST", re.compile(r"^/api/plugins/highway_3d/files$")),
("DELETE", re.compile(r"^/api/plugins/highway_3d/files$")),
# fee[dB]ack v0.3.0 write endpoints — demo mode is read-only, so block the
# new profile / XP / stats / playlists / saved mutators too.
("POST", re.compile(r"^/api/profile$")),
("POST", re.compile(r"^/api/profile/avatar$")),
("POST", re.compile(r"^/api/xp/award$")),
("POST", re.compile(r"^/api/stats$")),
("POST", re.compile(r"^/api/playlists$")),
("PATCH", re.compile(r"^/api/playlists/[^/]+$")),
("DELETE", re.compile(r"^/api/playlists/[^/]+$")),
("POST", re.compile(r"^/api/playlists/[^/]+/songs$")),
("DELETE", re.compile(r"^/api/playlists/[^/]+/songs/.+$")),
("POST", re.compile(r"^/api/playlists/[^/]+/reorder$")),
("POST", re.compile(r"^/api/playlists/[^/]+/cover$")),
("DELETE", re.compile(r"^/api/playlists/[^/]+/cover$")),
("POST", re.compile(r"^/api/saved/toggle$")),
# Progression (spec 010) write endpoints — demo mode stays read-only.
("POST", re.compile(r"^/api/progression/paths$")),
("POST", re.compile(r"^/api/progression/onboarding$")),
("POST", re.compile(r"^/api/progression/events$")),
("POST", re.compile(r"^/api/shop/buy$")),
("POST", re.compile(r"^/api/shop/equip$")),
]
@app.middleware("http")
async def _demo_mode_guard(request: Request, call_next):
if getenv_compat("FEEDBACK_DEMO_MODE") or getenv_compat("FEEDBACK_DEMO_MODE") == "1":
path = request.url.path
for method, pattern in _DEMO_BLOCKED:
if request.method == method and pattern.match(path):
return JSONResponse({"error": "demo mode: read-only"}, status_code=403)
response = await call_next(request)
if request.method == "GET" and path == "/" and "feedBack_demo_session" not in request.cookies:
forwarded_proto = (request.headers.get("x-forwarded-proto") or "").split(",")[0].strip()
is_secure = request.url.scheme == "https" or forwarded_proto.lower() == "https"
response.set_cookie(
"feedBack_demo_session", str(uuid.uuid4()),
max_age=86400, httponly=True, samesite="lax",
secure=is_secure,
)
return response
return await call_next(request)
from asgi_correlation_id import CorrelationIdMiddleware
# validator=None accepts any non-empty inbound X-Request-ID value, including
# opaque proxy-generated hex strings, not just RFC-4122 UUIDs.
app.add_middleware(CorrelationIdMiddleware, validator=None)
STATIC_DIR = Path(__file__).parent / "static"
try:
STATIC_DIR.mkdir(exist_ok=True)
except OSError:
pass # Read-only in packaged installs
# Distinguish "env not set / empty" from "explicitly set". Path("") collapses
# to Path(".") so we can't recover that signal after the cast — capture the
# raw env-var string up front and let _get_dlc_dir() consult both. This way
# `DLC_DIR=.` remains a valid opt-in for cwd while `DLC_DIR=""` (or unset)
# falls through to the config.json fallback.
_DLC_DIR_ENV = os.environ.get("DLC_DIR", "").strip()
DLC_DIR = Path(_DLC_DIR_ENV) if _DLC_DIR_ENV else Path("")
CONFIG_DIR = Path(os.environ.get("CONFIG_DIR", str(Path.home() / ".local" / "share" / "feedback")))
# Writable cache directories (use CONFIG_DIR, not STATIC_DIR which may be read-only)
ART_CACHE_DIR = CONFIG_DIR / "art_cache"
AUDIO_CACHE_DIR = CONFIG_DIR / "audio_cache"
SLOPPAK_CACHE_DIR = CONFIG_DIR / "sloppak_cache"
def _env_flag(name: str) -> bool:
"""Parse a conventional boolean env flag (honours legacy SLOPSMITH_* alias)."""
return (getenv_compat(name, "") or "").strip().lower() in {"1", "true", "yes", "on"}
# Canonical Tuning-filter grouping key (feedBack#867). tuning_name collapses
# every non-standard tuning to "Custom Tuning"; for those rows we key on the
# raw offsets so distinct customs stay distinct, while named tunings keep
# grouping by name (stable across the offsets-column migration). Used by both
# the tuning-names listing and the filter WHERE so the contract matches.
_TUNING_GROUP_KEY_SQL = (
"CASE WHEN tuning_name = 'Custom Tuning' AND COALESCE(tuning_offsets, '') != '' "
"THEN tuning_offsets ELSE tuning_name END"
)
# ── SQLite metadata cache ─────────────────────────────────────────────────────
def _ensure_smart_names(arrangements: list[dict]) -> list[dict]:
"""Fill in missing ``smart_name`` fields and sort arrangements by smart order.
Applied to every library query result so the client always receives
arrangements in priority order:
Lead → Alt. Lead [1,2,…] → Bonus Lead [1,2,…]
→ Rhythm → Alt. Rhythm → Bonus Rhythm
→ Bass → Alt. Bass → Bonus Bass → other
Rows scanned before the smart-naming feature was introduced don't carry a
``smart_name`` key. The background scanner automatically rescans those rows
to populate the field from authoritative manifest JSON path flags.
In the meantime this function provides a best-effort on-the-fly computation.
However, when multiple arrangements share the same name (e.g. two "Combo"
tracks in a archive that bundles all path flags as zero), name-based inference
cannot distinguish Lead from Rhythm — so we emit ``smart_name: null`` and
let the UI fall back to the legacy name until the background rescan corrects
the row. Arrangements that already have the field are never modified.
"""
if not arrangements:
return arrangements
# Fill in missing smart_name values.
if not all("smart_name" in a for a in arrangements):
# Detect duplicate raw names across ALL arrangements (not just the
# missing subset). A duplicate anywhere means the name-based fallback
# may assign the same smart type a scanned row already owns — emit
# None for the missing entries and let the legacy name show through
# until the background rescan corrects them.
# Coerce to str so a malformed cached row with a list/dict name
# doesn't blow up the set() conversion (and every query that hits it).
all_names = [
a.get("name", "") if isinstance(a.get("name"), str) else str(a.get("name", ""))
for a in arrangements
]
has_duplicates = len(all_names) != len(set(all_names))
if has_duplicates:
for a in arrangements:
if "smart_name" not in a:
a["smart_name"] = None
else:
# No duplicates — name-based fallback is safe.
from song import Arrangement as _ArrCls
arr_objs = [
_ArrCls(
name=a.get("name", ""),
path_lead=a.get("_path_lead", False),
path_rhythm=a.get("_path_rhythm", False),
path_bass=a.get("_path_bass", False),
bonus_arr=a.get("_bonus_arr", False),
represent=a.get("_represent", 0),
)
for a in arrangements
]
smart = compute_smart_names(arr_objs)
for a, sn in zip(arrangements, smart):
if "smart_name" not in a:
a["smart_name"] = sn
# Always sort by smart priority order so the client receives a consistent
# list regardless of how the DB row was originally stored.
# _arr_smart_sort_key is defined later in this module but resolved at
# call-time, so the forward reference is safe.
arrangements.sort(key=_arr_smart_sort_key)
return arrangements
def _sqlite_file_integrity_ok(path: Path) -> bool:
"""True if `path` is a SQLite database that opens and passes
`PRAGMA quick_check`. Used to gate a DB restore so a truncated or
corrupt snapshot can never overwrite the live library DB."""
try:
with open(path, "rb") as f:
if f.read(16) != b"SQLite format 3\x00": # cheap header gate, no full read
return False
except OSError:
return False
conn = None
try:
conn = sqlite3.connect(str(path))
row = conn.execute("PRAGMA quick_check").fetchone()
return bool(row) and row[0] == "ok"
except sqlite3.Error:
return False
finally:
if conn is not None:
conn.close()
# quick_check on a non-WAL file makes no sidecars, but a malformed
# file can; sweep them so a probe never litters config_dir.
for suffix in ("-wal", "-shm"):
try:
path.with_name(path.name + suffix).unlink()
except FileNotFoundError:
pass
def _apply_pending_db_restore(config_dir: Path) -> None:
"""Swap in a library DB restored from a settings bundle, if one is
staged. A settings import writes the restored snapshot to
`web_library.db.restore` rather than over the live DB (the running
server holds the old file open, and a stale `-wal`/`-shm` could be
replayed onto a fresh main file → corruption). The swap happens here,
at startup, BEFORE the connection opens: delete the old DB and its WAL
sidecars, then rename the staged snapshot into place. The snapshot is a
fully-checkpointed single file (SQLite online-backup API), so it needs
no sidecars of its own. Idempotent and a no-op when nothing is staged.
The staged file is re-validated here before anything is destroyed: a
restore that fails its integrity check is discarded and the live DB is
left untouched, so a bad bundle can never brick startup or lose data."""
pending = config_dir / "web_library.db.restore"
if not pending.exists():
return
if not _sqlite_file_integrity_ok(pending):
log.error("pending library DB restore failed its integrity check; "
"discarding it and keeping the existing database")
try:
pending.unlink()
except FileNotFoundError:
pass
return
for suffix in ("", "-wal", "-shm"):
try:
(config_dir / f"web_library.db{suffix}").unlink()
except FileNotFoundError:
pass
os.replace(pending, config_dir / "web_library.db")
log.info("applied pending library DB restore from settings import")
# ── Keyset (cursor) pagination for the library grid (feedBack#636 item 3) ─────
# Forward-only, O(page) deep paging that doesn't grow with OFFSET. Only simple
# single-column sorts can keyset cleanly (the compound tuning/year sorts fall
# back to OFFSET). Every sort gets a unique `filename` tiebreak so the order is
# TOTAL — which also fixes a latent OFFSET skip/dupe across equal-key rows.
# (column, collate-clause, primary-direction) — tiebreak is always `filename` ASC.
_KEYSET_SORTS = {
"artist": ("artist", "COLLATE NOCASE", "ASC"),
"artist-desc": ("artist", "COLLATE NOCASE", "DESC"),
"title": ("title", "COLLATE NOCASE", "ASC"),
"title-desc": ("title", "COLLATE NOCASE", "DESC"),
"recent": ("mtime", "", "DESC"),
}
# Index into a query_page row tuple for each keyset column (see the SELECT in
# query_page: filename, title, artist, ... mtime at 9).
_KEYSET_ROW_IDX = {"artist": 2, "title": 1, "mtime": 9}
def _encode_cursor(values: list) -> str:
import base64
return base64.urlsafe_b64encode(json.dumps(values).encode("utf-8")).decode("ascii")
def _decode_cursor(cursor: str):
"""Decode an opaque keyset cursor to [sort_value, filename], or None if it's
malformed (a bad cursor degrades to the first page, never 500s)."""
import base64
try:
out = json.loads(base64.urlsafe_b64decode(cursor.encode("ascii")).decode("utf-8"))
except (ValueError, TypeError):
return None
return out if isinstance(out, list) and len(out) == 2 else None
def _effective_keyset_sort(sort: str, direction: str) -> str:
"""Fold the legacy `dir=desc` toggle into the canonical keyset sort key, so
the seek/cursor direction matches the ORDER BY that same toggle produces
(without this, `sort=artist&dir=desc` would seek with `>` against a DESC
order → gaps/dupes)."""
if direction == "desc" and sort in ("artist", "title"):
return sort + "-desc"
return sort
def _keyset_seek(col: str, collate: str, primary_dir: str, cv, fn: str):
"""(sql, params) for 'rows strictly after (cv, fn)' in the total order
`
, filename ASC`, matching SQLite's NULL placement
(NULLs sort first in ASC, last in DESC) so keyset is exactly OFFSET-
equivalent even for NULL sort keys."""
ce = f"{col} {collate}".strip()
if primary_dir == "ASC": # NULLs first
if cv is None:
return (f"(({col} IS NULL AND filename > ?) OR {col} IS NOT NULL)", [fn])
return (f"({col} IS NOT NULL AND ({ce} > ? OR ({ce} = ? AND filename > ?)))",
[cv, cv, fn])
# DESC — NULLs last
if cv is None:
return (f"({col} IS NULL AND filename > ?)", [fn])
return (f"({col} IS NULL OR ({col} IS NOT NULL AND "
f"({ce} < ? OR ({ce} = ? AND filename > ?))))", [cv, cv, fn])
def next_library_cursor(sort: str, last_song: dict | None) -> str | None:
"""The cursor for the last row of a page, so the next request resumes after
it. None when the sort can't keyset or the page was empty."""
if sort not in _KEYSET_SORTS or not last_song:
return None
col = _KEYSET_SORTS[sort][0]
key = "mtime" if col == "mtime" else col
if key not in last_song or "filename" not in last_song:
return None
return _encode_cursor([last_song[key], last_song["filename"]])
class MetadataDB:
def __init__(self):
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
_apply_pending_db_restore(CONFIG_DIR)
self.db_path = str(CONFIG_DIR / "web_library.db")
self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
self.conn.execute("PRAGMA journal_mode=WAL")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS songs (
filename TEXT PRIMARY KEY,
mtime REAL,
size INTEGER,
title TEXT,
artist TEXT,
album TEXT,
year TEXT,
duration REAL,
tuning TEXT,
arrangements TEXT,
has_lyrics INTEGER DEFAULT 0,
format TEXT DEFAULT 'archive',
stem_count INTEGER DEFAULT 0,
stem_ids TEXT DEFAULT '[]',
tuning_name TEXT DEFAULT '',
tuning_sort_key INTEGER DEFAULT 0,
tuning_offsets TEXT DEFAULT ''
)
""")
# Idempotent migrations for installs that predate each column.
for ddl in (
"ALTER TABLE songs ADD COLUMN format TEXT DEFAULT 'archive'",
"ALTER TABLE songs ADD COLUMN stem_count INTEGER DEFAULT 0",
# feedBack#129: per-stem filter needs the id list, not just count.
"ALTER TABLE songs ADD COLUMN stem_ids TEXT DEFAULT '[]'",
# feedBack#69 + #22: denormalized canonical tuning name + numeric
# sort key (sum of offsets). The existing `tuning` text column
# stays — these are caches, repopulated on rescan.
"ALTER TABLE songs ADD COLUMN tuning_name TEXT DEFAULT ''",
"ALTER TABLE songs ADD COLUMN tuning_sort_key INTEGER DEFAULT 0",
# feedBack#867: raw per-string offsets (space-joined ints) so the
# v3 client can render target notes and the Tuning filter can keep
# distinct custom tunings distinct (tuning_name collapses them all
# to "Custom Tuning"). Cache; repopulated on rescan.
"ALTER TABLE songs ADD COLUMN tuning_offsets TEXT DEFAULT ''",
):
try:
self.conn.execute(ddl)
except sqlite3.OperationalError:
pass
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_songs_artist ON songs(artist COLLATE NOCASE)")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_songs_title ON songs(title COLLATE NOCASE)")
# Composite (sort col, filename) indexes cover the grid's ORDER BY +
# its unique filename tiebreak — for both the OFFSET scan and keyset
# seek (feedBack#636 item 3). idx_songs_artist/title above stay for the
# distinct-artist / letter-bar aggregates.
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_songs_artist_fn ON songs(artist COLLATE NOCASE, filename)")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_songs_title_fn ON songs(title COLLATE NOCASE, filename)")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_songs_mtime_fn ON songs(mtime, filename)")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_songs_tuning_name ON songs(tuning_name COLLATE NOCASE)")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_songs_tuning_sort_key ON songs(tuning_sort_key)")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_songs_year ON songs(year)")
self.conn.execute("CREATE TABLE IF NOT EXISTS favorites (filename TEXT PRIMARY KEY)")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS loops (
id INTEGER PRIMARY KEY AUTOINCREMENT,
filename TEXT NOT NULL,
name TEXT NOT NULL,
start_time REAL NOT NULL,
end_time REAL NOT NULL,
created_at TEXT DEFAULT (datetime('now'))
)
""")
# fee[dB]ack v0.3.0 — single-user player profile (id=1), streak, and the
# unified XP store. Peers of favorites/loops; additive + idempotent.
# `player_hash` is a future-leaderboard identity label (SHA-256 of the
# first display name + a once-generated salt), never an auth credential.
self.conn.execute("""
CREATE TABLE IF NOT EXISTS profile (
id INTEGER PRIMARY KEY CHECK (id = 1),
display_name TEXT,
avatar_path TEXT,
player_hash TEXT,
player_salt TEXT,
onboarded INTEGER NOT NULL DEFAULT 0,
created_at TEXT
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS profile_progress (
id INTEGER PRIMARY KEY CHECK (id = 1),
current_streak INTEGER NOT NULL DEFAULT 0,
best_streak INTEGER NOT NULL DEFAULT 0,
last_active_date TEXT -- YYYY-MM-DD (local)
)
""")
# Unified XP store: the single source of truth the profile badge reads.
# Song-play, minigames, and tutorials all feed THIS via award_xp() — no
# second XP curve (lib/xp.py owns the math).
self.conn.execute("""
CREATE TABLE IF NOT EXISTS xp_profile (
id INTEGER PRIMARY KEY CHECK (id = 1),
xp INTEGER NOT NULL DEFAULT 0,
total_awards INTEGER NOT NULL DEFAULT 0,
minigames_seeded INTEGER NOT NULL DEFAULT 0,
updated_at TEXT
)
""")
# Per-source XP ledger: the unified `xp` total above is a single number,
# but a source (minigames, tutorials, song-play, …) needs to know its own
# contribution so it can be reset/reversed independently (a minigames
# profile-reset must subtract only its share, not song-play XP).
self.conn.execute("""
CREATE TABLE IF NOT EXISTS xp_sources (
source TEXT PRIMARY KEY,
xp INTEGER NOT NULL DEFAULT 0
)
""")
# Per-song/arrangement practice stats (best score + accuracy, plays,
# last position for Continue-Playing). Fed by the highway note-detection
# scorer via POST /api/stats. Additive + idempotent; a 0.2.9 build
# tolerates it and the new build opens an old db without it.
self.conn.execute("""
CREATE TABLE IF NOT EXISTS song_stats (
filename TEXT NOT NULL,
arrangement INTEGER NOT NULL DEFAULT 0,
plays INTEGER NOT NULL DEFAULT 0,
best_score INTEGER NOT NULL DEFAULT 0,
best_accuracy REAL NOT NULL DEFAULT 0,
last_score INTEGER NOT NULL DEFAULT 0,
last_accuracy REAL NOT NULL DEFAULT 0,
last_position REAL NOT NULL DEFAULT 0,
last_played_at TEXT,
updated_at TEXT,
PRIMARY KEY (filename, arrangement)
)
""")
self.conn.execute("CREATE INDEX IF NOT EXISTS idx_song_stats_recent ON song_stats(last_played_at DESC)")
# Playlists + the reserved "Saved for Later" system playlist. Additive.
self.conn.execute("""
CREATE TABLE IF NOT EXISTS playlists (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
system_key TEXT, -- 'saved_for_later' for reserved playlists, else NULL
created_at TEXT,
updated_at TEXT
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS playlist_songs (
playlist_id INTEGER NOT NULL,
filename TEXT NOT NULL,
position INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (playlist_id, filename)
)
""")
self.conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_playlists_system_key ON playlists(system_key) WHERE system_key IS NOT NULL")
# Smart collections (feedBack#636 item 2): a playlist row whose `rules`
# JSON is non-NULL is a smart/dynamic collection — its membership is the
# LIVE result of those library filter params, not a stored song list.
# It surfaces as a registered library provider (the v3 source picker),
# so it inherits the whole Songs UI. Additive, idempotent migration.
try:
self.conn.execute("ALTER TABLE playlists ADD COLUMN rules TEXT")
except sqlite3.OperationalError:
pass
# Wishlist / "wanted" (feedBack#636 item 4): a persisted, actionable
# list of songs the user does NOT own yet — the *arr "Wanted/Monitored"
# analogue. Unlike playlists (which reference owned local songs by
# filename), a wanted entry has no local file, so it lives in its own
# table keyed by descriptive identity. Producers (the find_more plugin's
# ownership-diff, or a manual add) POST here; the consuming UI reads it.
# Additive + idempotent.
self.conn.execute("""
CREATE TABLE IF NOT EXISTS wanted (
id INTEGER PRIMARY KEY AUTOINCREMENT,
artist TEXT NOT NULL DEFAULT '',
title TEXT NOT NULL DEFAULT '',
source TEXT NOT NULL DEFAULT '', -- e.g. 'find_more', 'manual'
source_ref TEXT NOT NULL DEFAULT '', -- opaque id/url within that source
note TEXT NOT NULL DEFAULT '',
created_at TEXT
)
""")
# Identity = (artist, title, source, source_ref), case-insensitive on
# the human fields, so re-running an ownership-diff doesn't duplicate.
self.conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_wanted_identity "
"ON wanted(artist COLLATE NOCASE, title COLLATE NOCASE, source, source_ref)"
)
# Progression (spec 010): instrument paths, challenges, quests, the
# Decibels wallet, and the cosmetics shop. Targets/titles live in the
# bundled content (data/progression/); these tables hold only player
# state (counters, completion timestamps, spend, ownership) so content
# edits update live displays without migrations. Additive + idempotent.
self.conn.execute("""
CREATE TABLE IF NOT EXISTS progression_state (
id INTEGER PRIMARY KEY CHECK (id = 1),
calibration_status TEXT NOT NULL DEFAULT 'pending', -- pending|completed|skipped
calibration_completed_at TEXT,
created_at TEXT
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS player_paths (
path_id TEXT PRIMARY KEY, -- 'guitar' | 'bass' | 'drums' | future
level INTEGER NOT NULL DEFAULT 0,
selected_at TEXT
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS challenge_progress (
challenge_id TEXT PRIMARY KEY, -- namespaced 'guitar.l1.clean-run'
path_id TEXT NOT NULL,
level INTEGER NOT NULL, -- the level whose set this belongs to
count INTEGER NOT NULL DEFAULT 0,
progress_detail TEXT, -- JSON, e.g. {"seen": [...]} for distinct goals
completed_at TEXT
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS quest_state (
period_type TEXT NOT NULL, -- 'daily' | 'weekly'
period_key TEXT NOT NULL, -- '2026-06-12' | '2026-W24'
quest_id TEXT NOT NULL,
count INTEGER NOT NULL DEFAULT 0,
reward_db INTEGER NOT NULL DEFAULT 0, -- snapshot at instantiation
progress_detail TEXT,
completed_at TEXT,
PRIMARY KEY (period_type, period_key, quest_id)
)
""")
# Spend is tracked separately from xp_profile.xp on purpose: the xp
# total stays the monotonic lifetime-earned stat (db_earned goals,
# xp_sources reset semantics) and balance = MAX(0, xp - spent).
self.conn.execute("""
CREATE TABLE IF NOT EXISTS wallet (
id INTEGER PRIMARY KEY CHECK (id = 1),
spent INTEGER NOT NULL DEFAULT 0
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS shop_owned (
item_id TEXT PRIMARY KEY,
cost_paid INTEGER NOT NULL DEFAULT 0,
acquired_at TEXT
)
""")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS shop_equipped (
slot TEXT PRIMARY KEY, -- 'theme' | 'avatar_frame'
item_id TEXT
)
""")
# Ensure the singleton rows exist so reads never special-case "no row".
self.conn.execute("INSERT OR IGNORE INTO profile (id, onboarded, created_at) VALUES (1, 0, datetime('now'))")
self.conn.execute("INSERT OR IGNORE INTO profile_progress (id) VALUES (1)")
self.conn.execute("INSERT OR IGNORE INTO xp_profile (id, xp, total_awards, updated_at) VALUES (1, 0, 0, datetime('now'))")
self.conn.execute("INSERT OR IGNORE INTO progression_state (id, created_at) VALUES (1, datetime('now'))")
self.conn.execute("INSERT OR IGNORE INTO wallet (id) VALUES (1)")
self.conn.commit()
self._lock = threading.Lock()
# One-time repair of pre-fix rows written under URL-encoded filenames
# (idempotent: a no-op once every row is canonical).
self._migrate_decode_stat_filenames()
def _song_exists(self, filename: str) -> bool:
return self.conn.execute(
"SELECT 1 FROM songs WHERE filename = ?", (filename,)).fetchone() is not None
def _canonical_song_filename(self, filename: str) -> str:
"""Map a (possibly URL-encoded) filename to the `songs` library key.
The recorder relays encodeURIComponent'd names ('/'→'%2F', ' '→'%20'),
but `songs` keys on the decoded on-disk path. Decoding is LIBRARY-AWARE so
a real filename that legitimately contains literal %XX is never corrupted:
prefer the form that already exists in `songs`, and decode only when the
decoded form resolves to a real song. When NEITHER form is in the library
(e.g. a play recorded before the library scan finishes) keep the stored
name unchanged — the next-startup migration canonicalizes it once the song
is scanned, rather than risk corrupting a real %XX name now."""
if not isinstance(filename, str):
return filename
if self._song_exists(filename):
return filename # already a real library key (may contain %)
from urllib.parse import unquote
decoded = unquote(filename)
if decoded != filename and self._song_exists(decoded):
return decoded # encoded → real library key
return filename # neither in library: leave as-is (heals on migrate)
def _migrate_decode_stat_filenames(self):
"""Rewrite URL-encoded song_stats.filename rows to the decoded
library-path key (the form `songs` uses). Pre-fix, the recorder stored
encodeURIComponent'd names, so every recorded best was invisible to the
reads that filter on `filename IN (SELECT filename FROM songs)`. Merge on
collision — two encoded rows decoding to the same name, or an encoded row
meeting an already-decoded one — with the same best=max / plays=sum /
last-wins semantics as song_score.merge_stats, so the (filename,
arrangement) primary key is never violated.
Library-aware via the shared _canonical_song_filename rule: only decode a
row when the decoded form is a real song, so a correctly-stored name
containing literal %XX is never rewritten, and dead-song/orphan rows
(neither form in the library) are left exactly as-is."""
cols = self._STATS_COLS
with self._lock:
rows = [dict(zip(cols, r)) for r in self.conn.execute(
"SELECT " + ", ".join(cols) + " FROM song_stats").fetchall()]
canon = self._canonical_song_filename
if all(canon(r["filename"]) == r["filename"] for r in rows):
return # every row already canonical (or an untouchable orphan)
merged: dict = {}
for r in rows:
key = (canon(r["filename"]), int(r["arrangement"]))
cur = merged.get(key)
if cur is None:
merged[key] = dict(r, filename=key[0], arrangement=key[1])
continue
# Most-recently-updated row wins the "last_*"/position fields.
def _stamp(x):
return str(x.get("updated_at") or x.get("last_played_at") or "")
newer = r if _stamp(r) >= _stamp(cur) else cur
merged[key] = {
"filename": key[0], "arrangement": key[1],
"plays": (cur["plays"] or 0) + (r["plays"] or 0),
"best_score": max(cur["best_score"] or 0, r["best_score"] or 0),
"best_accuracy": max(cur["best_accuracy"] or 0.0, r["best_accuracy"] or 0.0),
"last_score": newer["last_score"], "last_accuracy": newer["last_accuracy"],
"last_position": newer["last_position"],
"last_played_at": newer["last_played_at"], "updated_at": newer["updated_at"],
}
# Atomic swap: clear and reinsert the canonicalized set in one txn.
try:
self.conn.execute("DELETE FROM song_stats")
self.conn.executemany(
"INSERT INTO song_stats (" + ", ".join(cols) + ") VALUES ("
+ ", ".join("?" * len(cols)) + ")",
[tuple(m[c] for c in cols) for m in merged.values()],
)
self.conn.commit()
except Exception:
self.conn.rollback()
raise
def is_favorite(self, filename: str) -> bool:
return self.conn.execute("SELECT 1 FROM favorites WHERE filename = ?", (filename,)).fetchone() is not None
def toggle_favorite(self, filename: str) -> bool:
"""Toggle favorite status. Returns new state."""
with self._lock:
if self.is_favorite(filename):
self.conn.execute("DELETE FROM favorites WHERE filename = ?", (filename,))
self.conn.commit()
return False
else:
self.conn.execute("INSERT OR IGNORE INTO favorites VALUES (?)", (filename,))
self.conn.commit()
return True
# ── Player profile (fee[dB]ack v0.3.0) ─────────────────────────────────
def get_profile(self) -> dict:
row = self.conn.execute(
"SELECT display_name, avatar_path, player_hash, onboarded FROM profile WHERE id = 1"
).fetchone()
if not row:
return {"display_name": None, "avatar_url": None, "player_hash": None, "onboarded": False}
return {
"display_name": row[0],
"avatar_url": row[1],
"player_hash": row[2],
"onboarded": bool(row[3]),
}
def set_profile(self, display_name: str, avatar_url: str | None) -> dict:
"""Set/update the display name (+ avatar). Computes player_hash ONCE
from the first name + a stored random salt; it stays stable across
later name changes. Marks onboarded=1."""
with self._lock:
cur = self.conn.execute(
"SELECT player_hash, player_salt FROM profile WHERE id = 1"
).fetchone()
player_hash = cur[0] if cur else None
salt = cur[1] if cur else None
if not player_hash:
salt = secrets.token_hex(16)
player_hash = hashlib.sha256((display_name + salt).encode("utf-8")).hexdigest()
self.conn.execute(
"UPDATE profile SET display_name = ?, "
"avatar_path = COALESCE(?, avatar_path), "
"player_hash = ?, player_salt = ?, onboarded = 1 WHERE id = 1",
(display_name, avatar_url, player_hash, salt),
)
self.conn.commit()
return self.get_profile()
# ── Unified XP store ────────────────────────────────────────────────────
def get_xp(self) -> int:
row = self.conn.execute("SELECT xp FROM xp_profile WHERE id = 1").fetchone()
return int(row[0]) if row else 0
def award_xp(self, amount: int, source: str | None = None) -> int:
"""Add XP to the unified store; returns the new total. `amount` may be
NEGATIVE — used internally to REVERSE a failed award (the total and the
per-source bucket both clamp at 0). `source` (when given) is tracked in
the xp_sources ledger so it can be reset independently.
Service boundary: the plugin hook (context["award_xp"]) passes this
straight through, so coerce defensively — bad input (bool, NaN/Inf,
non-integral, out-of-int64-range) must neither raise NOR mutate state.
_as_int rejects bool/non-integral; bad → no-op (0)."""
try:
amount = _as_int(amount)
except (TypeError, ValueError, OverflowError):
amount = 0
amount = max(-10_000_000, min(amount, 10_000_000))
with self._lock:
# MAX(0, …) clamps the result so a reversal can't drive XP negative.
self.conn.execute(
"UPDATE xp_profile SET xp = MAX(0, xp + ?), "
"total_awards = total_awards + ?, updated_at = datetime('now') WHERE id = 1",
(amount, 1 if amount > 0 else 0),
)
if source:
self.conn.execute(
"INSERT INTO xp_sources (source, xp) VALUES (?, MAX(0, ?)) "
"ON CONFLICT(source) DO UPDATE SET xp = MAX(0, xp + ?)",
(source, amount, amount),
)
self.conn.commit()
row = self.conn.execute("SELECT xp FROM xp_profile WHERE id = 1").fetchone()
return int(row[0]) if row else 0
def reset_source_xp(self, source: str) -> dict:
"""Subtract a single source's tracked contribution from the unified
total and zero its bucket (e.g. a minigames profile-reset removes only
minigames XP, leaving song-play/tutorials XP intact). Returns progress."""
with self._lock:
row = self.conn.execute("SELECT xp FROM xp_sources WHERE source = ?", (source,)).fetchone()
amt = int(row[0]) if row and row[0] else 0
if amt:
self.conn.execute(
"UPDATE xp_profile SET xp = MAX(0, xp - ?), updated_at = datetime('now') WHERE id = 1",
(amt,),
)
self.conn.execute("UPDATE xp_sources SET xp = 0 WHERE source = ?", (source,))
self.conn.commit()
return self.get_progress()
def seed_xp_once(self, amount: int, marker: str = "minigames") -> bool:
"""One-time seed of the unified store from a pre-unification source
(e.g. the minigames plugin's profile.json), so existing earned XP is
preserved. No-ops if already seeded or the store already has XP.
Returns True if it seeded."""
# Same no-raise / no-silent-mutate contract as award_xp(): this is a
# plugin-facing service (context["seed_xp"]). _as_int rejects bool /
# non-integral; bad input becomes a 0 (no-op) seed rather than raising.
try:
amount = _as_int(amount)
except (TypeError, ValueError, OverflowError):
amount = 0
amount = max(0, min(amount, 10_000_000))
if marker != "minigames":
return False
with self._lock:
row = self.conn.execute(
"SELECT xp, minigames_seeded FROM xp_profile WHERE id = 1"
).fetchone()
xp_now, seeded = (row[0], row[1]) if row else (0, 0)
if seeded or xp_now > 0 or amount <= 0:
if not seeded:
self.conn.execute("UPDATE xp_profile SET minigames_seeded = 1 WHERE id = 1")
self.conn.commit()
return False
self.conn.execute(
"UPDATE xp_profile SET xp = ?, minigames_seeded = 1, updated_at = datetime('now') WHERE id = 1",
(amount,),
)
# Record the seeded amount in the source ledger too, so a later
# minigames reset subtracts the migrated XP rather than orphaning it.
self.conn.execute(
"INSERT INTO xp_sources (source, xp) VALUES (?, ?) "
"ON CONFLICT(source) DO UPDATE SET xp = xp + ?",
(marker, amount, amount),
)
self.conn.commit()
return True
# ── Streak ──────────────────────────────────────────────────────────────
def record_active_day(self, today: str) -> dict:
"""Mark `today` (YYYY-MM-DD, local) as an active day. Any session on a
calendar day keeps the streak: yesterday→+1, today→unchanged, gap or
first-ever→reset to 1. Updates best_streak."""
from datetime import date, timedelta
with self._lock:
row = self.conn.execute(
"SELECT current_streak, best_streak, last_active_date FROM profile_progress WHERE id = 1"
).fetchone()
cur, best, last = (row[0], row[1], row[2]) if row else (0, 0, None)
if last != today:
try:
yesterday = (date.fromisoformat(today) - timedelta(days=1)).isoformat()
except ValueError:
yesterday = None
cur = cur + 1 if (last and last == yesterday) else 1
best = max(best or 0, cur)
self.conn.execute(
"UPDATE profile_progress SET current_streak = ?, best_streak = ?, last_active_date = ? WHERE id = 1",
(cur, best, today),
)
self.conn.commit()
last = today
return {"current_streak": cur, "best_streak": best, "last_active_date": last}
def get_progress(self) -> dict:
"""The full profile-badge payload: XP/level (lib/xp) + streak."""
from xp import progress as _xp_progress
p = self.conn.execute(
"SELECT current_streak, best_streak, last_active_date FROM profile_progress WHERE id = 1"
).fetchone()
cur, best, last = (p[0], p[1], p[2]) if p else (0, 0, None)
out = _xp_progress(self.get_xp())
out.update({"current_streak": cur, "best_streak": best, "last_active_date": last})
return out
# ── Progression (spec 010): paths, challenges, quests, wallet, shop ────
# Lock discipline: self._lock is NOT reentrant and award_xp() takes it, so
# record_progression_event() applies state inside the lock but awards quest
# dB (and re-enters for quest_completed goals) only after releasing it.
def get_progression_state(self) -> dict:
row = self.conn.execute(
"SELECT calibration_status, calibration_completed_at FROM progression_state WHERE id = 1"
).fetchone()
status = row[0] if row else "pending"
return {"calibration_status": status, "calibration_completed_at": row[1] if row else None}
def skip_calibration(self) -> dict:
"""pending → skipped (no-op once completed/skipped). Either way the
player holds onboarding rank 1 afterwards."""
with self._lock:
self.conn.execute(
"UPDATE progression_state SET calibration_status = 'skipped' "
"WHERE id = 1 AND calibration_status = 'pending'"
)
self.conn.commit()
return self.get_progression_state()
def get_player_paths(self) -> dict:
"""{path_id: level} for every selected path."""
rows = self.conn.execute("SELECT path_id, level FROM player_paths").fetchall()
return {r[0]: int(r[1]) for r in rows}
def add_player_paths(self, path_ids) -> dict:
"""Select paths (idempotent; re-adding never resets a level)."""
with self._lock:
for pid in path_ids:
self.conn.execute(
"INSERT OR IGNORE INTO player_paths (path_id, level, selected_at) "
"VALUES (?, 0, datetime('now'))",
(pid,),
)
self.conn.commit()
return self.get_player_paths()
def get_challenge_state(self) -> dict:
"""{challenge_id: {count, completed, detail}} for every touched challenge."""
rows = self.conn.execute(
"SELECT challenge_id, count, progress_detail, completed_at FROM challenge_progress"
).fetchall()
out = {}
for cid, count, detail, completed_at in rows:
try:
parsed = json.loads(detail) if detail else None
except (ValueError, TypeError):
parsed = None
out[cid] = {
"count": int(count or 0),
"completed": completed_at is not None,
"completed_at": completed_at,
"detail": parsed,
}
return out
def ensure_quest_period(self, content, now) -> None:
"""Lazily instantiate the current daily/weekly quest rows (deterministic
per period key; rewards snapshot so live quests survive content edits)."""
import progression as progression_mod
keys = progression_mod.period_keys(now)
with self._lock:
for period_type in ("daily", "weekly"):
cfg = (content.get("quests") or {}).get(period_type) or {}
pool = cfg.get("pool") or {}
count = int(cfg.get("count") or 0)
if not pool or count < 1:
continue
key = keys[period_type]
exists = self.conn.execute(
"SELECT 1 FROM quest_state WHERE period_type = ? AND period_key = ? LIMIT 1",
(period_type, key),
).fetchone()
if exists:
continue
for qid in progression_mod.select_quests(pool.keys(), period_type, key, count):
self.conn.execute(
"INSERT OR IGNORE INTO quest_state "
"(period_type, period_key, quest_id, reward_db) VALUES (?, ?, ?, ?)",
(period_type, key, qid, int(pool[qid].get("reward_db") or 0)),
)
self.conn.commit()
def get_quest_rows(self, period_keys_map: dict) -> list:
"""Current-period quest instances as snapshot/API rows."""
out = []
for period_type, key in period_keys_map.items():
rows = self.conn.execute(
"SELECT quest_id, count, reward_db, progress_detail, completed_at "
"FROM quest_state WHERE period_type = ? AND period_key = ? ORDER BY quest_id",
(period_type, key),
).fetchall()
for qid, count, reward, detail, completed_at in rows:
try:
parsed = json.loads(detail) if detail else None
except (ValueError, TypeError):
parsed = None
out.append({
"period_type": period_type,
"period_key": key,
"quest_id": qid,
"count": int(count or 0),
"reward_db": int(reward or 0),
"detail": parsed,
"completed": completed_at is not None,
"completed_at": completed_at,
})
return out
def get_wallet(self) -> dict:
"""{balance, lifetime_db, spent} — see the wallet table comment for
why spend never mutates xp_profile.xp."""
import progression as progression_mod
row = self.conn.execute("SELECT spent FROM wallet WHERE id = 1").fetchone()
spent = int(row[0]) if row and row[0] else 0
lifetime = self.get_xp()
return {
"balance": progression_mod.wallet_balance(lifetime, spent),
"lifetime_db": lifetime,
"spent": spent,
}
def buy_shop_item(self, item: dict) -> tuple:
"""Atomic purchase: balance check + spend + ownership in one
transaction. Returns ("ok"|"owned"|"insufficient", wallet)."""
with self._lock:
owned = self.conn.execute(
"SELECT 1 FROM shop_owned WHERE item_id = ?", (item["id"],)
).fetchone()
if owned:
status = "owned"
else:
xp_row = self.conn.execute("SELECT xp FROM xp_profile WHERE id = 1").fetchone()
spent_row = self.conn.execute("SELECT spent FROM wallet WHERE id = 1").fetchone()
balance = max(0, int(xp_row[0] if xp_row else 0) - int(spent_row[0] if spent_row else 0))
cost = int(item.get("cost") or 0)
if cost < 0:
status = "invalid"
elif balance < cost:
status = "insufficient"
else:
self.conn.execute(
"UPDATE wallet SET spent = spent + ? WHERE id = 1", (cost,)
)
self.conn.execute(
"INSERT INTO shop_owned (item_id, cost_paid, acquired_at) "
"VALUES (?, ?, datetime('now'))",
(item["id"], cost),
)
self.conn.commit()
status = "ok"
return status, self.get_wallet()
def get_owned_items(self) -> dict:
rows = self.conn.execute(
"SELECT item_id, cost_paid, acquired_at FROM shop_owned"
).fetchall()
return {r[0]: {"cost_paid": int(r[1] or 0), "acquired_at": r[2]} for r in rows}
def get_equipped(self) -> dict:
rows = self.conn.execute("SELECT slot, item_id FROM shop_equipped").fetchall()
return {r[0]: r[1] for r in rows if r[1]}
def equip_item(self, slot: str, item_id) -> dict:
"""Equip an owned item into a slot (item_id=None unequips)."""
with self._lock:
if item_id is None:
self.conn.execute("DELETE FROM shop_equipped WHERE slot = ?", (slot,))
else:
self.conn.execute(
"INSERT INTO shop_equipped (slot, item_id) VALUES (?, ?) "
"ON CONFLICT(slot) DO UPDATE SET item_id = excluded.item_id",
(slot, item_id),
)
self.conn.commit()
return self.get_equipped()
def progression_snapshot(self, content, now) -> dict:
"""The plain-dict state view lib/progression.evaluate_event reads."""
import progression as progression_mod
keys = progression_mod.period_keys(now)
streak_row = self.conn.execute(
"SELECT current_streak FROM profile_progress WHERE id = 1"
).fetchone()
return {
"calibration_status": self.get_progression_state()["calibration_status"],
"paths": self.get_player_paths(),
"challenges": self.get_challenge_state(),
"quests": self.get_quest_rows(keys),
"streak": int(streak_row[0]) if streak_row and streak_row[0] else 0,
"xp_total": self.get_xp(),
}
def record_progression_event(self, event_type: str, payload, content,
now=None, _depth: int = 0) -> dict:
"""The single progression choke point: evaluate one event, persist the
deltas, award quest dB, and re-enter once for quest_completed goals.
Returns a toast-ready summary."""
import progression as progression_mod
from datetime import datetime as _dt
now = now or _dt.now()
self.ensure_quest_period(content, now)
snapshot = self.progression_snapshot(content, now)
outcome = progression_mod.evaluate_event(
{"type": event_type, "payload": payload or {}}, content, snapshot
)
keys = progression_mod.period_keys(now)
challenge_index = content.get("challenge_index") or {}
quest_pools = content.get("quests") or {}
summary = {
"challenges_completed": [],
"quests_completed": [],
"level_ups": list(outcome["level_ups"]),
"calibration_completed": bool(outcome["calibration_completed"]),
}
with self._lock:
for ch in outcome["challenges"]:
detail = json.dumps(ch["detail"]) if ch.get("detail") else None
self.conn.execute(
"INSERT INTO challenge_progress "
"(challenge_id, path_id, level, count, progress_detail, completed_at) "
"VALUES (?, ?, ?, ?, ?, CASE WHEN ? THEN datetime('now') END) "
"ON CONFLICT(challenge_id) DO UPDATE SET "
"count = excluded.count, progress_detail = excluded.progress_detail, "
"completed_at = COALESCE(challenge_progress.completed_at, excluded.completed_at)",
(ch["challenge_id"], ch["path_id"], ch["level"], ch["count"],
detail, 1 if ch["completed"] else 0),
)
if ch["completed"]:
info = challenge_index.get(ch["challenge_id"]) or {}
title = (info.get("challenge") or {}).get("title") or ch["challenge_id"]
summary["challenges_completed"].append(
{"id": ch["challenge_id"], "title": title, "path_id": ch["path_id"]}
)
for lu in outcome["level_ups"]:
# Guard on the old level so a stale evaluation can't double-bump.
self.conn.execute(
"UPDATE player_paths SET level = ? WHERE path_id = ? AND level = ?",
(lu["new_level"], lu["path_id"], lu["new_level"] - 1),
)
# Only quests whose row actually TRANSITIONED to completed in this
# call get rewarded/re-entered. The pure outcome was computed from
# a pre-lock snapshot, so a concurrent event may have completed the
# same quest first — its guarded UPDATE (completed_at IS NULL)
# then touches 0 rows here, and paying it again would double-award
# Decibels and double-advance quest_completed challenges.
newly_completed_quests = []
for q in outcome["quests"]:
detail = json.dumps(q["detail"]) if q.get("detail") else None
cur = self.conn.execute(
"UPDATE quest_state SET count = ?, progress_detail = ?, "
"completed_at = COALESCE(completed_at, CASE WHEN ? THEN datetime('now') END) "
"WHERE period_type = ? AND period_key = ? AND quest_id = ? AND completed_at IS NULL",
(q["count"], detail, 1 if q["completed"] else 0,
q["period_type"], keys.get(q["period_type"], ""), q["quest_id"]),
)
if q["completed"] and cur.rowcount > 0:
newly_completed_quests.append(q)
if outcome["calibration_completed"]:
self.conn.execute(
"UPDATE progression_state SET calibration_status = 'completed', "
"calibration_completed_at = datetime('now') "
"WHERE id = 1 AND calibration_status != 'completed'"
)
self.conn.commit()
# Quest awards + bounded re-entry, outside the lock (award_xp locks).
for q in newly_completed_quests:
pool = (quest_pools.get(q["period_type"]) or {}).get("pool") or {}
qdef = pool.get(q["quest_id"]) or {}
summary["quests_completed"].append({
"id": q["quest_id"],
"title": qdef.get("title") or q["quest_id"],
"period_type": q["period_type"],
"reward_db": q["reward_db"],
})
if q["reward_db"]:
self.award_xp(q["reward_db"], "quests")
if _depth < 1:
sub = self.record_progression_event(
"quest_completed",
{"period_type": q["period_type"], "quest_id": q["quest_id"]},
content, now=now, _depth=_depth + 1,
)
summary["challenges_completed"].extend(sub["challenges_completed"])
summary["quests_completed"].extend(sub["quests_completed"])
summary["level_ups"].extend(sub["level_ups"])
summary["mastery_rank"] = progression_mod.mastery_rank(
self.get_progression_state()["calibration_status"], self.get_player_paths()
)
return summary
# ── Per-song practice stats ───────────────────────────────────────────---
_STATS_COLS = (
"filename", "arrangement", "plays", "best_score", "best_accuracy",
"last_score", "last_accuracy", "last_position", "last_played_at", "updated_at",
)
def _stats_row(self, filename: str, arrangement: int) -> dict | None:
r = self.conn.execute(
"SELECT " + ", ".join(self._STATS_COLS) +
" FROM song_stats WHERE filename = ? AND arrangement = ?",
(filename, int(arrangement)),
).fetchone()
return dict(zip(self._STATS_COLS, r)) if r else None
# Constant SQL fragment restricting stats reads to songs that still exist.
# Unconditional: a genuinely empty (but scanned) library must still hide
# stale stats/playlist ghosts. We rely on `songs` NEVER being transiently
# empty mid-scan — /api/rescan/full bumps mtime to force a full re-scan
# rather than DELETEing rows — so the only times `songs` is empty are a
# fresh install (no stats anyway) or a truly empty library (ghosts should be
# hidden). Race-free orphan handling: dead-song stats are hidden here, never
# deleted on scan (see delete_missing).
_EXISTING_SONG_FILTER = " AND filename IN (SELECT filename FROM songs) "
def _existing_song_filter(self) -> str:
return self._EXISTING_SONG_FILTER
def record_session(self, filename: str, arrangement: int, *, score: int,
accuracy: float, last_position=None) -> dict:
"""Record a scored play: plays += 1, best_* = max, last_* = new."""
from song_score import merge_stats
with self._lock:
existing = self._stats_row(filename, int(arrangement))
merged = merge_stats(existing, {
"score": score, "accuracy": accuracy, "last_position": last_position,
})
self.conn.execute(
"""INSERT INTO song_stats
(filename, arrangement, plays, best_score, best_accuracy,
last_score, last_accuracy, last_position, last_played_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?,
strftime('%Y-%m-%d %H:%M:%f','now'), strftime('%Y-%m-%d %H:%M:%f','now'))
ON CONFLICT(filename, arrangement) DO UPDATE SET
plays = excluded.plays,
best_score = excluded.best_score,
best_accuracy = excluded.best_accuracy,
last_score = excluded.last_score,
last_accuracy = excluded.last_accuracy,
last_position = excluded.last_position,
last_played_at = excluded.last_played_at,
updated_at = excluded.updated_at""",
(filename, int(arrangement), merged["plays"], merged["best_score"],
merged["best_accuracy"], merged["last_score"], merged["last_accuracy"],
merged["last_position"]),
)
self.conn.commit()
return self._stats_row(filename, int(arrangement))
def touch_position(self, filename: str, arrangement: int, last_position: float) -> dict:
"""Persist just the resume position (no plays/score change), so
Continue-Playing works for non-scored plays. Also stamps
last_played_at — both /api/stats/recent and /api/session/continue
filter/order on it, so a position-only touch must set it or the song
never surfaces as 'recent' / 'continue playing'."""
with self._lock:
self.conn.execute(
"""INSERT INTO song_stats (filename, arrangement, last_position,
last_played_at, updated_at)
VALUES (?, ?, ?, strftime('%Y-%m-%d %H:%M:%f','now'),
strftime('%Y-%m-%d %H:%M:%f','now'))
ON CONFLICT(filename, arrangement) DO UPDATE SET
last_position = excluded.last_position,
last_played_at = excluded.last_played_at,
updated_at = excluded.updated_at""",
(filename, int(arrangement), float(last_position)),
)
self.conn.commit()
return self._stats_row(filename, int(arrangement))
def get_song_stats(self, filename: str) -> dict:
"""Best/last/plays across all arrangements of a song, plus per-arrangement rows."""
rows = self.conn.execute(
"SELECT " + ", ".join(self._STATS_COLS) +
" FROM song_stats WHERE filename = ? ORDER BY arrangement",
(filename,),
).fetchall()
arr = [dict(zip(self._STATS_COLS, r)) for r in rows]
best_acc = max((a["best_accuracy"] for a in arr), default=0.0)
best_score = max((a["best_score"] for a in arr), default=0)
plays = sum(a["plays"] for a in arr)
return {
"filename": filename,
"best_accuracy": best_acc,
"best_score": best_score,
"plays": plays,
"arrangements": arr,
}
def recent_stats(self, limit: int = 12) -> list[dict]:
"""Recently-played rows (most recent first) for 'Jump back in'."""
limit = max(1, min(100, int(limit)))
rows = self.conn.execute(
"SELECT " + ", ".join(self._STATS_COLS) +
" FROM song_stats WHERE last_played_at IS NOT NULL " +
self._existing_song_filter() +
"ORDER BY last_played_at DESC LIMIT ?",
(limit,),
).fetchall()
return [dict(zip(self._STATS_COLS, r)) for r in rows]
def best_accuracy_map(self) -> dict:
"""{filename: best_accuracy} across all arrangements, for batch-badging
the library grid in one request. Includes every SCORED song (plays > 0)
— even a genuine 0% best — but excludes resume-only rows (plays == 0,
which carry a default best_accuracy of 0 and shouldn't badge)."""
rows = self.conn.execute(
"SELECT filename, MAX(best_accuracy), SUM(plays) FROM song_stats "
"WHERE 1=1 " + self._existing_song_filter() + # skip dead songs (race-free)
"GROUP BY filename"
).fetchall()
return {r[0]: r[1] for r in rows if r[2] and r[2] > 0}
def top_stats(self, limit: int = 5) -> list[dict]:
"""Top scored songs (best score first) for the profile 'Your best
scores' panel. Aggregated per-song across arrangements (best score,
best accuracy, total plays), only SCORED songs (plays > 0), dead songs
skipped. Mirrors best_accuracy_map's grouping; enriched with metadata
by the /api/stats/top route."""
limit = max(1, min(50, int(limit)))
rows = self.conn.execute(
"SELECT filename, MAX(best_score), MAX(best_accuracy), SUM(plays) "
"FROM song_stats WHERE 1=1 " + self._existing_song_filter() + # skip dead songs
"GROUP BY filename HAVING SUM(plays) > 0 "
"ORDER BY MAX(best_score) DESC, MAX(best_accuracy) DESC LIMIT ?",
(limit,),
).fetchall()
return [
{"filename": r[0], "best_score": r[1], "best_accuracy": r[2], "plays": r[3]}
for r in rows
]
# ── Playlists ─────────────────────────────────────────────────────────--
SAVED_KEY = "saved_for_later"
def _playlist_count(self, pid: int) -> int:
# Count only songs that still exist (mirrors the stats read-filter — dead
# songs are hidden, not deleted on scan), passing through when the songs
# table is empty. Single statement → no probe-then-read race.
return self.conn.execute(
"SELECT COUNT(*) FROM playlist_songs ps WHERE ps.playlist_id = ? "
"AND EXISTS (SELECT 1 FROM songs s WHERE s.filename = ps.filename)",
(pid,),
).fetchone()[0]
def arrangement_count(self, filename: str):
"""Number of arrangements for a song, or None if the song isn't in the
library (so callers can skip validation when it can't be checked)."""
row = self.conn.execute("SELECT arrangements FROM songs WHERE filename = ?", (filename,)).fetchone()
if not row or not row[0]:
return None
try:
arr = json.loads(row[0])
except (ValueError, TypeError):
return None
return len(arr) if isinstance(arr, list) else None
def arrangement_entry(self, filename: str, index: int):
"""One arrangement's metadata dict for a library song, or None when
the song/index is unknown (progression then falls back to guitar)."""
row = self.conn.execute("SELECT arrangements FROM songs WHERE filename = ?", (filename,)).fetchone()
if not row or not row[0]:
return None
try:
arr = json.loads(row[0])
except (ValueError, TypeError):
return None
if isinstance(arr, list) and 0 <= index < len(arr) and isinstance(arr[index], dict):
return arr[index]
return None
def list_playlists(self) -> list[dict]:
from urllib.parse import quote
rows = self.conn.execute(
"SELECT id, name, system_key, created_at, updated_at FROM playlists "
"WHERE rules IS NULL " # smart collections live in the source picker, not here
"ORDER BY (system_key IS NULL), name COLLATE NOCASE"
).fetchall()
out = []
for r in rows:
pid = r[0]
# First few still-present songs (in order) → art URLs, for a
# content-dependent playlist cover (single art / 2x2 mosaic). The
# JOIN drops dead songs, matching get_playlist's visibility.
arts = self.conn.execute(
"SELECT ps.filename FROM playlist_songs ps "
"JOIN songs s ON s.filename = ps.filename "
"WHERE ps.playlist_id = ? ORDER BY ps.position LIMIT 4",
(pid,),
).fetchall()
out.append({
"id": pid, "name": r[1], "system_key": r[2],
"created_at": r[3], "updated_at": r[4],
"count": self._playlist_count(pid),
"art_urls": [f"/api/song/{quote(a[0])}/art" for a in arts],
})
return out
def create_playlist(self, name: str, system_key: str | None = None) -> dict:
with self._lock:
cur = self.conn.execute(
"INSERT INTO playlists (name, system_key, created_at, updated_at) "
"VALUES (?, ?, datetime('now'), datetime('now'))",
(name, system_key),
)
self.conn.commit()
pid = cur.lastrowid
return self.get_playlist(pid)
def saved_playlist_id(self) -> int:
"""Id of the reserved Saved-for-Later playlist, created on first use.
Tolerates a create race: two concurrent first-use toggles can both see
no row and try to insert; the unique system_key index makes the loser
raise IntegrityError, so catch it and re-read the winner's row rather
than 500."""
row = self.conn.execute(
"SELECT id FROM playlists WHERE system_key = ?", (self.SAVED_KEY,)
).fetchone()
if row:
return row[0]
try:
return self.create_playlist("Saved for Later", self.SAVED_KEY)["id"]
except sqlite3.IntegrityError:
row = self.conn.execute(
"SELECT id FROM playlists WHERE system_key = ?", (self.SAVED_KEY,)
).fetchone()
if row:
return row[0]
raise
def rename_playlist(self, pid: int, name: str) -> bool:
with self._lock:
cur = self.conn.execute(
"UPDATE playlists SET name = ?, updated_at = datetime('now') WHERE id = ?",
(name, pid),
)
self.conn.commit()
return cur.rowcount > 0
def delete_playlist(self, pid: int) -> bool:
"""Delete a user playlist (system playlists are protected — caller checks)."""
with self._lock:
self.conn.execute("DELETE FROM playlist_songs WHERE playlist_id = ?", (pid,))
cur = self.conn.execute("DELETE FROM playlists WHERE id = ?", (pid,))
self.conn.commit()
return cur.rowcount > 0
# ── Smart collections (feedBack#636 item 2) ───────────────────────────
@staticmethod
def _collection_row(r) -> dict:
rules = {}
if r[3]:
try:
parsed = json.loads(r[3])
if isinstance(parsed, dict):
rules = parsed
except (ValueError, TypeError):
rules = {}
return {"id": r[0], "name": r[1], "system_key": r[2], "rules": rules,
"created_at": r[4], "updated_at": r[5]}
def is_collection(self, pid: int) -> bool:
row = self.conn.execute(
"SELECT rules IS NOT NULL FROM playlists WHERE id = ?", (pid,)
).fetchone()
return bool(row and row[0])
def list_collections(self) -> list[dict]:
rows = self.conn.execute(
"SELECT id, name, system_key, rules, created_at, updated_at FROM playlists "
"WHERE rules IS NOT NULL ORDER BY name COLLATE NOCASE"
).fetchall()
return [self._collection_row(r) for r in rows]
def get_collection(self, pid: int) -> dict | None:
r = self.conn.execute(
"SELECT id, name, system_key, rules, created_at, updated_at FROM playlists "
"WHERE id = ? AND rules IS NOT NULL", (pid,)
).fetchone()
return self._collection_row(r) if r else None
def create_collection(self, name: str, rules: dict) -> dict:
with self._lock:
cur = self.conn.execute(
"INSERT INTO playlists (name, system_key, rules, created_at, updated_at) "
"VALUES (?, NULL, ?, datetime('now'), datetime('now'))",
(name, json.dumps(rules or {})),
)
self.conn.commit()
pid = cur.lastrowid
return self.get_collection(pid)
def update_collection(self, pid: int, name: str | None = None,
rules: dict | None = None) -> dict | None:
if not self.is_collection(pid):
return None
with self._lock:
if name is not None:
self.conn.execute("UPDATE playlists SET name = ? WHERE id = ?", (name, pid))
if rules is not None:
self.conn.execute("UPDATE playlists SET rules = ? WHERE id = ?",
(json.dumps(rules or {}), pid))
self.conn.execute("UPDATE playlists SET updated_at = datetime('now') WHERE id = ?", (pid,))
self.conn.commit()
return self.get_collection(pid)
def get_playlist(self, pid: int) -> dict | None:
# A path-param int outside SQLite's 64-bit range raises OverflowError at
# bind time (→ 500). Treat it as a miss; every mutating playlist handler
# gates on this first, so the guard covers them too.
if not isinstance(pid, int) or not (-(2**63) <= pid < 2**63):
return None
# `rules IS NULL` excludes smart collections (#636 item 2): they share
# the playlists table but their membership is rules-based, so every
# manual-playlist mutation (add/remove/reorder/cover) that gates on
# get_playlist uniformly 404s on a collection id — collections are
# managed only through /api/collections.
head = self.conn.execute(
"SELECT id, name, system_key, created_at, updated_at FROM playlists "
"WHERE id = ? AND rules IS NULL", (pid,)
).fetchone()
if not head:
return None
rows = self.conn.execute(
"""SELECT ps.filename, ps.position, s.title, s.artist, s.tuning_name
FROM playlist_songs ps LEFT JOIN songs s ON s.filename = ps.filename
WHERE ps.playlist_id = ?
-- hide dead songs (race-free; not deleted on scan)
AND s.filename IS NOT NULL
ORDER BY ps.position, ps.filename""",
(pid,),
).fetchall()
from urllib.parse import quote
songs = [{
"filename": r[0], "position": r[1],
"title": r[2] or r[0], "artist": r[3] or "", "tuning_name": r[4] or "",
"art_url": f"/api/song/{quote(r[0])}/art",
} for r in rows]
return {
"id": head[0], "name": head[1], "system_key": head[2],
"created_at": head[3], "updated_at": head[4], "songs": songs,
}
def add_playlist_song(self, pid: int, filename: str):
with self._lock:
# Re-check existence INSIDE the lock: the handler's earlier 404 check
# is a separate step, so a concurrent delete_playlist could land
# between them and leave an orphan playlist_songs row. Returning None
# lets the handler answer 404 instead of inserting an orphan.
if not self.conn.execute("SELECT 1 FROM playlists WHERE id = ?", (pid,)).fetchone():
return None
nxt = self.conn.execute(
"SELECT COALESCE(MAX(position), -1) + 1 FROM playlist_songs WHERE playlist_id = ?", (pid,)
).fetchone()[0]
cur = self.conn.execute(
"INSERT OR IGNORE INTO playlist_songs (playlist_id, filename, position) VALUES (?, ?, ?)",
(pid, filename, nxt),
)
self.conn.execute("UPDATE playlists SET updated_at = datetime('now') WHERE id = ?", (pid,))
self.conn.commit()
return cur.rowcount > 0
def remove_playlist_song(self, pid: int, filename: str) -> bool:
with self._lock:
cur = self.conn.execute(
"DELETE FROM playlist_songs WHERE playlist_id = ? AND filename = ?", (pid, filename)
)
self.conn.execute("UPDATE playlists SET updated_at = datetime('now') WHERE id = ?", (pid,))
self.conn.commit()
return cur.rowcount > 0
def reorder_playlist(self, pid: int, ordered_filenames: list[str]) -> bool:
with self._lock:
for pos, fn in enumerate(ordered_filenames):
self.conn.execute(
"UPDATE playlist_songs SET position = ? WHERE playlist_id = ? AND filename = ?",
(pos, pid, fn),
)
self.conn.execute("UPDATE playlists SET updated_at = datetime('now') WHERE id = ?", (pid,))
self.conn.commit()
return True
def toggle_saved(self, filename: str) -> bool:
"""Add/remove a song on the Saved-for-Later playlist. Returns new state.
The presence check and the add/remove run under one lock so two
concurrent toggles of the same song can't both take the add path (or
both remove) and leave an inconsistent saved state."""
pid = self.saved_playlist_id()
with self._lock:
present = self.conn.execute(
"SELECT 1 FROM playlist_songs WHERE playlist_id = ? AND filename = ?", (pid, filename)
).fetchone() is not None
if present:
self.conn.execute(
"DELETE FROM playlist_songs WHERE playlist_id = ? AND filename = ?", (pid, filename))
new_state = False
else:
nxt = self.conn.execute(
"SELECT COALESCE(MAX(position), -1) + 1 FROM playlist_songs WHERE playlist_id = ?", (pid,)
).fetchone()[0]
self.conn.execute(
"INSERT OR IGNORE INTO playlist_songs (playlist_id, filename, position) VALUES (?, ?, ?)",
(pid, filename, nxt))
new_state = True
self.conn.execute("UPDATE playlists SET updated_at = datetime('now') WHERE id = ?", (pid,))
self.conn.commit()
return new_state
# ── Wishlist / "wanted" (feedBack#636 item 4) ─────────────────────────
_WANTED_COLS = ("id", "artist", "title", "source", "source_ref", "note", "created_at")
def add_wanted(self, artist: str, title: str, source: str = "manual",
source_ref: str = "", note: str = "") -> dict:
"""Add a not-owned song to the wishlist (or return the existing row if
an entry with the same identity is already wanted — idempotent, so a
re-run of an ownership-diff doesn't duplicate). Returns the row."""
artist = (artist or "").strip()
title = (title or "").strip()
source = (source or "manual").strip() or "manual"
source_ref = (source_ref or "").strip()
note = (note or "").strip()
with self._lock:
self.conn.execute(
"INSERT OR IGNORE INTO wanted (artist, title, source, source_ref, note, created_at) "
"VALUES (?, ?, ?, ?, ?, datetime('now'))",
(artist, title, source, source_ref, note),
)
row = self.conn.execute(
"SELECT " + ", ".join(self._WANTED_COLS) + " FROM wanted "
"WHERE artist = ? COLLATE NOCASE AND title = ? COLLATE NOCASE "
"AND source = ? AND source_ref = ?",
(artist, title, source, source_ref),
).fetchone()
self.conn.commit()
return dict(zip(self._WANTED_COLS, row)) if row else {}
def list_wanted(self) -> list[dict]:
"""All wishlist entries, newest first."""
rows = self.conn.execute(
"SELECT " + ", ".join(self._WANTED_COLS) + " FROM wanted "
"ORDER BY created_at DESC, id DESC"
).fetchall()
return [dict(zip(self._WANTED_COLS, r)) for r in rows]
def remove_wanted(self, wanted_id: int) -> bool:
"""Drop a wishlist entry by id. Returns True if a row was removed."""
with self._lock:
cur = self.conn.execute("DELETE FROM wanted WHERE id = ?", (wanted_id,))
self.conn.commit()
return cur.rowcount > 0
def count_wanted(self) -> int:
return self.conn.execute("SELECT COUNT(*) FROM wanted").fetchone()[0]
def continue_session(self) -> dict | None:
"""Most-recently-played song (from song_stats) + metadata, for the
Continue-Playing card. Null when nothing has been played."""
row = self.conn.execute(
"SELECT filename, arrangement, last_position FROM song_stats "
"WHERE last_played_at IS NOT NULL " +
self._existing_song_filter() + # skip dead songs (race-free)
"ORDER BY last_played_at DESC LIMIT 1"
).fetchone()
if not row:
return None
filename, arrangement, last_position = row
meta = self.conn.execute(
"SELECT title, artist, tuning_name, duration FROM songs WHERE filename = ?", (filename,)
).fetchone()
title, artist, tuning_name, duration = meta if meta else (None, None, None, None)
from urllib.parse import quote
return {
"filename": filename, "arrangement": arrangement,
"title": title or filename, "artist": artist or "",
"tuning_name": tuning_name or "", "duration": duration or 0,
"last_position": last_position,
"art_url": f"/api/song/{quote(filename)}/art",
}
def favorite_set(self) -> set[str]:
return {r[0] for r in self.conn.execute("SELECT filename FROM favorites").fetchall()}
def get(self, filename: str, mtime: float, size: int) -> dict | None:
cache_key = str(filename)
with self._lock:
row = self.conn.execute(
"SELECT mtime, size, title, artist, album, year, duration, tuning, arrangements, has_lyrics, "
"format, stem_count, stem_ids, tuning_name, tuning_sort_key, tuning_offsets "
"FROM songs WHERE filename = ?", (cache_key,)
).fetchone()
if row and row[0] == mtime and row[1] == size and row[2]:
return {
"title": row[2], "artist": row[3], "album": row[4],
"year": row[5], "duration": row[6], "tuning": row[7],
"arrangements": json.loads(row[8]) if row[8] else [],
"has_lyrics": bool(row[9]),
"format": row[10] or "archive",
"stem_count": int(row[11] or 0),
"stem_ids": json.loads(row[12]) if row[12] else [],
"tuning_name": row[13] or "",
"tuning_sort_key": int(row[14] or 0),
"tuning_offsets": row[15] or "",
}
return None
def put(self, filename: str, mtime: float, size: int, meta: dict):
with self._lock:
self.conn.execute(
"INSERT OR REPLACE INTO songs "
"(filename, mtime, size, title, artist, album, year, duration, tuning, arrangements, "
"has_lyrics, format, stem_count, stem_ids, tuning_name, tuning_sort_key, tuning_offsets) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(filename, mtime, size, meta.get("title", ""), meta.get("artist", ""),
meta.get("album", ""), meta.get("year", ""), meta.get("duration", 0),
meta.get("tuning", ""), json.dumps(meta.get("arrangements", [])),
1 if meta.get("has_lyrics") else 0,
meta.get("format", "archive"),
int(meta.get("stem_count", 0) or 0),
json.dumps(meta.get("stem_ids", []) or []),
meta.get("tuning_name", "") or "",
int(meta.get("tuning_sort_key", 0) or 0),
meta.get("tuning_offsets", "") or ""),
)
self.conn.commit()
def count(self) -> int:
return self.conn.execute("SELECT COUNT(*) FROM songs WHERE title != ''").fetchone()[0]
def delete_missing(self, current_filenames: set[str]):
"""Remove `songs` rows for files no longer on disk.
Deliberately does NOT purge song_stats / playlist_songs here: a scan is a
point-in-time snapshot, so a song that briefly disappears mid-scan (e.g.
a directory-form .sloppak being overwritten via rmtree-then-extract, or a
delete+reupload) and returns under the same filename would otherwise lose
its stats/playlist membership permanently. Instead, stats are purged on
the EXPLICIT delete path (DELETE /api/song) and dead-song rows are
filtered at read time (recent_stats / continue_session /
best_accuracy_map gate on the song still existing)."""
with self._lock:
db_files = {r[0] for r in self.conn.execute("SELECT filename FROM songs").fetchall()}
stale = db_files - current_filenames
if stale:
self.conn.executemany("DELETE FROM songs WHERE filename = ?", [(f,) for f in stale])
self.conn.commit()
return len(stale)
def _estd_set(self) -> set[str]:
"""Get set of filenames that have a retuned variant (_EStd_ or _DropD_) in the DB."""
rows = self.conn.execute(
"SELECT filename FROM songs WHERE filename LIKE '%\\_EStd\\_%' ESCAPE '\\' "
"OR filename LIKE '%\\_DropD\\_%' ESCAPE '\\'"
).fetchall()
originals = set()
for (fname,) in rows:
originals.add(fname.replace("_EStd_", "_").replace("_DropD_", "_"))
return originals
# Manifest-allowed filter values. Whitelisted before binding so a
# malformed query string can't push arbitrary text through to SQL —
# parameters are bound, but capping the input space is still cheap
# defense-in-depth (see feedBack#129).
_ALLOWED_ARRANGEMENT_NAMES = {"Lead", "Rhythm", "Bass", "Combo"}
# Per-smart-type list of (sql_op, sql_param) pairs appended to the SQL
# name-fallback branch (key-absent smart_name). Covers legacy raw names
# and load_song()'s synthesised display names that map to each smart type.
_SMART_NULL_FALLBACK_EXTRAS: dict[str, tuple[tuple[str, str], ...]] = {
"Lead": (("=", "Combo"), ("LIKE", "Alt. Combo%"), ("LIKE", "Bonus Combo%")),
"Bass": (("=", "Bass 2"),),
}
# Stem ids match the bare strings sloppak manifests use today —
# `full`, `guitar`, `bass`, `drums`, `vocals`, `piano`, `other`. The
# frontend filter UI omits `full` (it's the always-on fallback mix
# and would match every sloppak), but the server-side whitelist
# keeps it so a hand-rolled API client can still ask for it.
_ALLOWED_STEM_IDS = {"full", "guitar", "bass", "drums", "vocals", "piano", "other"}
@classmethod
def _smart_null_extras(cls, arr_type: str) -> tuple[str, list[str]]:
"""Return (sql_fragment, bound_params) for the extra raw-name terms to
OR into the key-absent NULL-smart_name fallback branch for arr_type.
Empty when no extras are defined."""
terms = cls._SMART_NULL_FALLBACK_EXTRAS.get(arr_type, ())
fragment = "".join(
f" OR json_extract(value, '$.name') {op} ?" for op, _ in terms
)
return fragment, [val for _, val in terms]
def _build_where(self, q: str = "", favorites_only: bool = False,
format_filter: str = "",
artist_filter: str = "",
album_filter: str = "",
arrangements_has: list[str] | None = None,
arrangements_lacks: list[str] | None = None,
stems_has: list[str] | None = None,
stems_lacks: list[str] | None = None,
has_lyrics: int | None = None,
tunings: list[str] | None = None,
naming_mode: str = "legacy") -> tuple[str, list]:
"""Shared WHERE-clause builder for query_page / query_artists /
query_stats. Returns (where_sql, params). Leading 'WHERE' is
included so callers paste it directly. See feedBack#129/#69.
"""
where = "WHERE title != ''"
params: list = []
if favorites_only:
where += " AND filename IN (SELECT filename FROM favorites)"
if format_filter:
where += " AND format = ?"
params.append(format_filter)
if artist_filter:
where += " AND artist = ? COLLATE NOCASE"
params.append(artist_filter)
if album_filter:
where += " AND album = ? COLLATE NOCASE"
params.append(album_filter)
if q:
where += " AND (title LIKE ? COLLATE NOCASE OR artist LIKE ? COLLATE NOCASE OR album LIKE ? COLLATE NOCASE)"
params += [f"%{q}%"] * 3
# arrangements_has / arrangements_lacks: OR within axis (any-of).
# Uses JSON1's json_each which yields one row per arrangement, then
# matches the relevant field. The whole subquery is wrapped in EXISTS
# so we don't multiply rows in the outer SELECT.
#
# Smart mode: each requested type (Lead/Rhythm/Bass) matches against
# smart_name when present. "Lead" matches smart_name in
# ('Lead', 'Alt. Lead', 'Alt. Lead N', 'Bonus Lead', 'Bonus Lead N').
# Falls back to matching `name` for older rows without smart_name.
# Legacy mode: matches `name` directly (original behaviour).
arr_has = [a for a in (arrangements_has or []) if a in self._ALLOWED_ARRANGEMENT_NAMES]
if arr_has and naming_mode == "smart":
# Smart mode subsumes "Combo" into "Lead" — normalize here so a
# hand-rolled API client matches the client-side behaviour and
# the SQL doesn't need a "Combo" smart-type branch.
arr_has = list(dict.fromkeys("Lead" if a == "Combo" else a for a in arr_has))
if arr_has:
if naming_mode == "smart":
clauses = []
for arr_type in arr_has:
# Extra raw-name fragments matched only in the key-absent
# NULL-smart_name fallback branch — they cover the legacy
# display names that map to this smart type:
# Lead: "Combo" (combined guitar) + Alt./Bonus Combo
# Bass: "Bass 2" (load_song synthesises for real_bass_22)
extra_null, extra_null_params = self._smart_null_extras(arr_type)
# json_type() returns NULL when the key is absent and the
# string 'null' when the key exists with explicit JSON null
# (set by the scanner for ambiguous duplicate-name rows).
# Name-fallback only applies to key-absent rows so an
# explicit null suppresses the fallback and lets the
# background rescan resolve the ambiguity authoritatively.
clauses.append(
"(json_extract(value, '$.smart_name') IS NOT NULL AND ("
f"json_extract(value, '$.smart_name') = ? OR "
f"json_extract(value, '$.smart_name') LIKE ? OR "
f"json_extract(value, '$.smart_name') LIKE ?"
")) OR ("
"json_type(value, '$.smart_name') IS NULL AND ("
"json_extract(value, '$.name') = ? OR "
"json_extract(value, '$.name') LIKE ? OR "
f"json_extract(value, '$.name') LIKE ?{extra_null}))"
)
params += [
arr_type,
f"Alt. {arr_type}%",
f"Bonus {arr_type}%",
arr_type,
f"Alt. {arr_type}%",
f"Bonus {arr_type}%",
] + extra_null_params
where += (
" AND EXISTS (SELECT 1 FROM json_each(songs.arrangements) WHERE "
+ " OR ".join(f"({c})" for c in clauses)
+ ")"
)
else:
placeholders = ",".join(["?"] * len(arr_has))
where += (" AND EXISTS (SELECT 1 FROM json_each(songs.arrangements) "
f"WHERE json_extract(value, '$.name') IN ({placeholders}))")
params += arr_has
arr_lacks = [a for a in (arrangements_lacks or []) if a in self._ALLOWED_ARRANGEMENT_NAMES]
if arr_lacks and naming_mode == "smart":
arr_lacks = list(dict.fromkeys("Lead" if a == "Combo" else a for a in arr_lacks))
if arr_lacks:
if naming_mode == "smart":
clauses = []
for arr_type in arr_lacks:
extra_null, extra_null_params = self._smart_null_extras(arr_type)
# See "has" branch above for the json_type rationale.
# Extra branch (vs `has`): an explicit smart_name=null
# arrangement is ambiguous; we don't know whether it's
# `arr_type` or not. Be conservative and treat it as
# potentially matching, so `arrangements_lacks` excludes
# the parent row instead of falsely claiming it lacks
# `arr_type`. The background rescan resolves the ambiguity.
clauses.append(
"(json_extract(value, '$.smart_name') IS NOT NULL AND ("
f"json_extract(value, '$.smart_name') = ? OR "
f"json_extract(value, '$.smart_name') LIKE ? OR "
f"json_extract(value, '$.smart_name') LIKE ?"
")) OR ("
"json_type(value, '$.smart_name') = 'null'"
") OR ("
"json_type(value, '$.smart_name') IS NULL AND ("
"json_extract(value, '$.name') = ? OR "
"json_extract(value, '$.name') LIKE ? OR "
f"json_extract(value, '$.name') LIKE ?{extra_null}))"
)
params += [
arr_type,
f"Alt. {arr_type}%",
f"Bonus {arr_type}%",
arr_type,
f"Alt. {arr_type}%",
f"Bonus {arr_type}%",
] + extra_null_params
where += (
" AND NOT EXISTS (SELECT 1 FROM json_each(songs.arrangements) WHERE "
+ " OR ".join(f"({c})" for c in clauses)
+ ")"
)
else:
placeholders = ",".join(["?"] * len(arr_lacks))
where += (" AND NOT EXISTS (SELECT 1 FROM json_each(songs.arrangements) "
f"WHERE json_extract(value, '$.name') IN ({placeholders}))")
params += arr_lacks
stems_h = [s for s in (stems_has or []) if s in self._ALLOWED_STEM_IDS]
if stems_h:
placeholders = ",".join(["?"] * len(stems_h))
where += (" AND EXISTS (SELECT 1 FROM json_each(songs.stem_ids) "
f"WHERE value IN ({placeholders}))")
params += stems_h
stems_l = [s for s in (stems_lacks or []) if s in self._ALLOWED_STEM_IDS]
if stems_l:
placeholders = ",".join(["?"] * len(stems_l))
where += (" AND NOT EXISTS (SELECT 1 FROM json_each(songs.stem_ids) "
f"WHERE value IN ({placeholders}))")
params += stems_l
if has_lyrics in (0, 1):
where += " AND has_lyrics = ?"
params.append(has_lyrics)
if tunings:
# Keep the input cap conservative (32) so a hostile caller
# can't blow out the parameter list. Real tuning sets in the
# wild number in the low double digits.
tn = [t for t in tunings if isinstance(t, str) and t][:32]
if tn:
placeholders = ",".join(["?"] * len(tn))
# Match the same grouping key tuning_names() returns so a single
# "Custom Tuning" pill selects exactly its offset set while named
# tunings still match by name.
where += (f" AND {_TUNING_GROUP_KEY_SQL} "
f"COLLATE NOCASE IN ({placeholders})")
params += tn
return where, params
def query_page(self, q: str = "", page: int = 0, size: int = 24,
sort: str = "artist", direction: str = "asc",
favorites_only: bool = False,
format_filter: str = "",
artist_filter: str = "",
album_filter: str = "",
arrangements_has: list[str] | None = None,
arrangements_lacks: list[str] | None = None,
stems_has: list[str] | None = None,
stems_lacks: list[str] | None = None,
has_lyrics: int | None = None,
tunings: list[str] | None = None,
after: str | None = None,
naming_mode: str = "legacy") -> tuple[list[dict], int]:
"""Server-side paginated search. Returns (songs, total_count).
`after` is an opaque keyset cursor (the last row of the previous page).
When supplied and the sort can keyset, the page is fetched with a
WHERE-seek instead of OFFSET — O(page), independent of depth. Unknown
sorts / bad cursors fall back to OFFSET, so it's always safe."""
where, params = self._build_where(
q=q, favorites_only=favorites_only, format_filter=format_filter,
artist_filter=artist_filter, album_filter=album_filter,
arrangements_has=arrangements_has, arrangements_lacks=arrangements_lacks,
stems_has=stems_has, stems_lacks=stems_lacks,
has_lyrics=has_lyrics, tunings=tunings, naming_mode=naming_mode,
)
sort_map = {
"artist": "artist COLLATE NOCASE", "artist-desc": "artist COLLATE NOCASE DESC",
"title": "title COLLATE NOCASE", "title-desc": "title COLLATE NOCASE DESC",
"recent": "mtime DESC",
# Tuning sort uses musical distance from E Standard
# (feedBack#22 — was alphabetical). `tuning_sort_key` is
# the sum of per-string offsets, so |sort_key| is the
# magnitude of the down/up-tune. ABS ascending puts E
# Standard (0) first, then ±2 (Drop D, F Standard), then
# ±6 (Eb Standard, F# Standard), and so on. Within a
# magnitude tier we break ties by signed key ASC so the
# negative (down-tuned) variant comes before the positive
# (up-tuned) one — Eb Standard before F Standard, matching
# how the app groups its tuning list. Final tiebreak by
# name keeps the order fully deterministic.
#
# Leading term pushes pre-migration / unscanned rows to
# the bottom — without it ABS(0) collides with E
# Standard's 0 and unindexed rows would sort first.
# COALESCE on every column the clause references guards
# against NULL values — SQLite's literal-constant ADD
# COLUMN does backfill on most versions, but raw SQL
# inserts that bypass `put()`, edge-case migration paths,
# or future code that writes None could still leave NULLs
# behind, and a NULL `tuning_name` in `(tuning_name = '')`
# evaluates to NULL itself (which sorts ahead of 0 in
# ASC), defeating the push-to-bottom intent.
"tuning": (
"(COALESCE(tuning_name, '') = '') ASC, "
"ABS(COALESCE(tuning_sort_key, 0)), "
"COALESCE(tuning_sort_key, 0) ASC, "
"COALESCE(tuning_name, '') COLLATE NOCASE"
),
# Year sort (feedBack#128). Empty-year rows pushed to the
# bottom for both directions; otherwise CAST so '2010' >
# '2005' rather than alphabetic.
"year": "(year = '') ASC, CAST(year AS INTEGER) ASC",
"year-desc": "(year = '') ASC, CAST(year AS INTEGER) DESC",
}
order = sort_map.get(sort, "artist COLLATE NOCASE")
# Legacy `dir=desc` toggle: only safe to append on simple sort
# clauses that don't already encode a direction. Compound /
# multi-term entries above (tuning, year, year-desc) bake their
# ASC/DESC into the clause, so a global ` DESC` append would
# produce invalid SQL like `CAST(year AS INTEGER) ASC DESC`.
# Skip the append in that case — clients flipping direction on
# those sorts use the explicit `-desc` sort key instead.
if direction == "desc" and " ASC" not in order and " DESC" not in order:
order += " DESC"
# Unique, deterministic tiebreak → a TOTAL order. Without it, rows with
# an equal sort key can reshuffle between OFFSET pages (skip/dupe); it's
# also what makes keyset seeking correct.
order += ", filename"
total = self.conn.execute(f"SELECT COUNT(*) FROM songs {where}", params).fetchone()[0]
cols = ("SELECT filename, title, artist, album, year, duration, tuning, "
"arrangements, has_lyrics, mtime, format, stem_count, stem_ids, "
"tuning_name, tuning_offsets FROM songs ")
cursor = _decode_cursor(after) if after else None
eff_sort = _effective_keyset_sort(sort, direction)
if cursor and eff_sort in _KEYSET_SORTS:
# Keyset seek: rows strictly after the cursor in the total order
# `