""" 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() _global_conn: sqlite3.Connection | None = None def _get_conn() -> sqlite3.Connection: """Returnerer en global forbindelse i autocommit mode.""" global _global_conn if _global_conn is None: DB_PATH.parent.mkdir(parents=True, exist_ok=True) _global_conn = sqlite3.connect(str(DB_PATH), check_same_thread=False, isolation_level=None) # autocommit _global_conn.row_factory = sqlite3.Row _global_conn.execute("PRAGMA journal_mode=WAL") _global_conn.execute("PRAGMA foreign_keys=ON") return _global_conn def new_conn() -> sqlite3.Connection: """Åbn en frisk forbindelse til brug i tag_editor og dialogs.""" conn = sqlite3.connect(str(DB_PATH), check_same_thread=False, timeout=10) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=OFF") return conn @contextmanager def get_db(): """Context manager der bruger app-forbindelsen i autocommit mode. Hver statement committer med det samme — ingen eksplicit transaktion.""" conn = _get_conn() try: yield conn except Exception: raise def get_db_raw() -> sqlite3.Connection: return _get_conn() def init_db(): """Opret alle tabeller hvis de ikke findes.""" conn = _get_conn() # executescript committer automatisk og nulstiller isolation_level # Kør det direkte på den underliggende connection conn.executescript(""" 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')) ); 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, extra_tags TEXT NOT NULL DEFAULT '{}', api_song_id TEXT, last_synced_at TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS dance_levels ( id INTEGER PRIMARY KEY AUTOINCREMENT, sort_order INTEGER NOT NULL, name TEXT NOT NULL UNIQUE, description TEXT NOT NULL DEFAULT '', synced_at TEXT ); -- Dans-entitet: navn + niveau er unik kombination CREATE TABLE IF NOT EXISTS dances ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL COLLATE NOCASE, level_id INTEGER REFERENCES dance_levels(id), use_count INTEGER NOT NULL DEFAULT 1, source TEXT NOT NULL DEFAULT 'local', synced_at TEXT, UNIQUE(name, level_id) ); -- Hoveddanse på en sang CREATE TABLE IF NOT EXISTS song_dances ( id INTEGER PRIMARY KEY AUTOINCREMENT, song_id TEXT NOT NULL REFERENCES songs(id) ON DELETE CASCADE, dance_id INTEGER NOT NULL REFERENCES dances(id), dance_order INTEGER NOT NULL DEFAULT 1, UNIQUE(song_id, dance_id) ); -- Alternativ-danse på en sang CREATE TABLE IF NOT EXISTS song_alt_dances ( id INTEGER PRIMARY KEY AUTOINCREMENT, song_id TEXT NOT NULL REFERENCES songs(id) ON DELETE CASCADE, dance_id INTEGER NOT NULL REFERENCES dances(id), note TEXT NOT NULL DEFAULT '', source TEXT NOT NULL DEFAULT 'local', created_at TEXT NOT NULL DEFAULT (datetime('now')), UNIQUE(song_id, dance_id) ); CREATE TABLE IF NOT EXISTS playlists ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', api_project_id TEXT, last_synced_at TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')) ); 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', UNIQUE(playlist_id, position) ); CREATE TABLE IF NOT EXISTS sync_queue ( id INTEGER PRIMARY KEY AUTOINCREMENT, entity_type TEXT NOT NULL, entity_id TEXT NOT NULL, action TEXT NOT NULL, payload TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS event_state ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); 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); CREATE INDEX IF NOT EXISTS idx_song_alt_dances ON song_alt_dances(song_id); CREATE INDEX IF NOT EXISTS idx_dances_name ON dances(name); """) # executescript slår foreign_keys fra — genaktiver conn.execute("PRAGMA foreign_keys=ON") # Tilføj db_version tabel hvis den ikke findes conn.execute(""" CREATE TABLE IF NOT EXISTS db_version ( version INTEGER PRIMARY KEY ) """) # Kør versionsbaserede migrationer _run_versioned_migrations(conn) # Seed standard-niveauer count = conn.execute("SELECT COUNT(*) FROM dance_levels").fetchone()[0] if count == 0: defaults = [ (10, "Absolute Beginner", "Ingen tidligere danse-erfaring kræves"), (20, "Beginner", "Lidt tidligere erfaring"), (30, "High Beginner", "God begynder, klar til mere"), (40, "Low Improver", "Begyndende øvet"), (50, "Improver", "Grundlæggende færdigheder på plads"), (60, "High Improver", "Stærk øvet, næsten intermediate"), (70, "Low Intermediate", "Begyndende intermediate"), (80, "Intermediate", "Erfaren danser"), (90, "High Intermediate", "Stærk intermediate"), (99, "Advanced", "Fuld beherskelse af trin og teknik"), ] for row in defaults: conn.execute( "INSERT OR IGNORE INTO dance_levels (sort_order, name, description) VALUES (?,?,?)", row ) # ── Versionsbaserede migrationer ────────────────────────────────────────────── # Tilføj aldrig gamle — tilføj kun nye versioner nederst. MIGRATIONS: dict[int, list[str]] = { 1: [ "ALTER TABLE songs ADD COLUMN extra_tags TEXT NOT NULL DEFAULT '{}'", ], 2: [ # Ny dans-entitet model """CREATE TABLE IF NOT EXISTS dances ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL COLLATE NOCASE, level_id INTEGER REFERENCES dance_levels(id), use_count INTEGER NOT NULL DEFAULT 1, source TEXT NOT NULL DEFAULT 'local', synced_at TEXT, UNIQUE(name, level_id) )""", """CREATE TABLE IF NOT EXISTS song_alt_dances ( id INTEGER PRIMARY KEY AUTOINCREMENT, song_id TEXT NOT NULL REFERENCES songs(id) ON DELETE CASCADE, dance_id INTEGER NOT NULL REFERENCES dances(id), note TEXT NOT NULL DEFAULT '', source TEXT NOT NULL DEFAULT 'local', created_at TEXT NOT NULL DEFAULT (datetime('now')), UNIQUE(song_id, dance_id) )""", # Migrer eksisterende song_dances data til ny model # (kører kun på ældre databaser der har dance_name kolonnen) """INSERT OR IGNORE INTO dances (name, level_id, source) SELECT DISTINCT dance_name, level_id, 'local' FROM song_dances WHERE dance_name IS NOT NULL AND dance_name != ''""", ], 3: [ "ALTER TABLE playlists ADD COLUMN tags TEXT NOT NULL DEFAULT ''", ], 4: [ "ALTER TABLE dances ADD COLUMN choreographer TEXT NOT NULL DEFAULT ''", "ALTER TABLE dances ADD COLUMN video_url TEXT NOT NULL DEFAULT ''", "ALTER TABLE dances ADD COLUMN stepsheet_url TEXT NOT NULL DEFAULT ''", "ALTER TABLE dances ADD COLUMN notes TEXT NOT NULL DEFAULT ''", ], 5: [ # Workshop-markering på sang+dans kombination (ikke dans alene) """ALTER TABLE song_dances ADD COLUMN is_workshop INTEGER NOT NULL DEFAULT 0""", """ALTER TABLE song_alt_dances ADD COLUMN is_workshop INTEGER NOT NULL DEFAULT 0""", ], 6: [ # Workshop og dans-valg på selve playlist-sangen """ALTER TABLE playlist_songs ADD COLUMN is_workshop INTEGER NOT NULL DEFAULT 0""", """ALTER TABLE playlist_songs ADD COLUMN dance_override TEXT NOT NULL DEFAULT ''""", ], 7: [ # Linkede server-playlister """ALTER TABLE playlists ADD COLUMN is_linked INTEGER NOT NULL DEFAULT 0""", """ALTER TABLE playlists ADD COLUMN server_permission TEXT NOT NULL DEFAULT 'view'""", ], 8: [ # MusicBrainz og AcoustID matching """ALTER TABLE songs ADD COLUMN mbid TEXT""", """ALTER TABLE songs ADD COLUMN acoustid TEXT""", ], 9: [ # Opdater niveau-navne til korrekte betegnelser i rigtig rækkefølge "DELETE FROM dance_levels", "INSERT INTO dance_levels (sort_order, name, description) VALUES (10, 'Absolute Beginner', 'Ingen tidligere danse-erfaring kræves')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (20, 'Beginner', 'Lidt tidligere erfaring')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (30, 'High Beginner', 'God begynder, klar til mere')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (40, 'Low Improver', 'Begyndende øvet')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (50, 'Improver', 'Grundlæggende færdigheder på plads')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (60, 'High Improver', 'Stærk øvet, næsten intermediate')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (70, 'Low Intermediate', 'Begyndende intermediate')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (80, 'Intermediate', 'Erfaren danser')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (90, 'High Intermediate', 'Stærk intermediate')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (99, 'Advanced', 'Fuld beherskelse af trin og teknik')", ], 10: [ # Ret stavefejl i eksisterende data "UPDATE dance_levels SET name='Low Intermediate' WHERE name='Low Intermidiate' OR name='Low Intermidiat'", "UPDATE dance_levels SET name='Intermediate' WHERE name='Intermidiate' OR name='Intermidate'", "UPDATE dance_levels SET name='High Intermediate' WHERE name='High Intermidiate' OR name='High Intermidiat'", ], 11: [ # Genopret dance_levels med korrekte navne og rækkefølge "DELETE FROM dance_levels", "INSERT INTO dance_levels (sort_order, name, description) VALUES (10, 'Absolute Beginner', 'Ingen tidligere danse-erfaring kræves')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (20, 'Beginner', 'Lidt tidligere erfaring')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (30, 'High Beginner', 'God begynder, klar til mere')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (40, 'Low Improver', 'Begyndende øvet')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (50, 'Improver', 'Grundlæggende færdigheder på plads')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (60, 'High Improver', 'Stærk øvet, næsten intermediate')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (70, 'Low Intermediate', 'Begyndende intermediate')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (80, 'Intermediate', 'Erfaren danser')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (90, 'High Intermediate', 'Stærk intermediate')", "INSERT INTO dance_levels (sort_order, name, description) VALUES (99, 'Advanced', 'Fuld beherskelse af trin og teknik')", ], } def _run_versioned_migrations(conn): """Kør kun migrationer der ikke allerede er kørt vha. db_version tabel.""" row = conn.execute("SELECT version FROM db_version").fetchone() current_version = row["version"] if row else 0 for version in sorted(MIGRATIONS.keys()): if version <= current_version: continue for sql in MIGRATIONS[version]: try: conn.execute(sql) except Exception: pass # kolonnen eksisterer allerede conn.execute( "INSERT OR REPLACE INTO db_version (version) VALUES (?)", (version,) ) # ── 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: # Marker sange som manglende og løsriv dem fra biblioteket conn.execute( "UPDATE songs SET file_missing=1, library_id=NULL WHERE library_id=?", (library_id,) ) # Nu kan biblioteket slettes uden FK-konflikt conn.execute("DELETE FROM libraries 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, json with get_db() as conn: existing = conn.execute( "SELECT id FROM songs WHERE local_path=?", (song_data["local_path"],) ).fetchone() extra_tags_json = json.dumps(song_data.get("extra_tags", {}), ensure_ascii=False) if existing: song_id = existing["id"] conn.execute(""" UPDATE songs SET library_id=?, title=?, artist=?, album=?, bpm=?, duration_sec=?, file_format=?, file_modified_at=?, file_missing=0, extra_tags=? WHERE id=? """, ( song_data.get("library_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", ""), extra_tags_json, 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, extra_tags) 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", ""), extra_tags_json, )) # Opdater danse hvis de er med i data — bevar eksisterende og merge if "dances" in song_data: file_dances = [] for dance in song_data["dances"]: name = dance.get("name", dance) if isinstance(dance, dict) else dance if name: file_dances.append(name.strip()) # Find eksisterende song_dances via dances tabel existing = conn.execute(""" SELECT sd.id, d.name, sd.dance_order, d.level_id, d.id as dance_id FROM song_dances sd JOIN dances d ON d.id = sd.dance_id WHERE sd.song_id=? ORDER BY sd.dance_order """, (song_id,)).fetchall() existing_map = {r["name"].lower(): r for r in existing} file_lower = [d.lower() for d in file_dances] # Slet danse der ikke længere er i filen for row in existing: if row["name"].lower() not in file_lower: conn.execute("DELETE FROM song_dances WHERE id=?", (row["id"],)) # Tilføj eller opdater danse fra filen for i, name in enumerate(file_dances, start=1): ex = existing_map.get(name.lower()) if ex: conn.execute( "UPDATE song_dances SET dance_order=? WHERE id=?", (i, ex["id"]) ) else: # Opret eller find dans (name + NULL level = ny dans uden niveau) dance_id = get_or_create_dance(name, None, conn) conn.execute( "INSERT OR IGNORE INTO song_dances (song_id, dance_id, dance_order) " "VALUES (?,?,?)", (song_id, dance_id, 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, album, dans, koreograf, niveau og øvrige tags.""" import logging as _log _log.getLogger(__name__).info(f"search_songs: '{query}'") pattern = f"%{query}%" with get_db() as conn: rows = conn.execute(""" SELECT DISTINCT s.* FROM songs s LEFT JOIN song_dances sd ON sd.song_id = s.id LEFT JOIN dances d ON d.id = sd.dance_id LEFT JOIN dance_levels dl ON dl.id = d.level_id WHERE s.file_missing = 0 AND ( s.title LIKE ? OR s.artist LIKE ? OR s.album LIKE ? OR d.name LIKE ? OR d.choreographer LIKE ? OR dl.name LIKE ? OR s.extra_tags LIKE ? ) ORDER BY s.artist, s.title LIMIT ? """, (pattern,)*7 + (limit,)).fetchall() _log.getLogger(__name__).info(f"search_songs: '{query}' → {len(rows)} resultater") return rows 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 = "", tags: str = "") -> int: with get_db() as conn: cur = conn.execute( "INSERT INTO playlists (name, description, tags) VALUES (?,?,?)", (name, description, tags) ) return cur.lastrowid def create_linked_playlist(name: str, api_project_id: str, permission: str = "view", description: str = "", tags: str = "") -> int: """Opret en playliste der er linket til en server-playliste.""" with get_db() as conn: cur = conn.execute( """INSERT INTO playlists (name, description, tags, api_project_id, is_linked, server_permission) VALUES (?,?,?,?,1,?)""", (name, description, tags, api_project_id, permission) ) return cur.lastrowid def update_playlist_tags(playlist_id: int, tags: str): with get_db() as conn: conn.execute( "UPDATE playlists SET tags=? WHERE id=?", (tags, playlist_id) ) def get_all_playlist_tags() -> list[str]: """Returnerer alle unikke tags på tværs af alle playlists, sorteret alfabetisk.""" with get_db() as conn: rows = conn.execute( "SELECT tags FROM playlists WHERE tags != '' AND name != ?", ("__aktiv__",) ).fetchall() tags = set() for row in rows: for tag in row["tags"].split(","): t = tag.strip().lower() if t: tags.add(t) return sorted(tags) def get_playlists(tag_filter: str | None = None) -> list[sqlite3.Row]: """Hent alle navngivne playlists med sang-antal. Filtrer på tag hvis angivet.""" with get_db() as conn: if tag_filter: rows = conn.execute(""" SELECT p.*, COUNT(ps.id) as song_count FROM playlists p LEFT JOIN playlist_songs ps ON ps.playlist_id = p.id WHERE p.name != ? AND ( p.tags LIKE ? OR p.tags LIKE ? OR p.tags LIKE ? OR p.tags = ? ) GROUP BY p.id ORDER BY p.created_at DESC """, ( "__aktiv__", f"{tag_filter},%", f"%, {tag_filter},%", f"%, {tag_filter}", tag_filter, )).fetchall() else: rows = conn.execute(""" SELECT p.*, COUNT(ps.id) as song_count FROM playlists p LEFT JOIN playlist_songs ps ON ps.playlist_id = p.id WHERE p.name != ? GROUP BY p.id ORDER BY p.created_at DESC """, ("__aktiv__",)).fetchall() return rows 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]} # ── Event-state (gemmes løbende så man kan genstarte efter strømsvigt) ──────── def save_event_state(current_idx: int, statuses: list[str]): """Gem event-fremgang — overskrives ved hver ændring.""" import json with get_db() as conn: conn.execute("INSERT OR REPLACE INTO event_state (key,value) VALUES ('current_idx',?)", (str(current_idx),)) conn.execute("INSERT OR REPLACE INTO event_state (key,value) VALUES ('statuses',?)", (json.dumps(statuses),)) def load_event_state() -> tuple[int, list[str]] | None: """Indlæs gemt event-fremgang. Returnerer None hvis ingen gemt tilstand.""" import json with get_db() as conn: idx_row = conn.execute( "SELECT value FROM event_state WHERE key='current_idx'" ).fetchone() sta_row = conn.execute( "SELECT value FROM event_state WHERE key='statuses'" ).fetchone() if not idx_row or not sta_row: return None return int(idx_row["value"]), json.loads(sta_row["value"]) def clear_event_state(): """Nulstil gemt event-tilstand (bruges ved 'Start event').""" with get_db() as conn: conn.execute("DELETE FROM event_state") # ── Dans-navne ordbog ───────────────────────────────────────────────────────── # ── Dans-entitet funktioner ─────────────────────────────────────────────────── def get_dance(dance_id: int) -> sqlite3.Row | None: with get_db() as conn: return conn.execute( "SELECT * FROM dances WHERE id=?", (dance_id,) ).fetchone() def update_dance_info(dance_id: int, choreographer: str = "", video_url: str = "", stepsheet_url: str = "", notes: str = ""): with get_db() as conn: conn.execute(""" UPDATE dances SET choreographer = ?, video_url = ?, stepsheet_url = ?, notes = ? WHERE id = ? """, (choreographer.strip(), video_url.strip(), stepsheet_url.strip(), notes.strip(), dance_id)) def get_or_create_dance(name: str, level_id: int | None, conn=None, choreographer: str = "") -> int: """Find eller opret en dans (name + level_id kombination). Returnerer dance_id. conn er valgfri — bruges ved nested kald.""" name = name.strip() choreo = choreographer.strip() close = False if conn is None: conn = new_conn() close = True try: existing = conn.execute( "SELECT id FROM dances WHERE name=? COLLATE NOCASE AND level_id IS ?", (name, level_id) ).fetchone() if existing: conn.execute( "UPDATE dances SET use_count=use_count+1" + (", choreographer=?" if choreo else "") + " WHERE id=?", ((choreo, existing["id"]) if choreo else (existing["id"],)) ) return existing["id"] conn.execute( "INSERT INTO dances (name, level_id, choreographer, use_count, source) VALUES (?,?,?,1,'local')", (name, level_id, choreo) ) return conn.execute( "SELECT id FROM dances WHERE name=? COLLATE NOCASE AND level_id IS ?", (name, level_id) ).fetchone()["id"] finally: if close: conn.commit() conn.close() def get_choreographer_suggestions(prefix: str, limit: int = 20) -> list[str]: """Returnerer koreografer der starter med prefix, sorteret alfabetisk.""" with get_db() as conn: rows = conn.execute(""" SELECT DISTINCT choreographer FROM dances WHERE choreographer LIKE ? COLLATE NOCASE AND choreographer != '' ORDER BY choreographer LIMIT ? """, (f"{prefix}%", limit)).fetchall() return [r["choreographer"] for r in rows] def get_dance_suggestions(prefix: str, limit: int = 20) -> list[dict]: """Returnerer danse der matcher prefix i navn ELLER koreograf. Sorteret efter popularitet — bruges til autoudfyld.""" with get_db() as conn: rows = conn.execute(""" SELECT d.id, d.name, d.level_id, d.use_count, d.choreographer, dl.name as level_name, dl.sort_order FROM dances d LEFT JOIN dance_levels dl ON dl.id = d.level_id WHERE d.name LIKE ? COLLATE NOCASE OR d.choreographer LIKE ? COLLATE NOCASE ORDER BY d.use_count DESC, dl.sort_order, d.name LIMIT ? """, (f"%{prefix}%", f"%{prefix}%", limit)).fetchall() return [dict(r) for r in rows] def get_dances_for_song(song_id: str) -> list[dict]: """Hent hoveddanse for en sang med niveau-info og workshop-flag.""" with get_db() as conn: rows = conn.execute(""" SELECT d.id as dance_id, d.name, d.level_id, d.choreographer, dl.name as level_name, sd.dance_order, sd.id as song_dance_id, sd.is_workshop FROM song_dances sd JOIN dances d ON d.id = sd.dance_id LEFT JOIN dance_levels dl ON dl.id = d.level_id WHERE sd.song_id=? ORDER BY sd.dance_order """, (song_id,)).fetchall() return [dict(r) for r in rows] def get_alt_dances_for_song(song_id: str) -> list[dict]: """Hent alternativ-danse for en sang med niveau-info.""" with get_db() as conn: rows = conn.execute(""" SELECT d.id as dance_id, d.name, d.level_id, dl.name as level_name, sad.note, sad.source, sad.id as alt_id FROM song_alt_dances sad JOIN dances d ON d.id = sad.dance_id LEFT JOIN dance_levels dl ON dl.id = d.level_id WHERE sad.song_id=? ORDER BY d.name """, (song_id,)).fetchall() return [dict(r) for r in rows] # ── Dans-niveauer ───────────────────────────────────────────────────────────── def get_dance_levels() -> list[sqlite3.Row]: """Hent alle niveauer sorteret efter sort_order.""" with get_db() as conn: return conn.execute( "SELECT * FROM dance_levels ORDER BY sort_order" ).fetchall() def sync_dance_levels_from_api(levels: list[dict]): """Synkroniser niveauer fra API.""" from datetime import datetime, timezone now = datetime.now(timezone.utc).isoformat() with get_db() as conn: for lvl in levels: conn.execute(""" INSERT INTO dance_levels (sort_order, name, description, synced_at) VALUES (?, ?, ?, ?) ON CONFLICT(name) DO UPDATE SET sort_order = excluded.sort_order, description = excluded.description, synced_at = excluded.synced_at """, (lvl["sort_order"], lvl["name"], lvl.get("description", ""), now)) def sync_dances_from_api(dances: list[dict]): """Synkroniser danse fra API — {name, level_id, use_count}.""" from datetime import datetime, timezone now = datetime.now(timezone.utc).isoformat() with get_db() as conn: for d in dances: conn.execute(""" INSERT INTO dances (name, level_id, use_count, source, synced_at) VALUES (?, ?, ?, 'community', ?) ON CONFLICT(name, level_id) DO UPDATE SET use_count = MAX(use_count, excluded.use_count), synced_at = excluded.synced_at """, (d["name"], d.get("level_id"), d.get("use_count", 1), now)) # Backwards compat alias def get_dance_name_suggestions(prefix: str, limit: int = 20) -> list[str]: """Returnerer dans-navne som strings — bruges af AutoLineEdit.""" suggestions = get_dance_suggestions(prefix, limit) result = [] for s in suggestions: if s.get("level_name"): result.append(f"{s['name']} / {s['level_name']}") else: result.append(s["name"]) return result