Dashboard detail, library filtering/pagination, simplified playlist times,

job descriptions

Dashboard: format breakdown (FLAC/MP3/etc. counts), approximate library
size (bitrate*length/8, same approximation `beet stats` itself uses --
verified to match its "100.5 GiB" output exactly against the real
library), and a credential auth-state summary per scope (configured y/n
plus, for Bandcamp, cookie expiry parsed locally from the stored cookie
jar -- deliberately not a live network probe like pipeline-status.sh's
Qobuz check, since this renders on every dashboard load).

Library: was one unfiltered page dumping all 4072 tracks. Added search
(artist/title/album), playlist and format filter dropdowns, and real
SQL-level pagination (LIMIT/OFFSET, not fetch-everything-then-slice).
beets_service gained count_items()/distinct_formats() to support this.

Playlists: cron_expr is still the stored/scheduled representation, but the
UI now shows and edits a plain daily time picker instead of raw cron
syntax -- every playlist schedule today is a simple daily HH:MM anyway.
playlist_service.cron_to_time()/time_to_cron() convert at the router
boundary; verified round-trip against all real playlist cron values.

Jobs: each maintenance job now shows a short one-line description of what
it actually does (MAINTENANCE_JOB_DESCRIPTIONS), and "next run" is
formatted consistently with playlists' time style (HH:MM, with a day
qualifier for non-today runs -- maintenance jobs can be weekly/monthly,
unlike playlists' plain daily schedule).

Verified end-to-end against the real production data (4072 tracks, 100.5
GiB, real credentials): stats/format-breakdown/search/pagination all
correct, all 7 credential scopes report configured with Bandcamp's real
cookie expiry (325 days), and a full authenticated page-render sweep
(dashboard, library with every filter combination, playlist detail,
jobs) all returned 200 with no template errors.

app/services/beets_service.py

@@ -34,22 +34,74 @@ def _row_to_dict(row: sqlite3.Row) -> dict:
     return d
 
 
-def query_items(grouping: str | None = None) -> list[dict]:
-    """All items, optionally filtered to one playlist's grouping tag."""
+def query_items(
+    grouping: str | None = None,
+    search: str | None = None,
+    format: str | None = None,
+    limit: int | None = None,
+    offset: int = 0,
+) -> list[dict]:
+    """Items, optionally filtered by playlist grouping, a free-text search
+    over artist/title/album, and/or exact format match. limit/offset give
+    SQL-level pagination -- the library has thousands of tracks, so this
+    must not be a fetch-everything-then-slice-in-Python operation."""
     if not db_exists():
         return []
     cols = ", ".join(_ITEM_COLUMNS)
+    where = []
+    params: list = []
+    if grouping is not None:
+        where.append("grouping = ?")
+        params.append(grouping)
+    if format is not None:
+        where.append("format = ?")
+        params.append(format)
+    if search:
+        where.append("(artist LIKE ? OR title LIKE ? OR albumartist LIKE ?)")
+        like = f"%{search}%"
+        params.extend([like, like, like])
+    where_sql = f"WHERE {' AND '.join(where)}" if where else ""
+    limit_sql = ""
+    if limit is not None:
+        limit_sql = "LIMIT ? OFFSET ?"
+        params.extend([limit, offset])
+
     conn = _connect()
     try:
-        if grouping is not None:
-            cur = conn.execute(f"SELECT {cols} FROM items WHERE grouping = ?", (grouping,))
-        else:
-            cur = conn.execute(f"SELECT {cols} FROM items")
+        cur = conn.execute(
+            f"SELECT {cols} FROM items {where_sql} ORDER BY artist, album, track {limit_sql}",
+            params,
+        )
         return [_row_to_dict(row) for row in cur.fetchall()]
     finally:
         conn.close()
 
 
+def count_items(grouping: str | None = None, search: str | None = None, format: str | None = None) -> int:
+    """Matching row count for query_items()'s filters -- for pagination."""
+    if not db_exists():
+        return 0
+    where = []
+    params: list = []
+    if grouping is not None:
+        where.append("grouping = ?")
+        params.append(grouping)
+    if format is not None:
+        where.append("format = ?")
+        params.append(format)
+    if search:
+        where.append("(artist LIKE ? OR title LIKE ? OR albumartist LIKE ?)")
+        like = f"%{search}%"
+        params.extend([like, like, like])
+    where_sql = f"WHERE {' AND '.join(where)}" if where else ""
+
+    conn = _connect()
+    try:
+        return conn.execute(f"SELECT COUNT(*) FROM items {where_sql}", params).fetchone()[0]
+    finally:
+        conn.close()
+
+
 def get_item(item_id: int) -> dict | None:
     if not db_exists():
         return None
@@ -63,20 +115,55 @@ def get_item(item_id: int) -> dict | None:
 
 
 def stats() -> dict:
-    """Cheap summary for the dashboard and for migration-verification
-    (compare against `beet stats` during Stage 0 cutover)."""
+    """Summary for the dashboard: total tracks, format breakdown, storage
+    estimate (bitrate*length/8, same approximation `beet stats` itself
+    uses -- not a filesystem stat() pass over every file), and groupings."""
     if not db_exists():
-        return {"db_exists": False, "total_tracks": 0, "groupings": []}
+        return {
+            "db_exists": False,
+            "total_tracks": 0,
+            "groupings": [],
+            "formats": [],
+            "total_bytes": 0,
+        }
     conn = _connect()
     try:
         total = conn.execute("SELECT COUNT(*) FROM items").fetchone()[0]
-        rows = conn.execute(
+
+        grouping_rows = conn.execute(
             "SELECT DISTINCT grouping FROM items WHERE grouping IS NOT NULL AND grouping != ''"
         ).fetchall()
+
+        format_rows = conn.execute(
+            "SELECT format, COUNT(*) as n FROM items "
+            "WHERE format IS NOT NULL AND format != '' "
+            "GROUP BY format ORDER BY n DESC"
+        ).fetchall()
+
+        total_bytes = conn.execute(
+            "SELECT SUM(length * bitrate) / 8 FROM items WHERE length IS NOT NULL AND bitrate IS NOT NULL"
+        ).fetchone()[0] or 0
+
         return {
             "db_exists": True,
             "total_tracks": total,
-            "groupings": sorted(r[0] for r in rows),
+            "groupings": sorted(r[0] for r in grouping_rows),
+            "formats": [{"format": r["format"], "count": r["n"]} for r in format_rows],
+            "total_bytes": int(total_bytes),
         }
     finally:
         conn.close()
+
+
+def distinct_formats() -> list[str]:
+    """For the library page's format filter dropdown."""
+    if not db_exists():
+        return []
+    conn = _connect()
+    try:
+        rows = conn.execute(
+            "SELECT DISTINCT format FROM items WHERE format IS NOT NULL AND format != '' ORDER BY format"
+        ).fetchall()
+        return [r[0] for r in rows]
+    finally:
+        conn.close()

app/services/credential_service.py

@@ -182,3 +182,39 @@ def render_scope(db: Session, scope: str) -> None:
 
     else:
         raise ValueError(f"unknown credential scope: {scope}")
+
+
+def _bandcamp_cookie_expiry(cookies_txt: str) -> float | None:
+    """Parse the Netscape-format cookie jar for the 'identity' cookie's
+    expiry (Unix timestamp) -- same field pipeline-status.sh already reads
+    via `awk -F'\\t' '$6=="identity" {print $5}'`. Purely local string
+    parsing, no network call, safe to run on every dashboard load."""
+    for line in cookies_txt.splitlines():
+        fields = line.split("\t")
+        if len(fields) >= 6 and fields[5] == "identity":
+            try:
+                return float(fields[4])
+            except ValueError:
+                return None
+    return None
+
+
+def auth_states(db: Session) -> list[dict]:
+    """Per-scope credential health for the dashboard: whether it's
+    configured at all, plus (bandcamp only) cookie expiry -- computed
+    locally from the stored cookie jar, not a live network probe (unlike
+    pipeline-status.sh's Qobuz check, which hits Qobuz's API on every run;
+    deliberately not replicated here since the dashboard renders on every
+    page load)."""
+    states = []
+    for scope in SCOPE_FIELDS:
+        values = get_scope(db, scope)
+        configured = bool(values)
+        entry = {"scope": scope, "configured": configured}
+        if scope == "bandcamp" and values.get("cookies_txt"):
+            expiry = _bandcamp_cookie_expiry(values["cookies_txt"])
+            if expiry is not None:
+                entry["expires_at"] = expiry
+                entry["days_left"] = int((expiry - time.time()) / 86400)
+        states.append(entry)
+    return states

app/services/playlist_service.py

@@ -8,6 +8,39 @@ from sqlalchemy.orm import Session
 from app.models import Playlist
 from app.settings import settings
 
+
+def cron_to_time(cron_expr: str | None) -> str:
+    """'30 1 * * *' -> '01:30'. Every playlist schedule today is a simple
+    daily HH:MM (no day-of-week/day-of-month complexity -- that's only used
+    by a couple of maintenance jobs), so the UI can offer a plain time
+    picker instead of raw cron syntax. Returns '' for anything else
+    (unscheduled, or a cron expression more complex than daily-at-HH:MM)."""
+    if not cron_expr:
+        return ""
+    parts = cron_expr.split()
+    if len(parts) == 5 and parts[2] == "*" and parts[3] == "*" and parts[4] == "*":
+        try:
+            hour, minute = int(parts[1]), int(parts[0])
+            if 0 <= hour < 24 and 0 <= minute < 60:
+                return f"{hour:02d}:{minute:02d}"
+        except ValueError:
+            pass
+    return ""
+
+
+def time_to_cron(time_str: str | None) -> str | None:
+    """'01:30' -> '30 1 * * *'. Empty/invalid input means unscheduled."""
+    if not time_str:
+        return None
+    try:
+        hour_str, minute_str = time_str.split(":")
+        hour, minute = int(hour_str), int(minute_str)
+    except ValueError:
+        return None
+    if not (0 <= hour < 24 and 0 <= minute < 60):
+        return None
+    return f"{minute} {hour} * * *"
+
 # The 17-entry array this project is migrating off of (was hardcoded in
 # /opt/sldl/configs/regen.sh). Used once by seed_legacy() during Stage 1/2
 # of the migration; the DB is the source of truth from then on. cron_expr

app/services/scheduler_service.py

@@ -153,6 +153,31 @@ MAINTENANCE_JOBS: dict[str, tuple[dict, callable]] = {
     ),
 }
 
+# One short line per job for the /jobs page -- what it actually does, not
+# just its key. Kept separate from MAINTENANCE_JOBS so the schedule/function
+# mapping above stays easy to scan.
+MAINTENANCE_JOB_DESCRIPTIONS: dict[str, str] = {
+    "maintenance:sync_bandcamp": "Pull new Bandcamp purchases and import them.",
+    "maintenance:normalize_casing": "Fix artist-name casing to match your canonical list.",
+    "maintenance:dedup": "Scan for duplicate tracks (dry-run only; confirm deletes in Dedup).",
+    "maintenance:gen_djmix_playlist": "Rebuild the DJ-mix M3U from configured albums.",
+    "maintenance:gen_vgm_playlist": "Rebuild the video-game-soundtrack M3U.",
+    "maintenance:navidrome_scan": "Trigger a Navidrome library rescan (share-health gated).",
+    "maintenance:export_laptop_playlists": "Export dj-* Navidrome playlists for the laptop.",
+    "maintenance:enrich_buy_url": "Look up buy links for tracks without one yet.",
+    "maintenance:build_fingerprint_index": "Refresh the acoustic fingerprint index.",
+    "maintenance:pipeline_status_report": "Write the daily health snapshot / Telegram digest.",
+    "maintenance:log_rotation": "Delete job logs older than 30 days.",
+    "maintenance:strip_mb_tags": "Strip stale MusicBrainz tags that fragment Navidrome albums.",
+    "maintenance:strip_watermark_art": "Remove Soulseek-uploader watermark cover art.",
+    "maintenance:scrub_watermark_text": "Remove Soulseek-uploader watermark text frames.",
+    "maintenance:clean_sldl_index": "Prune stale entries from each playlist's download index.",
+    "maintenance:clear_bad_genres": "Blank malformed/junk GENRE tags before Spotify refill.",
+    "maintenance:spotify_genre": "Refill GENRE tags from Spotify per-artist data.",
+    "maintenance:beets_update_sync": "Sync the beets DB to on-disk tag changes.",
+    "maintenance:upgrade_mp3_to_flac": "Look for FLAC replacements for MP3 tracks (monthly).",
+}
+
 
 # Module-level singleton so other services (playlist_service, future
 # routers) can reach the live scheduler without main.py threading it through