mirror of https://github.com/Nezreka/SoulSync.git
The lyrics sibling of the Cover Art Filler, plus retag integration — reusing
the existing LyricsClient (LRClib) the import pipeline already uses.
- lyrics_client: extracted the LRClib fetch (exact-match-with-duration →
search fallback) into a shared _fetch_remote_lyrics, used by both
create_lrc_file (unchanged behavior) and a new check-only has_remote_lyrics.
- MissingLyricsJob (core/repair_jobs/missing_lyrics.py): scans tracks with no
.lrc sidecar and — Option A — only flags ones LRClib actually has lyrics
for, so instrumentals/interludes are never surfaced or re-flagged. Registered
in the job list; default OFF; respects the lrclib_enabled toggle.
- _fix_missing_lyrics (repair_worker): applies a finding by fetching + writing
the .lrc and embedding lyrics via create_lrc_file.
- Re-tag tool: new 'lyrics' setting ('fetch'|'skip', default skip). When
'fetch', apply_track_plans now also fetches/refreshes the .lrc per track
(fetch-if-missing, re-embed-if-exists) — threaded through scan gates, finding
details, the auto-apply path, and the manual fix handler. Settings UI
auto-renders the dropdown from setting_options; no markup needed.
- Frontend: type/action/result labels for missing_lyrics + a finding detail
render case.
Tests: 12 — has_remote_lyrics truth table, sidecar detection, scan (only-
fixable / skip-existing / lrclib-disabled), the apply handler, and retag
lyrics_action on/off. 694 repair/lyrics/cover/retag tests pass.
pull/812/head
BoulderBadgeDad
5 days ago
parent
4e3241bfad
commit
1051ef2402
@ -0,0 +1,192 @@
|
||||
"""Missing Lyrics maintenance job (Sokhi) — the lyrics sibling of the Cover
|
||||
Art Filler.
|
||||
|
||||
Scans the library for tracks that have no ``.lrc`` sidecar, asks LRClib
|
||||
whether lyrics actually exist for them (so instrumentals/interludes that
|
||||
genuinely have no lyrics are never flagged — Option A), and creates a finding
|
||||
for each fixable track. Applying a finding fetches + writes the ``.lrc`` and
|
||||
embeds the lyrics, reusing the same LyricsClient the import pipeline uses.
|
||||
|
||||
Mirrors MissingCoverArtJob's "only surface actionable findings" design.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
from core.repair_jobs import register_job
|
||||
from core.repair_jobs.base import JobContext, JobResult, RepairJob
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
logger = get_logger("repair_jobs.missing_lyrics")
|
||||
|
||||
|
||||
def _has_lrc_sidecar(file_path: str) -> bool:
|
||||
"""True if a .lrc (or .txt lyrics) sidecar already sits next to the file."""
|
||||
if not file_path:
|
||||
return False
|
||||
base = os.path.splitext(file_path)[0]
|
||||
return os.path.exists(base + '.lrc') or os.path.exists(base + '.txt')
|
||||
|
||||
|
||||
@register_job
|
||||
class MissingLyricsJob(RepairJob):
|
||||
job_id = 'missing_lyrics'
|
||||
display_name = 'Lyrics Filler'
|
||||
description = 'Finds tracks with no .lrc lyrics and fetches synced lyrics from LRClib'
|
||||
help_text = (
|
||||
'Scans your library for tracks that have no .lrc lyrics file next to them. '
|
||||
'For each one it asks LRClib whether lyrics actually exist — tracks with no '
|
||||
'lyrics available (instrumentals, interludes) are skipped, so only fixable '
|
||||
'tracks are surfaced.\n\n'
|
||||
'When lyrics are found, a finding is created so you can review and apply it. '
|
||||
'Applying writes a synced .lrc sidecar (or plain text if no synced version '
|
||||
'exists) and embeds the lyrics in the file — the same way the import pipeline '
|
||||
'and the Library Re-tag tool do.\n\n'
|
||||
'Requires LRClib to be enabled (Settings > Metadata Enhancement).'
|
||||
)
|
||||
icon = 'repair-icon-lyrics'
|
||||
default_enabled = False
|
||||
default_interval_hours = 48
|
||||
default_settings = {}
|
||||
auto_fix = False
|
||||
|
||||
def scan(self, context: JobContext) -> JobResult:
|
||||
result = JobResult()
|
||||
|
||||
# Respect the same LRClib master toggle the import pipeline uses.
|
||||
if context.config_manager and context.config_manager.get(
|
||||
'metadata_enhancement.lrclib_enabled', True) is False:
|
||||
logger.info("[Lyrics Filler] LRClib disabled in settings — skipping scan")
|
||||
return result
|
||||
|
||||
try:
|
||||
from core.lyrics_client import lyrics_client
|
||||
except Exception as e:
|
||||
logger.warning("[Lyrics Filler] lyrics client unavailable: %s", e)
|
||||
return result
|
||||
if not getattr(lyrics_client, 'api', None):
|
||||
logger.info("[Lyrics Filler] LRClib API not available — skipping scan")
|
||||
return result
|
||||
|
||||
rows = []
|
||||
conn = None
|
||||
try:
|
||||
conn = context.db._get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT t.id, t.title, ar.name, al.title, t.file_path, t.duration
|
||||
FROM tracks t
|
||||
LEFT JOIN albums al ON al.id = t.album_id
|
||||
LEFT JOIN artists ar ON ar.id = t.artist_id
|
||||
WHERE t.file_path IS NOT NULL AND t.file_path != ''
|
||||
AND t.title IS NOT NULL AND t.title != ''
|
||||
""")
|
||||
rows = cursor.fetchall()
|
||||
except Exception as e:
|
||||
logger.error("[Lyrics Filler] Error reading tracks: %s", e, exc_info=True)
|
||||
result.errors += 1
|
||||
return result
|
||||
finally:
|
||||
if conn:
|
||||
conn.close()
|
||||
|
||||
total = len(rows)
|
||||
if context.update_progress:
|
||||
context.update_progress(0, total)
|
||||
if context.report_progress:
|
||||
context.report_progress(phase=f'Checking lyrics for {total} tracks...', total=total)
|
||||
|
||||
for i, row in enumerate(rows):
|
||||
if context.check_stop():
|
||||
return result
|
||||
if i % 10 == 0 and context.wait_if_paused():
|
||||
return result
|
||||
|
||||
track_id, title, artist_name, album_title, file_path, duration = row[:6]
|
||||
result.scanned += 1
|
||||
|
||||
# Already has a sidecar on disk → nothing to do.
|
||||
if _has_lrc_sidecar(file_path):
|
||||
result.skipped += 1
|
||||
continue
|
||||
|
||||
# Option A: only flag tracks LRClib actually has lyrics for. An
|
||||
# instrumental returns nothing here and is silently skipped (never
|
||||
# re-flagged on future scans).
|
||||
try:
|
||||
duration_s = int(duration) if duration else None
|
||||
except (TypeError, ValueError):
|
||||
duration_s = None
|
||||
try:
|
||||
available = lyrics_client.has_remote_lyrics(
|
||||
title, artist_name or '', album_title, duration_s)
|
||||
except Exception as e:
|
||||
logger.debug("[Lyrics Filler] availability check failed for '%s': %s", title, e)
|
||||
available = False
|
||||
|
||||
if not available:
|
||||
result.skipped += 1
|
||||
if context.update_progress and (i + 1) % 10 == 0:
|
||||
context.update_progress(i + 1, total)
|
||||
continue
|
||||
|
||||
if context.report_progress:
|
||||
context.report_progress(
|
||||
scanned=i + 1, total=total,
|
||||
log_line=f'Found lyrics: {title} — {artist_name or "Unknown"}',
|
||||
log_type='success')
|
||||
|
||||
if context.create_finding:
|
||||
try:
|
||||
inserted = context.create_finding(
|
||||
job_id=self.job_id,
|
||||
finding_type='missing_lyrics',
|
||||
severity='info',
|
||||
entity_type='track',
|
||||
entity_id=str(track_id),
|
||||
file_path=file_path,
|
||||
title=f'Missing lyrics: {title or "Unknown"}',
|
||||
description=f'"{title}" by {artist_name or "Unknown"} has no .lrc — lyrics found on LRClib.',
|
||||
details={
|
||||
'track_id': track_id,
|
||||
'track_title': title,
|
||||
'artist': artist_name,
|
||||
'album_title': album_title,
|
||||
'file_path': file_path,
|
||||
'duration': duration_s,
|
||||
})
|
||||
if inserted:
|
||||
result.findings_created += 1
|
||||
else:
|
||||
result.findings_skipped_dedup += 1
|
||||
except Exception as e:
|
||||
logger.debug("[Lyrics Filler] create finding failed for track %s: %s", track_id, e)
|
||||
result.errors += 1
|
||||
|
||||
if context.update_progress and (i + 1) % 5 == 0:
|
||||
context.update_progress(i + 1, total)
|
||||
|
||||
if context.update_progress:
|
||||
context.update_progress(total, total)
|
||||
logger.info("[Lyrics Filler] %d tracks checked, %d with lyrics found, %d skipped",
|
||||
result.scanned, result.findings_created, result.skipped)
|
||||
return result
|
||||
|
||||
def estimate_scope(self, context: JobContext) -> int:
|
||||
conn = None
|
||||
try:
|
||||
conn = context.db._get_connection()
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("""
|
||||
SELECT COUNT(*) FROM tracks
|
||||
WHERE file_path IS NOT NULL AND file_path != ''
|
||||
AND title IS NOT NULL AND title != ''
|
||||
""")
|
||||
row = cursor.fetchone()
|
||||
return row[0] if row else 0
|
||||
except Exception:
|
||||
return 0
|
||||
finally:
|
||||
if conn:
|
||||
conn.close()
|
||||
@ -0,0 +1,204 @@
|
||||
"""Missing Lyrics maintenance job + lyrics_client check-only seam (Sokhi).
|
||||
|
||||
Mirrors the Cover Art Filler: scan only flags tracks LRClib actually has
|
||||
lyrics for (Option A — instrumentals never flagged), and applying writes the
|
||||
.lrc via the shared LyricsClient.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from core.lyrics_client import LyricsClient
|
||||
from core.repair_jobs.missing_lyrics import MissingLyricsJob, _has_lrc_sidecar
|
||||
|
||||
|
||||
# ── lyrics_client.has_remote_lyrics (check-only seam) ────────────────────────
|
||||
|
||||
def _client_with_api(api):
|
||||
c = LyricsClient.__new__(LyricsClient)
|
||||
c.api = api
|
||||
return c
|
||||
|
||||
|
||||
def test_has_remote_lyrics_true_when_synced():
|
||||
api = MagicMock()
|
||||
api.get_lyrics.return_value = SimpleNamespace(synced_lyrics="[00:01]hi", plain_lyrics=None)
|
||||
c = _client_with_api(api)
|
||||
assert c.has_remote_lyrics("Song", "Artist", "Album", 200) is True
|
||||
|
||||
|
||||
def test_has_remote_lyrics_true_when_plain_only_via_search():
|
||||
api = MagicMock()
|
||||
api.get_lyrics.return_value = None
|
||||
api.search_lyrics.return_value = [SimpleNamespace(synced_lyrics=None, plain_lyrics="words")]
|
||||
c = _client_with_api(api)
|
||||
assert c.has_remote_lyrics("Song", "Artist") is True
|
||||
|
||||
|
||||
def test_has_remote_lyrics_false_when_none():
|
||||
api = MagicMock()
|
||||
api.get_lyrics.return_value = None
|
||||
api.search_lyrics.return_value = []
|
||||
assert _client_with_api(api).has_remote_lyrics("Instrumental", "Artist") is False
|
||||
|
||||
|
||||
def test_has_remote_lyrics_false_when_no_api():
|
||||
c = LyricsClient.__new__(LyricsClient)
|
||||
c.api = None
|
||||
assert c.has_remote_lyrics("Song", "Artist") is False
|
||||
|
||||
|
||||
# ── sidecar detection ────────────────────────────────────────────────────────
|
||||
|
||||
def test_has_lrc_sidecar(tmp_path):
|
||||
audio = tmp_path / "track.flac"
|
||||
audio.write_bytes(b"x")
|
||||
assert _has_lrc_sidecar(str(audio)) is False
|
||||
(tmp_path / "track.lrc").write_text("[00:01]hi")
|
||||
assert _has_lrc_sidecar(str(audio)) is True
|
||||
|
||||
|
||||
# ── the scan (Option A: only flag fixable tracks) ────────────────────────────
|
||||
|
||||
class _DB:
|
||||
def __init__(self, rows):
|
||||
self._rows = rows
|
||||
|
||||
def _get_connection(self):
|
||||
cur = MagicMock()
|
||||
cur.execute.return_value = None
|
||||
cur.fetchone.return_value = [len(self._rows)]
|
||||
cur.fetchall.return_value = self._rows
|
||||
conn = MagicMock()
|
||||
conn.cursor.return_value = cur
|
||||
return conn
|
||||
|
||||
|
||||
def _ctx(db, findings):
|
||||
return SimpleNamespace(
|
||||
db=db,
|
||||
config_manager=SimpleNamespace(get=lambda k, d=None: d),
|
||||
check_stop=lambda: False, wait_if_paused=lambda: False,
|
||||
update_progress=lambda *a, **k: None, report_progress=lambda *a, **k: None,
|
||||
create_finding=lambda **kw: (findings.append(kw) or True),
|
||||
)
|
||||
|
||||
|
||||
def test_scan_flags_only_tracks_with_available_lyrics(tmp_path, monkeypatch):
|
||||
# Two tracks, neither has a .lrc. LRClib has lyrics for the first, not the second.
|
||||
t1 = tmp_path / "song.flac"; t1.write_bytes(b"x")
|
||||
t2 = tmp_path / "instrumental.flac"; t2.write_bytes(b"x")
|
||||
rows = [
|
||||
(1, "Song", "Artist", "Album", str(t1), 200),
|
||||
(2, "Interlude", "Artist", "Album", str(t2), 60),
|
||||
]
|
||||
fake_client = SimpleNamespace(
|
||||
api=object(),
|
||||
has_remote_lyrics=lambda title, artist, album, dur: title == "Song",
|
||||
)
|
||||
monkeypatch.setattr("core.lyrics_client.lyrics_client", fake_client)
|
||||
|
||||
findings = []
|
||||
result = MissingLyricsJob().scan(_ctx(_DB(rows), findings))
|
||||
|
||||
assert result.findings_created == 1
|
||||
assert findings[0]["entity_type"] == "track"
|
||||
assert findings[0]["finding_type"] == "missing_lyrics"
|
||||
assert findings[0]["details"]["track_title"] == "Song" # the instrumental was skipped
|
||||
|
||||
|
||||
def test_scan_skips_tracks_that_already_have_lrc(tmp_path, monkeypatch):
|
||||
t1 = tmp_path / "song.flac"; t1.write_bytes(b"x")
|
||||
(tmp_path / "song.lrc").write_text("[00:01]hi") # already has lyrics
|
||||
rows = [(1, "Song", "Artist", "Album", str(t1), 200)]
|
||||
fake_client = SimpleNamespace(api=object(),
|
||||
has_remote_lyrics=lambda *a, **k: True)
|
||||
monkeypatch.setattr("core.lyrics_client.lyrics_client", fake_client)
|
||||
|
||||
findings = []
|
||||
result = MissingLyricsJob().scan(_ctx(_DB(rows), findings))
|
||||
assert result.findings_created == 0
|
||||
assert findings == []
|
||||
|
||||
|
||||
def test_scan_noops_when_lrclib_disabled(monkeypatch):
|
||||
db = _DB([(1, "Song", "Artist", "Album", "/x.flac", 200)])
|
||||
ctx = _ctx(db, [])
|
||||
ctx.config_manager = SimpleNamespace(
|
||||
get=lambda k, d=None: False if k == 'metadata_enhancement.lrclib_enabled' else d)
|
||||
result = MissingLyricsJob().scan(ctx)
|
||||
assert result.scanned == 0 and result.findings_created == 0
|
||||
|
||||
|
||||
# ── _fix_missing_lyrics apply handler ────────────────────────────────────────
|
||||
|
||||
def test_fix_missing_lyrics_calls_create_lrc(tmp_path, monkeypatch):
|
||||
from core.repair_worker import RepairWorker
|
||||
audio = tmp_path / "song.flac"; audio.write_bytes(b"x")
|
||||
|
||||
w = RepairWorker.__new__(RepairWorker)
|
||||
w.transfer_folder = str(tmp_path)
|
||||
w._config_manager = SimpleNamespace(get=lambda k, d=None: d)
|
||||
|
||||
calls = {}
|
||||
fake_client = SimpleNamespace(
|
||||
create_lrc_file=lambda path, title, artist, album_name=None, duration_seconds=None:
|
||||
calls.update(path=path, title=title, artist=artist) or True)
|
||||
monkeypatch.setattr("core.lyrics_client.lyrics_client", fake_client)
|
||||
# _resolve_file_path: the file is already real, so identity is fine.
|
||||
monkeypatch.setattr("core.repair_worker._resolve_file_path",
|
||||
lambda raw, *a, **k: raw)
|
||||
|
||||
res = w._fix_missing_lyrics("track", "1", None, {
|
||||
"file_path": str(audio), "track_title": "Song", "artist": "Artist",
|
||||
"album_title": "Album", "duration": 200})
|
||||
assert res["success"] is True and res["action"] == "applied_lyrics"
|
||||
assert calls["title"] == "Song" and calls["path"] == str(audio)
|
||||
|
||||
|
||||
def test_fix_missing_lyrics_missing_file(tmp_path, monkeypatch):
|
||||
from core.repair_worker import RepairWorker
|
||||
w = RepairWorker.__new__(RepairWorker)
|
||||
w.transfer_folder = str(tmp_path)
|
||||
w._config_manager = SimpleNamespace(get=lambda k, d=None: d)
|
||||
monkeypatch.setattr("core.repair_worker._resolve_file_path", lambda raw, *a, **k: raw)
|
||||
res = w._fix_missing_lyrics("track", "1", None, {"file_path": str(tmp_path / "gone.flac")})
|
||||
assert res["success"] is False
|
||||
|
||||
|
||||
# ── retag apply_track_plans lyrics_action ────────────────────────────────────
|
||||
|
||||
def test_apply_track_plans_lyrics_action(tmp_path, monkeypatch):
|
||||
from core.repair_jobs import library_retag
|
||||
audio = tmp_path / "t.flac"; audio.write_bytes(b"x")
|
||||
|
||||
monkeypatch.setattr(library_retag, "write_tags_to_file",
|
||||
lambda *a, **k: {"success": True}, raising=False)
|
||||
seen = {}
|
||||
fake_client = SimpleNamespace(
|
||||
create_lrc_file=lambda path, title, artist, album_name=None, duration_seconds=None:
|
||||
seen.update(title=title) or True)
|
||||
monkeypatch.setattr("core.lyrics_client.lyrics_client", fake_client)
|
||||
|
||||
plans = [{"file_path": str(audio), "db_data": {"title": "Song", "artist": "Artist"}}]
|
||||
res = library_retag.apply_track_plans(plans, lyrics_action=True)
|
||||
assert res["lyrics_written"] == 1 and seen["title"] == "Song"
|
||||
|
||||
|
||||
def test_apply_track_plans_no_lyrics_when_disabled(tmp_path, monkeypatch):
|
||||
from core.repair_jobs import library_retag
|
||||
audio = tmp_path / "t.flac"; audio.write_bytes(b"x")
|
||||
monkeypatch.setattr(library_retag, "write_tags_to_file",
|
||||
lambda *a, **k: {"success": True}, raising=False)
|
||||
called = []
|
||||
fake_client = SimpleNamespace(create_lrc_file=lambda *a, **k: called.append(1) or True)
|
||||
monkeypatch.setattr("core.lyrics_client.lyrics_client", fake_client)
|
||||
|
||||
plans = [{"file_path": str(audio), "db_data": {"title": "Song"}}]
|
||||
res = library_retag.apply_track_plans(plans, lyrics_action=False)
|
||||
assert res["lyrics_written"] == 0 and called == []
|
||||
Loading…
Reference in new issue