Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
73640d8258
|
|||
|
22e8b03e6e
|
+1
-1
@@ -254,7 +254,7 @@ def to_plain(
|
|||||||
prev_line = line
|
prev_line = line
|
||||||
lines = deduped_lines
|
lines = deduped_lines
|
||||||
|
|
||||||
return "\n".join(lines)
|
return "\n".join(lines).strip("\n")
|
||||||
|
|
||||||
|
|
||||||
def print_lyrics(
|
def print_lyrics(
|
||||||
|
|||||||
+2
-5
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "lrx-cli"
|
name = "lrx-cli"
|
||||||
version = "0.1.7"
|
version = "0.2.0"
|
||||||
description = "Fetch line-synced lyrics for your music player."
|
description = "Fetch line-synced lyrics for your music player."
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.13"
|
requires-python = ">=3.13"
|
||||||
@@ -25,7 +25,4 @@ lrx = "lrx_cli.cli:run"
|
|||||||
ignore = ["E402"]
|
ignore = ["E402"]
|
||||||
|
|
||||||
[dependency-groups]
|
[dependency-groups]
|
||||||
dev = [
|
dev = ["pytest>=9.0.2", "ruff>=0.15.8"]
|
||||||
"pytest>=9.0.2",
|
|
||||||
"ruff>=0.15.8",
|
|
||||||
]
|
|
||||||
|
|||||||
@@ -0,0 +1,347 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import sqlite3
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from lrx_cli.cache import (
|
||||||
|
CacheEngine,
|
||||||
|
_generate_key,
|
||||||
|
_normalize_artist,
|
||||||
|
_normalize_for_match,
|
||||||
|
)
|
||||||
|
from lrx_cli.config import DURATION_TOLERANCE_MS
|
||||||
|
from lrx_cli.models import CacheStatus, LyricResult, TrackMeta
|
||||||
|
|
||||||
|
|
||||||
|
def _track(
|
||||||
|
*,
|
||||||
|
artist: str | None = "Artist",
|
||||||
|
title: str | None = "Song",
|
||||||
|
album: str | None = "Album",
|
||||||
|
length: int | None = 180000,
|
||||||
|
trackid: str | None = None,
|
||||||
|
url: str | None = None,
|
||||||
|
) -> TrackMeta:
|
||||||
|
return TrackMeta(
|
||||||
|
artist=artist,
|
||||||
|
title=title,
|
||||||
|
album=album,
|
||||||
|
length=length,
|
||||||
|
trackid=trackid,
|
||||||
|
url=url,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _result(
|
||||||
|
status: CacheStatus,
|
||||||
|
lyrics: str | None,
|
||||||
|
source: str,
|
||||||
|
) -> LyricResult:
|
||||||
|
return LyricResult(status=status, lyrics=lyrics, source=source)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def cache_db(tmp_path: Path) -> CacheEngine:
|
||||||
|
db_path = tmp_path / "cache.db"
|
||||||
|
return CacheEngine(str(db_path))
|
||||||
|
|
||||||
|
|
||||||
|
def test_normalize_for_match_covers_nfkc_punct_feat_and_whitespace() -> None:
|
||||||
|
text = " Test! feat. SOMEONE "
|
||||||
|
|
||||||
|
normalized = _normalize_for_match(text)
|
||||||
|
|
||||||
|
assert normalized == "test"
|
||||||
|
|
||||||
|
|
||||||
|
def test_normalize_artist_splits_separators_and_sorts_parts() -> None:
|
||||||
|
artist = "B / A feat. C; D vs. E × F 、 G"
|
||||||
|
|
||||||
|
normalized = _normalize_artist(artist)
|
||||||
|
|
||||||
|
assert normalized == "a\0b\0d\0e\0f\0g"
|
||||||
|
|
||||||
|
|
||||||
|
def test_generate_key_uses_spotify_trackid_and_url_fallback() -> None:
|
||||||
|
spotify_track = _track(
|
||||||
|
trackid="abc123", artist=None, title=None, album=None, length=None
|
||||||
|
)
|
||||||
|
local_track = _track(
|
||||||
|
artist=None, title=None, album=None, length=None, url="file:///x.flac"
|
||||||
|
)
|
||||||
|
|
||||||
|
assert _generate_key(spotify_track, "spotify") == "spotify:abc123"
|
||||||
|
assert _generate_key(local_track, "local") == "local:url:file:///x.flac"
|
||||||
|
|
||||||
|
|
||||||
|
def test_generate_key_raises_when_metadata_missing() -> None:
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
_generate_key(
|
||||||
|
_track(artist=None, title=None, album=None, length=None, url=None), "lrclib"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_set_and_get_roundtrip_with_ttl(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, cache_db: CacheEngine
|
||||||
|
) -> None:
|
||||||
|
monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 1_000_000)
|
||||||
|
|
||||||
|
track = _track()
|
||||||
|
cache_db.set(
|
||||||
|
track,
|
||||||
|
"lrclib",
|
||||||
|
_result(CacheStatus.SUCCESS_SYNCED, "[00:01.00]line", "lrclib"),
|
||||||
|
ttl_seconds=120,
|
||||||
|
)
|
||||||
|
|
||||||
|
cached = cache_db.get(track, "lrclib")
|
||||||
|
|
||||||
|
assert cached is not None
|
||||||
|
assert cached.status is CacheStatus.SUCCESS_SYNCED
|
||||||
|
assert cached.lyrics == "[00:01.00]line"
|
||||||
|
assert cached.source == "lrclib"
|
||||||
|
assert cached.ttl == 120
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_expired_entry_returns_none_and_removes_row(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, cache_db: CacheEngine
|
||||||
|
) -> None:
|
||||||
|
track = _track()
|
||||||
|
monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 2_000_000)
|
||||||
|
cache_db.set(
|
||||||
|
track,
|
||||||
|
"netease",
|
||||||
|
_result(CacheStatus.SUCCESS_UNSYNCED, "line", "netease"),
|
||||||
|
ttl_seconds=10,
|
||||||
|
)
|
||||||
|
|
||||||
|
monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 2_000_020)
|
||||||
|
cached = cache_db.get(track, "netease")
|
||||||
|
|
||||||
|
assert cached is None
|
||||||
|
assert cache_db.query_all() == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_backfills_missing_length_when_track_provides_it(
|
||||||
|
cache_db: CacheEngine,
|
||||||
|
) -> None:
|
||||||
|
track_without_length = _track(
|
||||||
|
trackid="spotify-track-1",
|
||||||
|
artist=None,
|
||||||
|
title=None,
|
||||||
|
album=None,
|
||||||
|
length=None,
|
||||||
|
)
|
||||||
|
cache_db.set(
|
||||||
|
track_without_length,
|
||||||
|
"spotify",
|
||||||
|
_result(CacheStatus.SUCCESS_SYNCED, "line", "spotify"),
|
||||||
|
)
|
||||||
|
|
||||||
|
track_with_length = _track(
|
||||||
|
trackid="spotify-track-1",
|
||||||
|
artist=None,
|
||||||
|
title=None,
|
||||||
|
album=None,
|
||||||
|
length=200000,
|
||||||
|
)
|
||||||
|
cached = cache_db.get(track_with_length, "spotify")
|
||||||
|
|
||||||
|
assert cached is not None
|
||||||
|
|
||||||
|
with sqlite3.connect(cache_db.db_path) as conn:
|
||||||
|
row = conn.execute("SELECT length FROM cache LIMIT 1").fetchone()
|
||||||
|
assert row is not None
|
||||||
|
assert row[0] == 200000
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_best_prefers_synced_over_unsynced_and_negative(
|
||||||
|
cache_db: CacheEngine,
|
||||||
|
) -> None:
|
||||||
|
track = _track()
|
||||||
|
cache_db.set(
|
||||||
|
track,
|
||||||
|
"source-a",
|
||||||
|
_result(CacheStatus.NOT_FOUND, None, "source-a"),
|
||||||
|
)
|
||||||
|
cache_db.set(
|
||||||
|
track,
|
||||||
|
"source-b",
|
||||||
|
_result(CacheStatus.SUCCESS_UNSYNCED, "unsynced", "source-b"),
|
||||||
|
)
|
||||||
|
cache_db.set(
|
||||||
|
track,
|
||||||
|
"source-c",
|
||||||
|
_result(CacheStatus.SUCCESS_SYNCED, "synced", "source-c"),
|
||||||
|
)
|
||||||
|
|
||||||
|
best = cache_db.get_best(track, ["source-a", "source-b", "source-c"])
|
||||||
|
|
||||||
|
assert best is not None
|
||||||
|
assert best.status is CacheStatus.SUCCESS_SYNCED
|
||||||
|
assert best.lyrics == "synced"
|
||||||
|
|
||||||
|
|
||||||
|
def test_clear_track_and_clear_all_affect_expected_rows(cache_db: CacheEngine) -> None:
|
||||||
|
track_a = _track(artist="A", title="T", album="X")
|
||||||
|
track_b = _track(artist="B", title="T", album="X")
|
||||||
|
|
||||||
|
cache_db.set(track_a, "s1", _result(CacheStatus.SUCCESS_SYNCED, "a1", "s1"))
|
||||||
|
cache_db.set(track_a, "s2", _result(CacheStatus.SUCCESS_UNSYNCED, "a2", "s2"))
|
||||||
|
cache_db.set(track_b, "s1", _result(CacheStatus.SUCCESS_SYNCED, "b1", "s1"))
|
||||||
|
|
||||||
|
cache_db.clear_track(track_a)
|
||||||
|
rows_after_track_clear = cache_db.query_all()
|
||||||
|
assert len(rows_after_track_clear) == 1
|
||||||
|
assert rows_after_track_clear[0]["artist"] == "B"
|
||||||
|
|
||||||
|
cache_db.clear_all()
|
||||||
|
assert cache_db.query_all() == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_prune_removes_only_expired_rows(
|
||||||
|
monkeypatch: pytest.MonkeyPatch, cache_db: CacheEngine
|
||||||
|
) -> None:
|
||||||
|
track = _track()
|
||||||
|
monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 3_000_000)
|
||||||
|
cache_db.set(
|
||||||
|
track,
|
||||||
|
"s-expired",
|
||||||
|
_result(CacheStatus.SUCCESS_SYNCED, "x", "s-expired"),
|
||||||
|
ttl_seconds=1,
|
||||||
|
)
|
||||||
|
cache_db.set(
|
||||||
|
track,
|
||||||
|
"s-active",
|
||||||
|
_result(CacheStatus.SUCCESS_SYNCED, "y", "s-active"),
|
||||||
|
ttl_seconds=100,
|
||||||
|
)
|
||||||
|
|
||||||
|
monkeypatch.setattr("lrx_cli.cache.time.time", lambda: 3_000_010)
|
||||||
|
deleted = cache_db.prune()
|
||||||
|
|
||||||
|
assert deleted == 1
|
||||||
|
rows = cache_db.query_all()
|
||||||
|
assert len(rows) == 1
|
||||||
|
assert rows[0]["source"] == "s-active"
|
||||||
|
|
||||||
|
|
||||||
|
def test_find_best_positive_uses_exact_match_and_prefers_synced(
|
||||||
|
cache_db: CacheEngine,
|
||||||
|
) -> None:
|
||||||
|
track = _track(artist="Artist", title="Song", album="Album")
|
||||||
|
cache_db.set(track, "s1", _result(CacheStatus.SUCCESS_UNSYNCED, "u", "s1"))
|
||||||
|
cache_db.set(track, "s2", _result(CacheStatus.SUCCESS_SYNCED, "s", "s2"))
|
||||||
|
|
||||||
|
best = cache_db.find_best_positive(track)
|
||||||
|
|
||||||
|
assert best is not None
|
||||||
|
assert best.status is CacheStatus.SUCCESS_SYNCED
|
||||||
|
assert best.lyrics == "s"
|
||||||
|
# find_best_positive always reports cache-search source
|
||||||
|
assert best.source == "cache-search"
|
||||||
|
|
||||||
|
|
||||||
|
def test_search_by_meta_fuzzy_rules_and_duration_sorting(cache_db: CacheEngine) -> None:
|
||||||
|
# Same logical title/artist after normalization, different length quality.
|
||||||
|
base = _track(
|
||||||
|
artist="A / B",
|
||||||
|
title="Hello,World!",
|
||||||
|
album="Album",
|
||||||
|
length=200000,
|
||||||
|
)
|
||||||
|
close_synced = _track(
|
||||||
|
artist="B vs. A",
|
||||||
|
title="hello world",
|
||||||
|
album="Else",
|
||||||
|
length=200500,
|
||||||
|
)
|
||||||
|
close_unsynced = _track(
|
||||||
|
artist="A feat. C / B",
|
||||||
|
title="HELLO WORLD",
|
||||||
|
album="Else2",
|
||||||
|
length=201000,
|
||||||
|
)
|
||||||
|
unknown_len = _track(
|
||||||
|
artist="A & B",
|
||||||
|
title="Hello World",
|
||||||
|
album="Else3",
|
||||||
|
length=None,
|
||||||
|
)
|
||||||
|
far_len = _track(
|
||||||
|
artist="A / B",
|
||||||
|
title="Hello World",
|
||||||
|
album="Else4",
|
||||||
|
length=200000 + DURATION_TOLERANCE_MS + 1,
|
||||||
|
)
|
||||||
|
|
||||||
|
cache_db.set(base, "seed", _result(CacheStatus.SUCCESS_SYNCED, "seed", "seed"))
|
||||||
|
cache_db.set(
|
||||||
|
close_synced,
|
||||||
|
"close-synced",
|
||||||
|
_result(CacheStatus.SUCCESS_SYNCED, "cs", "close-synced"),
|
||||||
|
)
|
||||||
|
cache_db.set(
|
||||||
|
close_unsynced,
|
||||||
|
"close-unsynced",
|
||||||
|
_result(CacheStatus.SUCCESS_UNSYNCED, "cu", "close-unsynced"),
|
||||||
|
)
|
||||||
|
cache_db.set(
|
||||||
|
unknown_len,
|
||||||
|
"unknown-len",
|
||||||
|
_result(CacheStatus.SUCCESS_SYNCED, "ul", "unknown-len"),
|
||||||
|
)
|
||||||
|
cache_db.set(
|
||||||
|
far_len,
|
||||||
|
"far-len",
|
||||||
|
_result(CacheStatus.SUCCESS_SYNCED, "fl", "far-len"),
|
||||||
|
)
|
||||||
|
# Negative status should never appear in search results.
|
||||||
|
cache_db.set(
|
||||||
|
_track(artist="A / B", title="Hello World", album="Else5", length=200000),
|
||||||
|
"negative",
|
||||||
|
_result(CacheStatus.NOT_FOUND, None, "negative"),
|
||||||
|
)
|
||||||
|
|
||||||
|
rows = cache_db.search_by_meta(
|
||||||
|
artist="B ; A",
|
||||||
|
title=" hello world ",
|
||||||
|
length=200000,
|
||||||
|
)
|
||||||
|
|
||||||
|
sources = [r["source"] for r in rows]
|
||||||
|
assert "negative" not in sources
|
||||||
|
assert "far-len" not in sources
|
||||||
|
# Sorted by duration diff, then synced before unsynced for equal diff.
|
||||||
|
assert sources[0] == "seed"
|
||||||
|
assert sources[1] == "close-synced"
|
||||||
|
assert sources[2] == "close-unsynced"
|
||||||
|
# Unknown length remains candidate with fallback distance priority.
|
||||||
|
assert sources[-1] == "unknown-len"
|
||||||
|
|
||||||
|
|
||||||
|
def test_query_track_and_stats_return_expected_aggregates(
|
||||||
|
cache_db: CacheEngine,
|
||||||
|
) -> None:
|
||||||
|
cache_db.set(
|
||||||
|
_track(artist="A", title="T", album="AL"),
|
||||||
|
"s1",
|
||||||
|
_result(CacheStatus.SUCCESS_SYNCED, "x", "s1"),
|
||||||
|
)
|
||||||
|
cache_db.set(
|
||||||
|
_track(artist="A", title="T", album="AL"),
|
||||||
|
"s2",
|
||||||
|
_result(CacheStatus.SUCCESS_UNSYNCED, "y", "s2"),
|
||||||
|
)
|
||||||
|
|
||||||
|
rows = cache_db.query_track(_track(artist="A", title="T", album="AL"))
|
||||||
|
stats = cache_db.stats()
|
||||||
|
|
||||||
|
assert len(rows) == 2
|
||||||
|
assert stats["total"] == 2
|
||||||
|
assert stats["active"] == 2
|
||||||
|
assert stats["expired"] == 0
|
||||||
|
assert stats["by_status"][CacheStatus.SUCCESS_SYNCED.value] == 1
|
||||||
|
assert stats["by_status"][CacheStatus.SUCCESS_UNSYNCED.value] == 1
|
||||||
@@ -182,3 +182,11 @@ def test_to_plain_fallback_for_non_synced_text_strips_start_tags() -> None:
|
|||||||
plain = to_plain(text)
|
plain = to_plain(text)
|
||||||
|
|
||||||
assert plain == "only-zero\nplain line"
|
assert plain == "only-zero\nplain line"
|
||||||
|
|
||||||
|
|
||||||
|
def test_to_plain_trims_leading_and_trailing_blank_lines() -> None:
|
||||||
|
text = "\n\n[00:01.00]line1\n\n[00:01.00]\n[00:02.00]line2\nline3\n\n"
|
||||||
|
|
||||||
|
plain = to_plain(text)
|
||||||
|
|
||||||
|
assert plain == "line1\n\nline2"
|
||||||
|
|||||||
Reference in New Issue
Block a user