331 lines
12 KiB
Python
331 lines
12 KiB
Python
"""
|
|
local_db.py — Lokal SQLite database til offline brug.
|
|
|
|
Håndterer:
|
|
- Musikbiblioteker (stier der overvåges)
|
|
- Sange høstet fra filsystemet
|
|
- Lokale afspilningslister (offline-projekter)
|
|
- Synkroniseringsstatus mod API
|
|
"""
|
|
|
|
import sqlite3
|
|
import threading
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
DB_PATH = Path.home() / ".linedance" / "local.db"
|
|
|
|
_local = threading.local()
|
|
|
|
|
|
def _get_conn() -> sqlite3.Connection:
|
|
"""Returnerer en thread-lokal forbindelse."""
|
|
if not hasattr(_local, "conn") or _local.conn is None:
|
|
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA journal_mode=WAL") # bedre concurrent adgang
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
_local.conn = conn
|
|
return _local.conn
|
|
|
|
|
|
@contextmanager
|
|
def get_db():
|
|
conn = _get_conn()
|
|
try:
|
|
yield conn
|
|
conn.commit()
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
|
|
|
|
def init_db():
|
|
"""Opret alle tabeller hvis de ikke findes."""
|
|
with get_db() as conn:
|
|
conn.executescript("""
|
|
-- Musikbiblioteker der overvåges
|
|
CREATE TABLE IF NOT EXISTS libraries (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
path TEXT NOT NULL UNIQUE,
|
|
is_active INTEGER NOT NULL DEFAULT 1,
|
|
last_full_scan TEXT,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
|
);
|
|
|
|
-- Sange høstet fra filsystemet
|
|
CREATE TABLE IF NOT EXISTS songs (
|
|
id TEXT PRIMARY KEY,
|
|
library_id INTEGER REFERENCES libraries(id),
|
|
local_path TEXT NOT NULL UNIQUE,
|
|
title TEXT NOT NULL DEFAULT '',
|
|
artist TEXT NOT NULL DEFAULT '',
|
|
album TEXT NOT NULL DEFAULT '',
|
|
bpm INTEGER NOT NULL DEFAULT 0,
|
|
duration_sec INTEGER NOT NULL DEFAULT 0,
|
|
file_format TEXT NOT NULL DEFAULT '',
|
|
file_modified_at TEXT NOT NULL,
|
|
file_missing INTEGER NOT NULL DEFAULT 0,
|
|
api_song_id TEXT, -- NULL hvis ikke synkroniseret
|
|
last_synced_at TEXT,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
|
);
|
|
|
|
-- Danse knyttet til en sang (kun MP3 kan skrive tags)
|
|
CREATE TABLE IF NOT EXISTS song_dances (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
song_id TEXT NOT NULL REFERENCES songs(id) ON DELETE CASCADE,
|
|
dance_name TEXT NOT NULL,
|
|
dance_order INTEGER NOT NULL DEFAULT 1
|
|
);
|
|
|
|
-- Lokale afspilningslister (offline-projekter)
|
|
CREATE TABLE IF NOT EXISTS playlists (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
description TEXT NOT NULL DEFAULT '',
|
|
api_project_id TEXT, -- NULL hvis ikke synkroniseret
|
|
last_synced_at TEXT,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
|
);
|
|
|
|
-- Sange i en afspilningsliste
|
|
CREATE TABLE IF NOT EXISTS playlist_songs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
playlist_id INTEGER NOT NULL REFERENCES playlists(id) ON DELETE CASCADE,
|
|
song_id TEXT NOT NULL REFERENCES songs(id),
|
|
position INTEGER NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'pending', -- pending|playing|played|skipped
|
|
UNIQUE(playlist_id, position)
|
|
);
|
|
|
|
-- Synkroniseringskø — ændringer der venter på at komme online
|
|
CREATE TABLE IF NOT EXISTS sync_queue (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
entity_type TEXT NOT NULL, -- 'song'|'playlist'|'playlist_song'
|
|
entity_id TEXT NOT NULL,
|
|
action TEXT NOT NULL, -- 'create'|'update'|'delete'
|
|
payload TEXT NOT NULL, -- JSON
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
|
);
|
|
|
|
-- Indekser til hurtig søgning
|
|
CREATE INDEX IF NOT EXISTS idx_songs_title ON songs(title);
|
|
CREATE INDEX IF NOT EXISTS idx_songs_artist ON songs(artist);
|
|
CREATE INDEX IF NOT EXISTS idx_songs_missing ON songs(file_missing);
|
|
CREATE INDEX IF NOT EXISTS idx_songs_library ON songs(library_id);
|
|
CREATE INDEX IF NOT EXISTS idx_song_dances ON song_dances(song_id);
|
|
""")
|
|
|
|
|
|
# ── Biblioteker ───────────────────────────────────────────────────────────────
|
|
|
|
def add_library(path: str) -> int:
|
|
with get_db() as conn:
|
|
cur = conn.execute(
|
|
"INSERT OR IGNORE INTO libraries (path) VALUES (?)", (path,)
|
|
)
|
|
if cur.lastrowid:
|
|
return cur.lastrowid
|
|
row = conn.execute("SELECT id FROM libraries WHERE path=?", (path,)).fetchone()
|
|
return row["id"]
|
|
|
|
|
|
def get_libraries(active_only: bool = True) -> list[sqlite3.Row]:
|
|
with get_db() as conn:
|
|
if active_only:
|
|
return conn.execute(
|
|
"SELECT * FROM libraries WHERE is_active=1 ORDER BY path"
|
|
).fetchall()
|
|
return conn.execute("SELECT * FROM libraries ORDER BY path").fetchall()
|
|
|
|
|
|
def remove_library(library_id: int):
|
|
with get_db() as conn:
|
|
conn.execute("UPDATE libraries SET is_active=0 WHERE id=?", (library_id,))
|
|
|
|
|
|
def update_library_scan_time(library_id: int):
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"UPDATE libraries SET last_full_scan=? WHERE id=?", (now, library_id)
|
|
)
|
|
|
|
|
|
# ── Sange ─────────────────────────────────────────────────────────────────────
|
|
|
|
def upsert_song(song_data: dict) -> str:
|
|
"""
|
|
Indsæt eller opdater en sang baseret på local_path.
|
|
Returnerer song_id.
|
|
"""
|
|
import uuid
|
|
with get_db() as conn:
|
|
existing = conn.execute(
|
|
"SELECT id FROM songs WHERE local_path=?", (song_data["local_path"],)
|
|
).fetchone()
|
|
|
|
if existing:
|
|
song_id = existing["id"]
|
|
conn.execute("""
|
|
UPDATE songs SET
|
|
title=?, artist=?, album=?, bpm=?, duration_sec=?,
|
|
file_format=?, file_modified_at=?, file_missing=0
|
|
WHERE id=?
|
|
""", (
|
|
song_data.get("title", ""),
|
|
song_data.get("artist", ""),
|
|
song_data.get("album", ""),
|
|
song_data.get("bpm", 0),
|
|
song_data.get("duration_sec", 0),
|
|
song_data.get("file_format", ""),
|
|
song_data.get("file_modified_at", ""),
|
|
song_id,
|
|
))
|
|
else:
|
|
song_id = str(uuid.uuid4())
|
|
conn.execute("""
|
|
INSERT INTO songs
|
|
(id, library_id, local_path, title, artist, album,
|
|
bpm, duration_sec, file_format, file_modified_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?)
|
|
""", (
|
|
song_id,
|
|
song_data.get("library_id"),
|
|
song_data["local_path"],
|
|
song_data.get("title", ""),
|
|
song_data.get("artist", ""),
|
|
song_data.get("album", ""),
|
|
song_data.get("bpm", 0),
|
|
song_data.get("duration_sec", 0),
|
|
song_data.get("file_format", ""),
|
|
song_data.get("file_modified_at", ""),
|
|
))
|
|
|
|
# Opdater danse hvis de er med i data
|
|
if "dances" in song_data:
|
|
conn.execute("DELETE FROM song_dances WHERE song_id=?", (song_id,))
|
|
for i, dance_name in enumerate(song_data["dances"], start=1):
|
|
conn.execute(
|
|
"INSERT INTO song_dances (song_id, dance_name, dance_order) VALUES (?,?,?)",
|
|
(song_id, dance_name, i),
|
|
)
|
|
|
|
return song_id
|
|
|
|
|
|
def mark_song_missing(local_path: str):
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"UPDATE songs SET file_missing=1 WHERE local_path=?", (local_path,)
|
|
)
|
|
|
|
|
|
def get_song_by_path(local_path: str) -> sqlite3.Row | None:
|
|
with get_db() as conn:
|
|
return conn.execute(
|
|
"SELECT * FROM songs WHERE local_path=?", (local_path,)
|
|
).fetchone()
|
|
|
|
|
|
def search_songs(query: str, limit: int = 50) -> list[sqlite3.Row]:
|
|
"""Søg i titel, artist og dansenavne."""
|
|
pattern = f"%{query}%"
|
|
with get_db() as conn:
|
|
return conn.execute("""
|
|
SELECT DISTINCT s.* FROM songs s
|
|
LEFT JOIN song_dances sd ON sd.song_id = s.id
|
|
WHERE s.file_missing = 0
|
|
AND (s.title LIKE ? OR s.artist LIKE ? OR s.album LIKE ? OR sd.dance_name LIKE ?)
|
|
ORDER BY s.artist, s.title
|
|
LIMIT ?
|
|
""", (pattern, pattern, pattern, pattern, limit)).fetchall()
|
|
|
|
|
|
def get_songs_for_library(library_id: int) -> list[sqlite3.Row]:
|
|
with get_db() as conn:
|
|
return conn.execute(
|
|
"SELECT * FROM songs WHERE library_id=? ORDER BY artist, title",
|
|
(library_id,)
|
|
).fetchall()
|
|
|
|
|
|
def get_all_song_paths_for_library(library_id: int) -> dict[str, str]:
|
|
"""Returnerer {local_path: file_modified_at} — bruges til fuld scan."""
|
|
with get_db() as conn:
|
|
rows = conn.execute(
|
|
"SELECT local_path, file_modified_at FROM songs WHERE library_id=?",
|
|
(library_id,)
|
|
).fetchall()
|
|
return {row["local_path"]: row["file_modified_at"] for row in rows}
|
|
|
|
|
|
# ── Afspilningslister ─────────────────────────────────────────────────────────
|
|
|
|
def create_playlist(name: str, description: str = "") -> int:
|
|
with get_db() as conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO playlists (name, description) VALUES (?,?)",
|
|
(name, description)
|
|
)
|
|
return cur.lastrowid
|
|
|
|
|
|
def get_playlists() -> list[sqlite3.Row]:
|
|
with get_db() as conn:
|
|
return conn.execute(
|
|
"SELECT * FROM playlists ORDER BY created_at DESC"
|
|
).fetchall()
|
|
|
|
|
|
def add_song_to_playlist(playlist_id: int, song_id: str, position: int | None = None) -> int:
|
|
with get_db() as conn:
|
|
if position is None:
|
|
row = conn.execute(
|
|
"SELECT MAX(position) as max_pos FROM playlist_songs WHERE playlist_id=?",
|
|
(playlist_id,)
|
|
).fetchone()
|
|
position = (row["max_pos"] or 0) + 1
|
|
|
|
cur = conn.execute(
|
|
"INSERT INTO playlist_songs (playlist_id, song_id, position) VALUES (?,?,?)",
|
|
(playlist_id, song_id, position)
|
|
)
|
|
return cur.lastrowid
|
|
|
|
|
|
def update_playlist_song_status(playlist_song_id: int, status: str):
|
|
valid = {"pending", "playing", "played", "skipped"}
|
|
if status not in valid:
|
|
raise ValueError(f"Ugyldig status: {status}")
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"UPDATE playlist_songs SET status=? WHERE id=?",
|
|
(status, playlist_song_id)
|
|
)
|
|
|
|
|
|
def get_playlist_with_songs(playlist_id: int) -> dict:
|
|
with get_db() as conn:
|
|
playlist = conn.execute(
|
|
"SELECT * FROM playlists WHERE id=?", (playlist_id,)
|
|
).fetchone()
|
|
if not playlist:
|
|
return {}
|
|
|
|
songs = conn.execute("""
|
|
SELECT ps.id as ps_id, ps.position, ps.status,
|
|
s.*, GROUP_CONCAT(sd.dance_name ORDER BY sd.dance_order) as dances
|
|
FROM playlist_songs ps
|
|
JOIN songs s ON s.id = ps.song_id
|
|
LEFT JOIN song_dances sd ON sd.song_id = s.id
|
|
WHERE ps.playlist_id = ?
|
|
GROUP BY ps.id
|
|
ORDER BY ps.position
|
|
""", (playlist_id,)).fetchall()
|
|
|
|
return {"playlist": dict(playlist), "songs": [dict(s) for s in songs]}
|