This commit is contained in:
2026-04-14 16:21:06 +02:00
parent 287477753e
commit 460b41a8c5
5 changed files with 301 additions and 1 deletions

View File

@@ -31,6 +31,7 @@ class SongData(BaseModel):
duration_sec: int = 0
file_format: str = ""
mbid: str = ""
acoustid: str = ""
class DanceData(BaseModel):
name: str
@@ -115,6 +116,8 @@ def push(
existing.bpm = s.bpm
if s.mbid and not existing.mbid:
existing.mbid = s.mbid
if s.acoustid and not existing.acoustid:
existing.acoustid = s.acoustid
else:
song = Song(
owner_id=me.id,
@@ -122,6 +125,7 @@ def push(
bpm=s.bpm, duration_sec=s.duration_sec,
file_format=s.file_format,
mbid=s.mbid or None,
acoustid=s.acoustid or None,
)
db.add(song)
db.flush()

View File

@@ -0,0 +1,240 @@
"""
acoustid_worker.py — Baggrunds-fingerprinting via AcoustID/Chromaprint.
Finder sange uden MBID, kører fpcalc på dem, spørger AcoustID API
og gemmer MBID + acoustid lokalt. Kører som lavprioritets baggrundstråd.
Kræver fpcalc (Chromaprint) installeret:
Linux: sudo apt install libchromaprint-tools
Windows: download fra https://acoustid.org/chromaprint
"""
import json
import logging
import subprocess
import threading
import time
from pathlib import Path
logger = logging.getLogger(__name__)
# AcoustID API nøgle — gratis til open-source apps
# https://acoustid.org/api-key
ACOUSTID_API_KEY = "8XaBELgH" # Offentlig test-nøgle, skift til din egen ved produktion
ACOUSTID_API_URL = "https://api.acoustid.org/v2/lookup"
# Pause mellem API-kald (AcoustID tillader 3/sek)
API_DELAY = 0.4
def find_fpcalc() -> str | None:
"""Find fpcalc på systemet."""
import shutil, sys
# Tjek PATH
path = shutil.which("fpcalc")
if path:
return path
# Windows: tjek ved siden af exe
if sys.platform == "win32":
candidates = [
Path(sys.executable).parent / "fpcalc.exe",
Path("C:/Program Files/Chromaprint/fpcalc.exe"),
Path("C:/Program Files (x86)/Chromaprint/fpcalc.exe"),
]
for c in candidates:
if c.exists():
return str(c)
return None
def fingerprint_file(path: str, fpcalc: str) -> tuple[str, int] | None:
"""
Kør fpcalc på en fil og returnér (fingerprint, duration).
Returnerer None ved fejl.
"""
try:
result = subprocess.run(
[fpcalc, "-json", path],
capture_output=True, text=True, timeout=30
)
if result.returncode != 0:
return None
data = json.loads(result.stdout)
fp = data.get("fingerprint", "")
dur = int(data.get("duration", 0))
if fp and dur:
return fp, dur
except Exception as e:
logger.warning(f"fpcalc fejl for {path}: {e}")
return None
def lookup_acoustid(fingerprint: str, duration: int) -> dict | None:
"""
Spørg AcoustID API om MBID for et fingerprint.
Returnerer dict med 'mbid' og 'acoustid' eller None.
"""
try:
import urllib.request, urllib.parse
params = urllib.parse.urlencode({
"client": ACOUSTID_API_KEY,
"fingerprint": fingerprint,
"duration": duration,
"meta": "recordings",
})
url = f"{ACOUSTID_API_URL}?{params}"
req = urllib.request.Request(url)
req.add_header("User-Agent", "LineDancePlayer/1.0")
with urllib.request.urlopen(req, timeout=10) as resp:
data = json.loads(resp.read())
if data.get("status") != "ok":
return None
results = data.get("results", [])
if not results:
return None
# Tag det bedste resultat (højeste score)
best = max(results, key=lambda r: r.get("score", 0))
if best.get("score", 0) < 0.85:
return None # For usikkert
acoustid = best.get("id", "")
recordings = best.get("recordings", [])
mbid = recordings[0].get("id", "") if recordings else ""
return {"acoustid": acoustid, "mbid": mbid} if acoustid else None
except Exception as e:
logger.warning(f"AcoustID API fejl: {e}")
return None
def run_acoustid_scan(db_path: str, on_progress=None, stop_event: threading.Event = None):
"""
Gennemgå alle sange uden MBID og forsøg AcoustID fingerprinting.
Kører som baggrundsjob.
"""
import sqlite3
fpcalc = find_fpcalc()
if not fpcalc:
logger.info("AcoustID: fpcalc ikke fundet — spring over")
if on_progress:
on_progress(0, 0, "fpcalc ikke installeret")
return
logger.info(f"AcoustID: fpcalc fundet: {fpcalc}")
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
# Find sange uden MBID der har en lokal fil
rows = conn.execute("""
SELECT id, local_path, title, artist
FROM songs
WHERE (mbid IS NULL OR mbid = '')
AND file_missing = 0
AND local_path IS NOT NULL AND local_path != ''
ORDER BY RANDOM()
LIMIT 500
""").fetchall()
total = len(rows)
done = 0
found = 0
logger.info(f"AcoustID: {total} sange uden MBID")
for row in rows:
if stop_event and stop_event.is_set():
logger.info("AcoustID: stoppet af bruger")
break
path = row["local_path"]
if not Path(path).exists():
done += 1
continue
if on_progress:
on_progress(done, total, row["title"] or path)
# Fingerprint filen
fp_result = fingerprint_file(path, fpcalc)
if not fp_result:
done += 1
time.sleep(0.05)
continue
fingerprint, duration = fp_result
# Spørg AcoustID
result = lookup_acoustid(fingerprint, duration)
time.sleep(API_DELAY) # Respektér rate limit
if result:
mbid = result.get("mbid", "")
acoustid = result.get("acoustid", "")
conn.execute(
"UPDATE songs SET mbid=?, acoustid=? WHERE id=?",
(mbid or None, acoustid or None, row["id"])
)
conn.commit()
found += 1
logger.info(
f"AcoustID: {row['title']} → MBID={mbid[:8] if mbid else ''}"
)
done += 1
conn.close()
logger.info(f"AcoustID: færdig — {found}/{total} sange fik MBID")
if on_progress:
on_progress(total, total, f"Færdig — {found} sange fik MBID")
class AcoustIDWorker:
"""
Baggrunds-worker der kører AcoustID fingerprinting
som lavprioritets tråd.
"""
def __init__(self, db_path: str):
self._db_path = db_path
self._thread = None
self._stop_event = threading.Event()
self._running = False
def start(self, on_progress=None):
"""Start baggrunds-fingerprinting. Gør ingenting hvis allerede kører."""
if self._running:
return
self._stop_event.clear()
self._running = True
def _run():
try:
run_acoustid_scan(
self._db_path,
on_progress=on_progress,
stop_event=self._stop_event,
)
finally:
self._running = False
self._thread = threading.Thread(target=_run, daemon=True, name="acoustid")
self._thread.start()
logger.info("AcoustID: baggrunds-worker startet")
def stop(self):
"""Stop baggrunds-fingerprinting."""
self._stop_event.set()
def is_running(self) -> bool:
return self._running

View File

@@ -132,7 +132,7 @@ class SyncManager:
# Sange
songs = []
for row in conn.execute(
"SELECT id, title, artist, album, bpm, duration_sec, file_format, mbid "
"SELECT id, title, artist, album, bpm, duration_sec, file_format, mbid, acoustid "
"FROM songs WHERE file_missing=0"
).fetchall():
songs.append({
@@ -144,6 +144,7 @@ class SyncManager:
"duration_sec": row["duration_sec"] or 0,
"file_format": row["file_format"] or "",
"mbid": row["mbid"] or "",
"acoustid": row["acoustid"] or "",
})
# Danse

View File

@@ -565,6 +565,33 @@ class MainWindow(QMainWindow):
pass
QTimer.singleShot(5000, self.start_background_scan)
# Start AcoustID fingerprinting efter 60 sekunder hvis aktiveret
if self._settings.get("acoustid_enabled", False):
QTimer.singleShot(60000, self._start_acoustid)
def _start_acoustid(self):
"""Start AcoustID fingerprinting i baggrunden."""
from local.acoustid_worker import AcoustIDWorker, find_fpcalc
from local.local_db import DB_PATH
if not find_fpcalc():
logger.info("AcoustID: fpcalc ikke fundet — springer over")
return
if not hasattr(self, "_acoustid_worker"):
self._acoustid_worker = AcoustIDWorker(str(DB_PATH))
if self._acoustid_worker.is_running():
return
def on_progress(done, total, title):
if total > 0 and done % 10 == 0:
self._set_status(
f"AcoustID: {done}/{total}{title}", 3000
)
self._acoustid_worker.start(on_progress=on_progress)
self._set_status("AcoustID fingerprinting startet i baggrunden", 4000)
def start_background_scan(self):
"""Start scanning af alle aktive biblioteker i baggrunden."""

View File

@@ -28,6 +28,7 @@ SETTINGS_KEY_MAIN_DEVICE = "playback/audio_device_main"
SETTINGS_KEY_PREV_DEVICE = "playback/audio_device_preview"
SETTINGS_KEY_AFTER_SONG = "playback/after_song_mode"
SETTINGS_KEY_AFTER_DELAY = "playback/after_song_delay"
SETTINGS_KEY_ACOUSTID = "playback/acoustid_enabled"
def load_settings() -> dict:
@@ -50,6 +51,7 @@ def load_settings() -> dict:
"audio_device_preview":s.value(SETTINGS_KEY_PREV_DEVICE, ""),
"after_song_mode": s.value(SETTINGS_KEY_AFTER_SONG, "manual"),
"after_song_delay": s.value(SETTINGS_KEY_AFTER_DELAY, 2, type=int),
"acoustid_enabled": s.value(SETTINGS_KEY_ACOUSTID, False, type=bool),
}
@@ -72,6 +74,7 @@ def save_settings(values: dict):
s.setValue(SETTINGS_KEY_PREV_DEVICE, values.get("audio_device_preview", ""))
s.setValue(SETTINGS_KEY_AFTER_SONG, values.get("after_song_mode", "manual"))
s.setValue(SETTINGS_KEY_AFTER_DELAY, values.get("after_song_delay", 2))
s.setValue(SETTINGS_KEY_ACOUSTID, values.get("acoustid_enabled", False))
class SettingsDialog(QDialog):
@@ -260,6 +263,29 @@ class SettingsDialog(QDialog):
grp3_layout.addRow(note3)
layout.addWidget(grp3)
# AcoustID fingerprinting
grp4 = QGroupBox("AcoustID fingerprinting (valgfri)")
grp4_layout = QVBoxLayout(grp4)
grp4_layout.setSpacing(6)
self._chk_acoustid = QCheckBox("Kør AcoustID fingerprinting i baggrunden")
self._chk_acoustid.setToolTip(
"Analyserer sange uden MBID og slår dem op i AcoustID-databasen.\n"
"Kræver fpcalc (Chromaprint) installeret.\n"
"Startes automatisk 60 sekunder efter opstart."
)
grp4_layout.addWidget(self._chk_acoustid)
note4 = QLabel(
"fpcalc skal installeres separat:\n"
" Linux: sudo apt install libchromaprint-tools\n"
" Windows: download fra acoustid.org/chromaprint"
)
note4.setObjectName("result_count")
note4.setWordWrap(True)
grp4_layout.addWidget(note4)
layout.addWidget(grp4)
layout.addStretch()
return tab
@@ -449,6 +475,7 @@ class SettingsDialog(QDialog):
else:
self._radio_manual.setChecked(True)
self._spin_after_delay.setValue(v.get("after_song_delay", 2))
self._chk_acoustid.setChecked(v.get("acoustid_enabled", False))
# ── Gem ───────────────────────────────────────────────────────────────────
@@ -474,6 +501,7 @@ class SettingsDialog(QDialog):
"manual"
),
"after_song_delay": self._spin_after_delay.value(),
"acoustid_enabled": self._chk_acoustid.isChecked(),
}
save_settings(values)
self._values = values