Uyanide/lrx-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add api fixtures

Browse Source
This commit is contained in:
Uyanide
2026-04-10 08:45:24 +02:00
parent 60732f2986
commit 0f1f5b418a
23 changed files with 1324 additions and 483 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+4 -3
View File
@@ -133,7 +133,8 @@ socket_path = "" # Unix socket path; defaults to <cache_dir>/
for the Spotify source; leave empty to disable it.
- `musixmatch_usertoken` — found at
[Curators Settings Page](https://curators.musixmatch.com/settings) → Login → "Copy debug info".
If empty, an anonymous token is fetched at runtime.
If empty, an anonymous token will be fetched at runtime, which could be more likely to
hit the rate limits.
- `qq_music_api_url` — base URL of a self-hosted
[qq-music-api](https://github.com/tooplick/qq-music-api) (compatible) instance. Required
for the QQ Music source; leave empty to disable it.
@@ -154,13 +155,13 @@ uv venv .venv
uv sync
```
Run tests without network calls
Run tests without network calls:
```bash
uv run pytest -m "not network"
```
or full tests:
or run full tests. The **REAL** API calls will be made and some of them will be skipped if the required credentials are not configured as [above](#configuration). This might be useful to verify that the lyric sources are still valid and working as expected:
```bash
uv run pytest
pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "lrx-cli"
version = "0.7.3"
version = "0.7.4"
description = "Fetch line-synced lyrics for your music player."
readme = "README.md"
requires-python = ">=3.13"
src/lrx_cli/authenticators/qqmusic.py
+44
View File
@@ -5,6 +5,8 @@ Description: QQ Music API authenticator - currently only a proxy.
"""
from typing import Optional
import httpx
from loguru import logger
from .base import BaseAuthenticator
from ..cache import CacheEngine
@@ -26,3 +28,45 @@ class QQMusicAuthenticator(BaseAuthenticator):
async def authenticate(self) -> Optional[str]:
return self._credentials.qq_music_api_url.rstrip("/") or None
async def search(self, keyword: str, num: int) -> dict | None:
"""Call qq-music-api search endpoint and return raw JSON payload."""
base_url = await self.authenticate()
if not base_url:
return None
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
resp = await client.get(
f"{base_url}/api/search",
params={"keyword": keyword, "type": "song", "num": num},
)
resp.raise_for_status()
data = resp.json()
if not isinstance(data, dict):
return None
return data
except Exception as e:
logger.error(f"QQMusic: search request failed: {e}")
return None
async def get_lyric(self, mid: str) -> dict | None:
"""Call qq-music-api lyric endpoint and return raw JSON payload."""
base_url = await self.authenticate()
if not base_url:
return None
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
resp = await client.get(
f"{base_url}/api/lyric",
params={"mid": mid},
)
resp.raise_for_status()
data = resp.json()
if not isinstance(data, dict):
return None
return data
except Exception as e:
logger.error(f"QQMusic: lyric request failed for mid={mid}: {e}")
return None
src/lrx_cli/authenticators/spotify.py
+33
View File
@@ -18,6 +18,7 @@ from ..config import CredentialConfig, GeneralConfig, UA_BROWSER
_SPOTIFY_TOKEN_URL = "https://open.spotify.com/api/token"
_SPOTIFY_SERVER_TIME_URL = "https://open.spotify.com/api/server-time"
_SPOTIFY_LYRICS_URL = "https://spclient.wg.spotify.com/color-lyrics/v2/track/"
_SPOTIFY_SECRET_URL = (
"https://raw.githubusercontent.com/xyloflake/spot-secrets-go"
"/refs/heads/main/secrets/secrets.json"
@@ -208,3 +209,35 @@ class SpotifyAuthenticator(BaseAuthenticator):
except Exception as e:
logger.error(f"Spotify: token request failed: {e}")
return None
async def get_lyrics(self, track_id: str) -> dict | None:
"""Fetch raw lyrics JSON payload for a Spotify track."""
token = await self.authenticate()
if not token:
return None
url = (
f"{_SPOTIFY_LYRICS_URL}{track_id}"
"?format=json&vocalRemoval=false&market=from_token"
)
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {token}",
**SPOTIFY_BASE_HEADERS,
}
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
res = await client.get(url, headers=headers)
if res.status_code == 404:
return None
if res.status_code != 200:
logger.error(f"Spotify: lyrics API returned {res.status_code}")
return None
data = res.json()
if not isinstance(data, dict):
return None
return data
except Exception as e:
logger.error(f"Spotify: lyrics fetch failed: {e}")
return None
src/lrx_cli/fetchers/lrclib.py
+55 -36
View File
@@ -21,6 +21,38 @@ from ..config import (
_LRCLIB_API_URL = "https://lrclib.net/api/get"
def _parse_lrclib_response(data: dict) -> FetchResult:
"""Parse LRCLIB JSON response into synced/unsynced fetch result."""
synced = data.get("syncedLyrics")
unsynced = data.get("plainLyrics")
res_synced: LyricResult = LyricResult(
status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND
)
res_unsynced: LyricResult = LyricResult(
status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND
)
if isinstance(synced, str) and synced.strip():
lyrics = LRCData(synced)
res_synced = LyricResult(
status=CacheStatus.SUCCESS_SYNCED,
lyrics=lyrics,
source="lrclib",
)
if isinstance(unsynced, str) and unsynced.strip():
lyrics = LRCData(unsynced)
res_unsynced = LyricResult(
status=CacheStatus.SUCCESS_UNSYNCED,
lyrics=lyrics,
source="lrclib",
ttl=TTL_UNSYNCED,
)
return FetchResult(synced=res_synced, unsynced=res_unsynced)
class LrclibFetcher(BaseFetcher):
@property
def source_name(self) -> str:
@@ -29,12 +61,12 @@ class LrclibFetcher(BaseFetcher):
def is_available(self, track: TrackMeta) -> bool:
return track.is_complete
async def fetch(self, track: TrackMeta, bypass_cache: bool = False) -> FetchResult:
"""Fetch lyrics from LRCLIB. Requires complete metadata."""
if not track.is_complete:
logger.debug("LRCLIB: skipped — incomplete metadata")
return FetchResult()
async def _api_get(
self,
client: httpx.AsyncClient,
track: TrackMeta,
) -> httpx.Response:
"""Issue one LRCLIB get request using the same path as production fetch."""
params = {
"track_name": track.title,
"artist_name": track.artist,
@@ -42,11 +74,19 @@ class LrclibFetcher(BaseFetcher):
"duration": track.length / 1000.0 if track.length else 0,
}
url = f"{_LRCLIB_API_URL}?{urlencode(params)}"
return await client.get(url, headers={"User-Agent": UA_LRX})
async def fetch(self, track: TrackMeta, bypass_cache: bool = False) -> FetchResult:
"""Fetch lyrics from LRCLIB. Requires complete metadata."""
if not track.is_complete:
logger.debug("LRCLIB: skipped — incomplete metadata")
return FetchResult()
logger.info(f"LRCLIB: fetching lyrics for {track.display_name()}")
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
resp = await client.get(url, headers={"User-Agent": UA_LRX})
resp = await self._api_get(client, track)
if resp.status_code == 404:
logger.debug(f"LRCLIB: not found for {track.display_name()}")
@@ -60,37 +100,16 @@ class LrclibFetcher(BaseFetcher):
if not isinstance(data, dict):
logger.error(f"LRCLIB: unexpected response type: {type(data).__name__}")
return FetchResult.from_network_error()
synced = data.get("syncedLyrics")
unsynced = data.get("plainLyrics")
res_synced: LyricResult = LyricResult(
status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND
result = _parse_lrclib_response(data)
if result.synced and result.synced.lyrics:
logger.info(
f"LRCLIB: got synced lyrics ({len(result.synced.lyrics)} lines)"
)
res_unsynced: LyricResult = LyricResult(
status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND
if result.unsynced and result.unsynced.lyrics:
logger.info(
f"LRCLIB: got unsynced lyrics ({len(result.unsynced.lyrics)} lines)"
)
if isinstance(synced, str) and synced.strip():
lyrics = LRCData(synced)
logger.info(f"LRCLIB: got synced lyrics ({len(lyrics)} lines)")
res_synced = LyricResult(
status=CacheStatus.SUCCESS_SYNCED,
lyrics=lyrics,
source=self.source_name,
)
if isinstance(unsynced, str) and unsynced.strip():
lyrics = LRCData(unsynced)
logger.info(f"LRCLIB: got unsynced lyrics ({len(lyrics)} lines)")
res_unsynced = LyricResult(
status=CacheStatus.SUCCESS_UNSYNCED,
lyrics=lyrics,
source=self.source_name,
ttl=TTL_UNSYNCED,
)
return FetchResult(synced=res_synced, unsynced=res_unsynced)
return result
except httpx.HTTPError as e:
logger.error(f"LRCLIB: HTTP error: {e}")
src/lrx_cli/fetchers/lrclib_search.py
+51 -32
View File
@@ -23,6 +23,24 @@ from ..config import (
_LRCLIB_SEARCH_URL = "https://lrclib.net/api/search"
def _parse_lrclib_search_results(items: list[dict]) -> list[SearchCandidate[dict]]:
"""Map LRCLIB search JSON items to normalized SearchCandidate entries."""
return [
SearchCandidate(
item=item,
duration_ms=item["duration"] * 1000
if isinstance(item.get("duration"), (int, float))
else None,
is_synced=isinstance(item.get("syncedLyrics"), str)
and bool(item["syncedLyrics"].strip()),
title=item.get("trackName"),
artist=item.get("artistName"),
album=item.get("albumName"),
)
for item in items
]
class LrclibSearchFetcher(BaseFetcher):
@property
def source_name(self) -> str:
@@ -59,22 +77,12 @@ class LrclibSearchFetcher(BaseFetcher):
return queries
async def fetch(self, track: TrackMeta, bypass_cache: bool = False) -> FetchResult:
if not track.title:
logger.debug("LRCLIB-search: skipped — no title")
return FetchResult()
queries = self._build_queries(track)
logger.info(f"LRCLIB-search: searching for {track.display_name()}")
seen_ids: set[int] = set()
candidates: list[dict] = []
had_error = False
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
async def _query(params: dict[str, str]) -> tuple[list[dict], bool]:
async def _api_query(
self,
client: httpx.AsyncClient,
params: dict[str, str],
) -> tuple[list[dict], bool]:
"""Issue one LRCLIB search query using production request path."""
url = f"{_LRCLIB_SEARCH_URL}?{urlencode(params)}"
logger.debug(f"LRCLIB-search: query {params}")
try:
@@ -90,8 +98,20 @@ class LrclibSearchFetcher(BaseFetcher):
return [], False
return [item for item in data if isinstance(item, dict)], False
all_results = await asyncio.gather(*(_query(p) for p in queries))
async def _api_candidates(
self,
client: httpx.AsyncClient,
track: TrackMeta,
) -> tuple[list[dict], bool]:
"""Request and merge LRCLIB-search candidates using built-in query strategy."""
queries = self._build_queries(track)
all_results = await asyncio.gather(
*(self._api_query(client, p) for p in queries)
)
seen_ids: set[int] = set()
candidates: list[dict] = []
had_error = False
for items, err in all_results:
if err:
had_error = True
@@ -102,6 +122,18 @@ class LrclibSearchFetcher(BaseFetcher):
if item_id is not None:
seen_ids.add(item_id)
candidates.append(item)
return candidates, had_error
async def fetch(self, track: TrackMeta, bypass_cache: bool = False) -> FetchResult:
if not track.title:
logger.debug("LRCLIB-search: skipped — no title")
return FetchResult()
logger.info(f"LRCLIB-search: searching for {track.display_name()}")
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
candidates, had_error = await self._api_candidates(client, track)
if not candidates:
if had_error:
@@ -111,23 +143,10 @@ class LrclibSearchFetcher(BaseFetcher):
logger.debug(
f"LRCLIB-search: got {len(candidates)} unique candidates "
f"from {len(queries)} queries"
f"from {len(self._build_queries(track))} queries"
)
mapped = [
SearchCandidate(
item=item,
duration_ms=item["duration"] * 1000
if isinstance(item.get("duration"), (int, float))
else None,
is_synced=isinstance(item.get("syncedLyrics"), str)
and bool(item["syncedLyrics"].strip()),
title=item.get("trackName"),
artist=item.get("artistName"),
album=item.get("albumName"),
)
for item in candidates
]
mapped = _parse_lrclib_search_results(candidates)
best, confidence = select_best(
mapped,
track.length,
src/lrx_cli/fetchers/musixmatch.py
+113 -53
View File
@@ -83,21 +83,8 @@ def _parse_subtitle(body: str) -> Optional[str]:
return None
async def _fetch_macro(
auth: MusixmatchAuthenticator,
params: dict,
) -> Optional[LRCData]:
"""Call macro.subtitles.get via auth.get_json.
Returns LRCData (richsync preferred over subtitle), or None when no usable
lyrics are found. Raises on HTTP/network errors.
"""
logger.debug(f"Musixmatch: macro call with {list(params.keys())}")
data = await auth.get_json(_MUSIXMATCH_MACRO_URL, {**_MXM_MACRO_PARAMS, **params})
if data is None:
return None
# Musixmatch returns body=[] (not {}) when the track is not found
def _parse_mxm_macro(data: dict) -> LRCData | None:
"""Parse macro.subtitles.get payload into LRCData (richsync preferred)."""
body = data.get("message", {}).get("body", {})
if not isinstance(body, dict):
return None
@@ -105,7 +92,6 @@ async def _fetch_macro(
if not isinstance(macro_calls, dict):
return None
# Prefer richsync (word-level timing)
richsync_msg = macro_calls.get("track.richsync.get", {}).get("message", {})
if (
isinstance(richsync_msg, dict)
@@ -119,10 +105,8 @@ async def _fetch_macro(
if lrc_text:
lrc = LRCData(lrc_text)
if lrc:
logger.debug("Musixmatch: got richsync lyrics")
return lrc
# Fall back to subtitle (line-level timing)
subtitle_msg = macro_calls.get("track.subtitles.get", {}).get("message", {})
if (
isinstance(subtitle_msg, dict)
@@ -136,13 +120,36 @@ async def _fetch_macro(
if lrc_text:
lrc = LRCData(lrc_text)
if lrc:
logger.debug("Musixmatch: got subtitle lyrics")
return lrc
logger.debug("Musixmatch: no usable lyrics in macro response")
return None
def _parse_mxm_search(data: dict) -> list[SearchCandidate[int]]:
"""Parse track.search payload to normalized candidates."""
track_list = data.get("message", {}).get("body", {}).get("track_list", [])
if not isinstance(track_list, list) or not track_list:
return []
return [
SearchCandidate(
item=int(t["commontrack_id"]),
duration_ms=(
float(t["track_length"]) * 1000 if t.get("track_length") else None
),
is_synced=bool(t.get("has_subtitles") or t.get("has_richsync")),
title=t.get("track_name"),
artist=t.get("artist_name"),
album=t.get("album_name"),
)
for item in track_list
if isinstance(item, dict)
and isinstance(t := item.get("track", {}), dict)
and isinstance(t.get("commontrack_id"), int)
and not t.get("instrumental")
]
class MusixmatchSpotifyFetcher(BaseFetcher):
"""Direct lookup by Spotify track ID — no search, single request."""
@@ -158,14 +165,36 @@ class MusixmatchSpotifyFetcher(BaseFetcher):
def is_available(self, track: TrackMeta) -> bool:
return bool(track.trackid) and not self._auth.is_cooldown()
async def _api_macro(self, params: dict) -> dict | None:
"""Request macro payload through authenticator using production path."""
return await self._auth.get_json(
_MUSIXMATCH_MACRO_URL, {**_MXM_MACRO_PARAMS, **params}
)
async def _api_macro_track(self, track: TrackMeta) -> dict | None:
"""Request macro payload for one track using Spotify ID lookup path."""
if not track.trackid:
return None
return await self._api_macro({"track_spotify_id": track.trackid})
async def _fetch_macro(self, params: dict) -> LRCData | None:
"""Request and parse Musixmatch macro lyrics payload."""
logger.debug(f"Musixmatch: macro call with {list(params.keys())}")
data = await self._api_macro(params)
if data is None:
return None
lrc = _parse_mxm_macro(data)
if lrc is None:
logger.debug("Musixmatch: no usable lyrics in macro response")
return None
logger.debug("Musixmatch: parsed macro lyrics")
return lrc
async def fetch(self, track: TrackMeta, bypass_cache: bool = False) -> FetchResult:
logger.info(f"Musixmatch-Spotify: fetching lyrics for {track.display_name()}")
try:
lrc = await _fetch_macro(
self._auth,
{"track_spotify_id": track.trackid}, # type: ignore[dict-item]
)
lrc = await self._fetch_macro({"track_spotify_id": track.trackid}) # type: ignore[dict-item]
except AttributeError:
return FetchResult.from_not_found()
except Exception as e:
@@ -210,9 +239,13 @@ class MusixmatchFetcher(BaseFetcher):
def is_available(self, track: TrackMeta) -> bool:
return bool(track.title) and not self._auth.is_cooldown()
async def _search(self, track: TrackMeta) -> tuple[Optional[int], float]:
"""Search for track metadata. Raises on network/HTTP errors."""
params: dict = {
async def _api_search(self, params: dict) -> dict | None:
"""Request search payload through authenticator using production path."""
return await self._auth.get_json(_MUSIXMATCH_SEARCH_URL, params)
def _build_search_params(self, track: TrackMeta) -> dict[str, str]:
"""Build Musixmatch search params for one track."""
params: dict[str, str] = {
"q_track": track.title or "",
"page_size": "10",
"f_has_lyrics": "1",
@@ -221,36 +254,66 @@ class MusixmatchFetcher(BaseFetcher):
params["q_artist"] = track.artist
if track.album:
params["q_album"] = track.album
return params
async def _api_search_track(self, track: TrackMeta) -> dict | None:
"""Request search payload for one track using production path."""
return await self._api_search(self._build_search_params(track))
async def _api_macro(self, params: dict) -> dict | None:
"""Request macro payload through authenticator using production path."""
return await self._auth.get_json(
_MUSIXMATCH_MACRO_URL, {**_MXM_MACRO_PARAMS, **params}
)
async def _api_macro_track(self, track: TrackMeta) -> dict | None:
"""Request macro payload for top-ranked search candidate of one track."""
search_data = await self._api_search_track(track)
if search_data is None:
return None
candidates = _parse_mxm_search(search_data)
if not candidates:
return None
commontrack_id, _confidence = select_best(
candidates,
track.length,
title=track.title,
artist=track.artist,
album=track.album,
)
if commontrack_id is None:
return None
return await self._api_macro({"commontrack_id": str(commontrack_id)})
async def _fetch_macro(self, params: dict) -> LRCData | None:
"""Request and parse Musixmatch macro lyrics payload."""
logger.debug(f"Musixmatch: macro call with {list(params.keys())}")
data = await self._api_macro(params)
if data is None:
return None
lrc = _parse_mxm_macro(data)
if lrc is None:
logger.debug("Musixmatch: no usable lyrics in macro response")
return None
logger.debug("Musixmatch: parsed macro lyrics")
return lrc
async def _search(self, track: TrackMeta) -> tuple[Optional[int], float]:
"""Search for track metadata. Raises on network/HTTP errors."""
logger.debug(f"Musixmatch: searching for '{track.display_name()}'")
data = await self._auth.get_json(_MUSIXMATCH_SEARCH_URL, params)
data = await self._api_search_track(track)
if data is None:
return None, 0.0
track_list = data.get("message", {}).get("body", {}).get("track_list", [])
if not isinstance(track_list, list) or not track_list:
candidates = _parse_mxm_search(data)
if not candidates:
logger.debug("Musixmatch: search returned 0 results")
return None, 0.0
logger.debug(f"Musixmatch: search returned {len(track_list)} candidates")
candidates = [
SearchCandidate(
item=int(t["commontrack_id"]),
duration_ms=(
float(t["track_length"]) * 1000 if t.get("track_length") else None
),
is_synced=bool(t.get("has_subtitles") or t.get("has_richsync")),
title=t.get("track_name"),
artist=t.get("artist_name"),
album=t.get("album_name"),
)
for item in track_list
if isinstance(item, dict)
and isinstance(t := item.get("track", {}), dict)
and isinstance(t.get("commontrack_id"), int)
and not t.get("instrumental")
]
logger.debug(f"Musixmatch: search returned {len(candidates)} candidates")
best_id, confidence = select_best(
candidates,
@@ -274,10 +337,7 @@ class MusixmatchFetcher(BaseFetcher):
logger.debug(f"Musixmatch: no match found for {track.display_name()}")
return FetchResult.from_not_found()
lrc = await _fetch_macro(
self._auth,
{"commontrack_id": str(commontrack_id)},
)
lrc = await self._fetch_macro({"commontrack_id": str(commontrack_id)})
except AttributeError:
return FetchResult.from_not_found()
except Exception as e:
src/lrx_cli/fetchers/netease.py
+129 -66
View File
@@ -30,6 +30,42 @@ _NETEASE_BASE_HEADERS = {
}
def _parse_netease_search(data: dict) -> list[SearchCandidate[int]]:
"""Parse Netease search response into scored candidates."""
result_body = data.get("result")
if not isinstance(result_body, dict):
return []
songs = result_body.get("songs")
if not isinstance(songs, list) or len(songs) == 0:
return []
return [
SearchCandidate(
item=song_id,
duration_ms=float(song["dt"]) if isinstance(song.get("dt"), int) else None,
title=song.get("name"),
artist=", ".join(a.get("name", "") for a in song.get("ar", [])) or None,
album=(song.get("al") or {}).get("name"),
)
for song in songs
if isinstance(song, dict) and isinstance(song_id := song.get("id"), int)
]
def _parse_netease_lyrics(data: dict) -> LRCData | None:
"""Parse Netease lyric response to LRCData."""
lrc_obj = data.get("lrc")
if not isinstance(lrc_obj, dict):
return None
lrc = lrc_obj.get("lyric", "")
if not isinstance(lrc, str) or not lrc.strip():
return None
return LRCData(lrc)
class NeteaseFetcher(BaseFetcher):
@property
def source_name(self) -> str:
@@ -38,6 +74,88 @@ class NeteaseFetcher(BaseFetcher):
def is_available(self, track: TrackMeta) -> bool:
return bool(track.title)
async def _api_search(
self,
client: httpx.AsyncClient,
query: str,
limit: int,
) -> dict | None:
"""Issue one Netease search request and return JSON payload."""
resp = await client.post(
_NETEASE_SEARCH_URL,
headers=_NETEASE_BASE_HEADERS,
data={"s": query, "type": "1", "limit": str(limit), "offset": "0"},
)
resp.raise_for_status()
data = resp.json()
if not isinstance(data, dict):
return None
return data
async def _api_search_track(
self,
client: httpx.AsyncClient,
track: TrackMeta,
limit: int,
) -> dict | None:
"""Request Netease search payload for one track using production query strategy."""
query = f"{track.artist or ''} {track.title or ''}".strip()
if not query:
return None
return await self._api_search(client, query, limit)
async def _api_lyric(
self,
client: httpx.AsyncClient,
song_id: int,
) -> dict | None:
"""Issue one Netease lyric request and return JSON payload."""
resp = await client.post(
_NETEASE_LYRIC_URL,
headers=_NETEASE_BASE_HEADERS,
data={
"id": str(song_id),
"cp": "false",
"tv": "0",
"lv": "0",
"rv": "0",
"kv": "0",
"yv": "0",
"ytv": "0",
"yrv": "0",
},
)
resp.raise_for_status()
data = resp.json()
if not isinstance(data, dict):
return None
return data
async def _api_lyric_track(
self,
client: httpx.AsyncClient,
track: TrackMeta,
limit: int,
) -> dict | None:
"""Request lyric payload for top-ranked candidate of a track."""
search_data = await self._api_search_track(client, track, limit)
if search_data is None:
return None
candidates = _parse_netease_search(search_data)
if not candidates:
return None
ranked = select_ranked(
candidates,
track.length,
title=track.title,
artist=track.artist,
album=track.album,
)
if not ranked:
return None
top_song_id = ranked[0][0]
return await self._api_lyric(client, top_song_id)
async def _search(
self, track: TrackMeta, limit: int = 10
) -> list[tuple[int, float]]:
@@ -49,46 +167,18 @@ class NeteaseFetcher(BaseFetcher):
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
resp = await client.post(
_NETEASE_SEARCH_URL,
headers=_NETEASE_BASE_HEADERS,
data={"s": query, "type": "1", "limit": str(limit), "offset": "0"},
)
resp.raise_for_status()
result = resp.json()
result = await self._api_search_track(client, track, limit)
if not isinstance(result, dict):
logger.error(
f"Netease: search returned non-dict: {type(result).__name__}"
)
if result is None:
logger.error("Netease: search returned non-dict payload")
return []
result_body = result.get("result")
if not isinstance(result_body, dict):
logger.debug("Netease: search 'result' field missing or invalid")
return []
songs = result_body.get("songs")
if not isinstance(songs, list) or len(songs) == 0:
candidates = _parse_netease_search(result)
if not candidates:
logger.debug("Netease: search returned 0 results")
return []
logger.debug(f"Netease: search returned {len(songs)} candidates")
candidates = [
SearchCandidate(
item=song_id,
duration_ms=float(song["dt"])
if isinstance(song.get("dt"), int)
else None,
title=song.get("name"),
artist=", ".join(a.get("name", "") for a in song.get("ar", []))
or None,
album=(song.get("al") or {}).get("name"),
)
for song in songs
if isinstance(song, dict) and isinstance(song_id := song.get("id"), int)
]
logger.debug(f"Netease: search returned {len(candidates)} candidates")
ranked = select_ranked(
candidates,
track.length,
@@ -114,43 +204,16 @@ class NeteaseFetcher(BaseFetcher):
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
resp = await client.post(
_NETEASE_LYRIC_URL,
headers=_NETEASE_BASE_HEADERS,
data={
"id": str(song_id),
"cp": "false",
"tv": "0",
"lv": "0",
"rv": "0",
"kv": "0",
"yv": "0",
"ytv": "0",
"yrv": "0",
},
)
resp.raise_for_status()
data = resp.json()
data = await self._api_lyric(client, song_id)
if not isinstance(data, dict):
logger.error(
f"Netease: lyric response is not dict: {type(data).__name__}"
)
if data is None:
logger.error("Netease: lyric response is not dict")
return FetchResult.from_network_error()
lrc_obj = data.get("lrc")
if not isinstance(lrc_obj, dict):
logger.debug(
f"Netease: no 'lrc' object in response for song_id={song_id}"
)
return FetchResult.from_not_found()
lrc: str = lrc_obj.get("lyric", "")
if not isinstance(lrc, str) or not lrc.strip():
lrcdata = _parse_netease_lyrics(data)
if lrcdata is None:
logger.debug(f"Netease: empty lyrics for song_id={song_id}")
return FetchResult.from_not_found()
lrcdata = LRCData(lrc)
status = lrcdata.detect_sync_status()
logger.info(
f"Netease: got {status.value} lyrics for song_id={song_id} "
src/lrx_cli/fetchers/qqmusic.py
+96 -60
View File
@@ -10,11 +10,11 @@ Description: QQ Music fetcher via self-hosted API proxy.
"""
import asyncio
import httpx
from loguru import logger
from .base import BaseFetcher, FetchResult
from .selection import SearchCandidate, select_ranked
from ..authenticators import QQMusicAuthenticator
from ..models import TrackMeta, LyricResult, CacheStatus
from ..lrc import LRCData
from ..config import (
@@ -23,9 +23,40 @@ from ..config import (
MULTI_CANDIDATE_DELAY_S,
)
_QQ_MUSIC_API_SEARCH_ENDPOINT = "/api/search"
_QQ_MUSIC_API_LYRIC_ENDPOINT = "/api/lyric"
from ..authenticators import QQMusicAuthenticator
def _parse_qq_search(data: dict) -> list[SearchCandidate[str]]:
"""Parse QQMusic search response into normalized candidates."""
if data.get("code") != 0:
return []
songs = data.get("data", {}).get("list", [])
if not isinstance(songs, list):
return []
return [
SearchCandidate(
item=mid,
duration_ms=float(song["interval"]) * 1000
if isinstance(song.get("interval"), int)
else None,
title=song.get("name"),
artist=", ".join(s.get("name", "") for s in song.get("singer", [])) or None,
album=(song.get("album") or {}).get("name"),
)
for song in songs
if isinstance(song, dict) and isinstance(mid := song.get("mid"), str)
]
def _parse_qq_lyrics(data: dict) -> LRCData | None:
"""Parse QQMusic lyric response to LRCData."""
if data.get("code") != 0:
return None
lrc = data.get("data", {}).get("lyric", "")
if not isinstance(lrc, str) or not lrc.strip():
return None
return LRCData(lrc)
class QQMusicFetcher(BaseFetcher):
@@ -41,49 +72,73 @@ class QQMusicFetcher(BaseFetcher):
def is_available(self, track: TrackMeta) -> bool:
return bool(track.title) and self._auth.is_configured()
async def _api_search(
self,
track: TrackMeta,
limit: int,
) -> dict | None:
"""Return raw QQMusic search payload for one track."""
query = f"{track.artist or ''} {track.title or ''}".strip()
if not query:
return None
data = await self._auth.search(query, limit)
if not isinstance(data, dict):
return None
return data
async def _api_lyric(
self,
mid: str,
) -> dict | None:
"""Return raw QQMusic lyric payload for one song MID."""
data = await self._auth.get_lyric(mid)
if not isinstance(data, dict):
return None
return data
async def _api_lyric_track(
self,
track: TrackMeta,
limit: int,
) -> dict | None:
"""Return raw QQMusic lyric payload for top-ranked search candidate."""
search_data = await self._api_search(track, limit)
if search_data is None:
return None
candidates = _parse_qq_search(search_data)
if not candidates:
return None
ranked = select_ranked(
candidates,
track.length,
title=track.title,
artist=track.artist,
album=track.album,
)
if not ranked:
return None
mid = ranked[0][0]
return await self._api_lyric(mid)
async def _search(
self, track: TrackMeta, limit: int = 10
) -> list[tuple[str, float]]:
query = f"{track.artist or ''} {track.title or ''}".strip()
if not query:
search_data = await self._api_search(track, limit)
if search_data is None:
return []
query = f"{track.artist or ''} {track.title or ''}".strip()
logger.debug(f"QQMusic: searching for '{query}' (limit={limit})")
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
resp = await client.get(
f"{await self._auth.authenticate()}{_QQ_MUSIC_API_SEARCH_ENDPOINT}",
params={"keyword": query, "type": "song", "num": limit},
)
resp.raise_for_status()
data = resp.json()
if data.get("code") != 0:
logger.error(f"QQMusic: search API error: {data}")
return []
songs = data.get("data", {}).get("list", [])
if not songs:
candidates = _parse_qq_search(search_data)
if not candidates:
logger.debug("QQMusic: search returned 0 results")
return []
logger.debug(f"QQMusic: search returned {len(songs)} candidates")
candidates = [
SearchCandidate(
item=mid,
duration_ms=float(song["interval"]) * 1000
if isinstance(song.get("interval"), int)
else None,
title=song.get("name"),
artist=", ".join(s.get("name", "") for s in song.get("singer", []))
or None,
album=(song.get("album") or {}).get("name"),
)
for song in songs
if isinstance(song, dict) and isinstance(mid := song.get("mid"), str)
]
logger.debug(f"QQMusic: search returned {len(candidates)} candidates")
ranked = select_ranked(
candidates,
track.length,
@@ -100,32 +155,17 @@ class QQMusicFetcher(BaseFetcher):
logger.debug("QQMusic: no suitable candidate found")
return ranked
except Exception as e:
logger.error(f"QQMusic: search failed: {e}")
return []
async def _get_lyric(self, mid: str, confidence: float = 0.0) -> FetchResult:
logger.debug(f"QQMusic: fetching lyrics for mid={mid}")
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
resp = await client.get(
f"{await self._auth.authenticate()}{_QQ_MUSIC_API_LYRIC_ENDPOINT}",
params={"mid": mid},
)
resp.raise_for_status()
data = resp.json()
if data.get("code") != 0:
logger.error(f"QQMusic: lyric API error: {data}")
data = await self._api_lyric(mid)
if data is None:
return FetchResult.from_network_error()
lrc = data.get("data", {}).get("lyric", "")
if not isinstance(lrc, str) or not lrc.strip():
lrcdata = _parse_qq_lyrics(data)
if lrcdata is None:
logger.debug(f"QQMusic: empty lyrics for mid={mid}")
return FetchResult.from_not_found()
lrcdata = LRCData(lrc)
status = lrcdata.detect_sync_status()
logger.info(
f"QQMusic: got {status.value} lyrics for mid={mid} ({len(lrcdata)} lines)"
@@ -151,10 +191,6 @@ class QQMusicFetcher(BaseFetcher):
),
)
except Exception as e:
logger.error(f"QQMusic: lyric fetch failed for mid={mid}: {e}")
return FetchResult.from_network_error()
async def fetch(self, track: TrackMeta, bypass_cache: bool = False) -> FetchResult:
if not self._auth.is_configured():
logger.debug("QQMusic: skipped — Auth not configured")
src/lrx_cli/fetchers/spotify.py
+70 -84
View File
@@ -4,16 +4,66 @@ Date: 2026-03-25 10:43:21
Description: Spotify fetcher — obtains synced lyrics via Spotify's internal color-lyrics API.
"""
import httpx
from loguru import logger
from .base import BaseFetcher, FetchResult
from ..authenticators.spotify import SpotifyAuthenticator, SPOTIFY_BASE_HEADERS
from ..authenticators.spotify import SpotifyAuthenticator
from ..models import TrackMeta, LyricResult, CacheStatus
from ..lrc import LRCData
from ..config import GeneralConfig, TTL_NOT_FOUND
_SPOTIFY_LYRICS_URL = "https://spclient.wg.spotify.com/color-lyrics/v2/track/"
def _format_lrc_line(start_ms: int, words: str) -> str:
minutes = start_ms // 60000
seconds = (start_ms // 1000) % 60
centiseconds = round((start_ms % 1000) / 10.0)
return f"[{minutes:02d}:{seconds:02d}.{centiseconds:02.0f}]{words}"
def _is_truly_synced(lines: list[dict]) -> bool:
for line in lines:
try:
ms = int(line.get("startTimeMs", "0"))
if ms > 0:
return True
except (ValueError, TypeError):
continue
return False
def _parse_spotify_lyrics(data: dict) -> LRCData | None:
"""Parse Spotify color-lyrics payload to LRCData."""
lyrics_data = data.get("lyrics")
if not isinstance(lyrics_data, dict):
return None
sync_type = lyrics_data.get("syncType", "")
lines = lyrics_data.get("lines", [])
if not isinstance(lines, list) or len(lines) == 0:
return None
is_synced = sync_type == "LINE_SYNCED" and _is_truly_synced(lines)
lrc_lines: list[str] = []
for line in lines:
if not isinstance(line, dict):
continue
words = line.get("words", "")
if not isinstance(words, str):
continue
try:
ms = int(line.get("startTimeMs", "0"))
except (ValueError, TypeError):
ms = 0
if is_synced:
lrc_lines.append(_format_lrc_line(ms, words))
else:
lrc_lines.append(f"[00:00.00]{words}")
if not lrc_lines:
return None
return LRCData("\n".join(lrc_lines))
class SpotifyFetcher(BaseFetcher):
@@ -29,23 +79,14 @@ class SpotifyFetcher(BaseFetcher):
def is_available(self, track: TrackMeta) -> bool:
return bool(track.trackid) and self._auth.is_configured()
@staticmethod
def _format_lrc_line(start_ms: int, words: str) -> str:
minutes = start_ms // 60000
seconds = (start_ms // 1000) % 60
centiseconds = round((start_ms % 1000) / 10.0)
return f"[{minutes:02d}:{seconds:02d}.{centiseconds:02.0f}]{words}"
@staticmethod
def _is_truly_synced(lines: list[dict]) -> bool:
for line in lines:
try:
ms = int(line.get("startTimeMs", "0"))
if ms > 0:
return True
except (ValueError, TypeError):
continue
return False
async def _api_lyrics(self, track: TrackMeta) -> dict | None:
"""Return raw Spotify lyrics payload for one track using production auth path."""
if not track.trackid:
return None
data = await self._auth.get_lyrics(track.trackid)
if not isinstance(data, dict):
return None
return data
async def fetch(self, track: TrackMeta, bypass_cache: bool = False) -> FetchResult:
if not track.trackid:
@@ -54,71 +95,20 @@ class SpotifyFetcher(BaseFetcher):
logger.info(f"Spotify: fetching lyrics for trackid={track.trackid}")
token = await self._auth.authenticate()
if not token:
logger.error("Spotify: cannot fetch lyrics without a token")
return FetchResult.from_network_error()
url = f"{_SPOTIFY_LYRICS_URL}{track.trackid}?format=json&vocalRemoval=false&market=from_token"
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {token}",
**SPOTIFY_BASE_HEADERS,
}
try:
async with httpx.AsyncClient(timeout=self._general.http_timeout) as client:
res = await client.get(url, headers=headers)
if res.status_code == 404:
logger.debug(f"Spotify: 404 for trackid={track.trackid}")
data = await self._api_lyrics(track)
if data is None:
logger.debug(f"Spotify: no lyrics payload for trackid={track.trackid}")
return FetchResult.from_not_found()
if res.status_code != 200:
logger.error(f"Spotify: lyrics API returned {res.status_code}")
return FetchResult.from_network_error()
data = res.json()
if not isinstance(data, dict) or "lyrics" not in data:
logger.error("Spotify: unexpected lyrics response structure")
return FetchResult.from_network_error()
lyrics_data = data["lyrics"]
sync_type = lyrics_data.get("syncType", "")
lines = lyrics_data.get("lines", [])
if not isinstance(lines, list) or len(lines) == 0:
logger.debug("Spotify: response contained no lyric lines")
content = _parse_spotify_lyrics(data)
if content is None:
logger.debug("Spotify: response contained no parseable lyric lines")
return FetchResult.from_not_found()
is_synced = sync_type == "LINE_SYNCED" and self._is_truly_synced(lines)
lrc_lines: list[str] = []
for line in lines:
words = line.get("words", "")
if not isinstance(words, str):
continue
try:
ms = int(line.get("startTimeMs", "0"))
except (ValueError, TypeError):
ms = 0
if is_synced:
lrc_lines.append(self._format_lrc_line(ms, words))
else:
lrc_lines.append(f"[00:00.00]{words}")
content = LRCData("\n".join(lrc_lines))
status = (
CacheStatus.SUCCESS_SYNCED
if is_synced
else CacheStatus.SUCCESS_UNSYNCED
)
logger.info(f"Spotify: got {status.value} lyrics ({len(lrc_lines)} lines)")
status = content.detect_sync_status()
logger.info(f"Spotify: got {status.value} lyrics ({len(content)} lines)")
not_found = LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
if is_synced:
if status == CacheStatus.SUCCESS_SYNCED:
return FetchResult(
synced=LyricResult(
status=CacheStatus.SUCCESS_SYNCED,
@@ -135,7 +125,3 @@ class SpotifyFetcher(BaseFetcher):
source=self.source_name,
),
)
except Exception as e:
logger.error(f"Spotify: lyrics fetch failed: {e}")
return FetchResult.from_network_error()
tests/fixtures/fetchers/lrclib_response.json
Vendored
+4
View File
@@ -0,0 +1,4 @@
{
"syncedLyrics": "[00:01.00]s1\n[00:02.00]s2",
"plainLyrics": "p1\np2"
}
tests/fixtures/fetchers/lrclib_search_results.json
Vendored
+20
View File
@@ -0,0 +1,20 @@
[
{
"id": 1,
"trackName": "My Love",
"artistName": "Westlife",
"albumName": "Coast To Coast",
"duration": 231.847,
"syncedLyrics": "[00:01.00]hello",
"plainLyrics": "hello"
},
{
"id": 2,
"trackName": "My Love (Live)",
"artistName": "Westlife",
"albumName": "Live",
"duration": 262.0,
"syncedLyrics": "",
"plainLyrics": "hello"
}
]
tests/fixtures/fetchers/musixmatch_macro_richsync.json
Vendored
+28
View File
@@ -0,0 +1,28 @@
{
"message": {
"body": {
"macro_calls": {
"track.richsync.get": {
"message": {
"header": {
"status_code": 200
},
"body": {
"richsync": {
"richsync_body": "[{\"ts\": 1.2, \"x\": \"hello\"}, {\"ts\": 2.34, \"x\": \"world\"}]"
}
}
}
},
"track.subtitles.get": {
"message": {
"header": {
"status_code": 404
},
"body": {}
}
}
}
}
}
}
tests/fixtures/fetchers/musixmatch_macro_subtitle.json
Vendored
+32
View File
@@ -0,0 +1,32 @@
{
"message": {
"body": {
"macro_calls": {
"track.richsync.get": {
"message": {
"header": {
"status_code": 404
},
"body": {}
}
},
"track.subtitles.get": {
"message": {
"header": {
"status_code": 200
},
"body": {
"subtitle_list": [
{
"subtitle": {
"subtitle_body": "[{\"text\": \"hello\", \"time\": {\"total\": 1.1}}, {\"text\": \"world\", \"time\": {\"total\": 2.22}}]"
}
}
]
}
}
}
}
}
}
}
tests/fixtures/fetchers/musixmatch_search.json
Vendored
+20
View File
@@ -0,0 +1,20 @@
{
"message": {
"body": {
"track_list": [
{
"track": {
"commontrack_id": 123,
"track_length": 232,
"has_subtitles": 1,
"has_richsync": 0,
"track_name": "My Love",
"artist_name": "Westlife",
"album_name": "Coast To Coast",
"instrumental": 0
}
}
]
}
}
}
tests/fixtures/fetchers/netease_lyrics.json
Vendored
+5
View File
@@ -0,0 +1,5 @@
{
"lrc": {
"lyric": "[00:01.00]line1\n[00:02.00]line2"
}
}
tests/fixtures/fetchers/netease_search.json
Vendored
+32
View File
@@ -0,0 +1,32 @@
{
"result": {
"songs": [
{
"id": 2080607,
"name": "My Love",
"dt": 231941,
"ar": [
{
"name": "Westlife"
}
],
"al": {
"name": "Unbreakable"
}
},
{
"id": 572412968,
"name": "My Love",
"dt": 231000,
"ar": [
{
"name": "Westlife"
}
],
"al": {
"name": "Pure... Love"
}
}
]
}
}
tests/fixtures/fetchers/qq_lyrics.json
Vendored
+6
View File
@@ -0,0 +1,6 @@
{
"code": 0,
"data": {
"lyric": "[00:01.00]hello\n[00:02.00]world"
}
}
tests/fixtures/fetchers/qq_search.json
Vendored
+33
View File
@@ -0,0 +1,33 @@
{
"code": 0,
"data": {
"list": [
{
"mid": "mid1",
"interval": 232,
"name": "My Love",
"singer": [
{
"name": "Westlife"
}
],
"album": {
"name": "Coast To Coast"
}
},
{
"mid": "mid2",
"interval": 248,
"name": "My Love (Album Version)",
"singer": [
{
"name": "Little Texas"
}
],
"album": {
"name": "Greatest Hits"
}
}
]
}
}
tests/fixtures/fetchers/spotify_synced.json
Vendored
+9
View File
@@ -0,0 +1,9 @@
{
"lyrics": {
"syncType": "LINE_SYNCED",
"lines": [
{"startTimeMs": "1000", "words": "hello"},
{"startTimeMs": "2500", "words": "world"}
]
}
}
tests/fixtures/fetchers/spotify_unsynced.json
Vendored
+9
View File
@@ -0,0 +1,9 @@
{
"lyrics": {
"syncType": "UNSYNCED",
"lines": [
{"startTimeMs": "0", "words": "plain one"},
{"startTimeMs": "0", "words": "plain two"}
]
}
}
tests/test_fetchers.py
+459 -77
View File
@@ -1,19 +1,40 @@
from dataclasses import replace
import asyncio
import json
from pathlib import Path
from typing import Callable
import httpx
import pytest
from lrx_cli.authenticators import create_authenticators
from lrx_cli.cache import CacheEngine
from lrx_cli.config import AppConfig, load_config
from lrx_cli.core import LrcManager
from lrx_cli.fetchers import FetcherMethodType
from lrx_cli.models import TrackMeta
from tests.marks import (
requires_musixmatch_token,
requires_qq_music,
requires_spotify,
from lrx_cli.fetchers import FetcherMethodType, create_fetchers
from lrx_cli.fetchers.lrclib import LrclibFetcher, _parse_lrclib_response
from lrx_cli.fetchers.lrclib_search import (
LrclibSearchFetcher,
_parse_lrclib_search_results,
)
from lrx_cli.fetchers.musixmatch import (
MusixmatchFetcher,
MusixmatchSpotifyFetcher,
_parse_mxm_macro,
_parse_mxm_search,
)
from lrx_cli.fetchers.netease import (
NeteaseFetcher,
_parse_netease_lyrics,
_parse_netease_search,
)
from lrx_cli.fetchers.qqmusic import QQMusicFetcher, _parse_qq_lyrics, _parse_qq_search
from lrx_cli.fetchers.spotify import SpotifyFetcher, _parse_spotify_lyrics
from lrx_cli.lrc import LRCData
from lrx_cli.models import CacheStatus, TrackMeta
from tests.marks import requires_musixmatch_token, requires_qq_music, requires_spotify
SAMPLE_SPOTIFY_TRACK: TrackMeta = TrackMeta(
SAMPLE_TRACK = TrackMeta(
title="One Last Kiss",
artist="Hikaru Utada",
album="One Last Kiss",
@@ -22,86 +43,152 @@ SAMPLE_SPOTIFY_TRACK: TrackMeta = TrackMeta(
url="https://open.spotify.com/track/5RhWszHMSKzb7KiXk4Ae0M",
)
SAMPLE_SPOTIFY_TRACK_ALBUM_MODIFIED = replace(SAMPLE_SPOTIFY_TRACK, album="BADモード")
SAMPLE_SPOTIFY_TRACK_ARTIST_MODIFIED = replace(
SAMPLE_SPOTIFY_TRACK, artist="宇多田ヒカル"
SAMPLE_TRACK_ALBUM_MODIFIED = replace(SAMPLE_TRACK, album="BADモード")
SAMPLE_TRACK_ARTIST_MODIFIED = replace(SAMPLE_TRACK, artist="宇多田ヒカル")
SAMPLE_TRACK_ALBUM_ARTIST_MODIFIED = replace(
SAMPLE_TRACK,
artist="宇多田ヒカル",
album="BADモード",
)
SAMPLE_SPOTIFY_TRACK_ALBUM_ARTIST_MODIFIED = replace(
SAMPLE_SPOTIFY_TRACK, artist="宇多田ヒカル", album="BADモード"
)
_FIXTURE_DIR = Path(__file__).parent / "fixtures" / "fetchers"
_NETWORK_TIMEOUT = 20.0
ParserFunc = Callable[[dict], LRCData | None]
@pytest.fixture
def lrc_manager(tmp_path: Path) -> LrcManager:
"""LrcManager with empty credentials (no auth required)."""
return LrcManager(str(tmp_path / "cache.db"), AppConfig())
@pytest.fixture
def cred_lrc_manager(tmp_path: Path) -> LrcManager:
"""LrcManager with credentials from config.toml (for CI/network tests)."""
return LrcManager(str(tmp_path / "cache.db"), load_config())
def _fetch_and_assert(
@pytest.fixture
def fetcher_runtime_anonymous(tmp_path: Path):
cfg = AppConfig()
cache = CacheEngine(str(tmp_path / "network-anon-cache.db"))
authenticators = create_authenticators(cache, cfg)
fetchers = create_fetchers(cache, authenticators, cfg)
return fetchers, cfg
@pytest.fixture
def fetcher_runtime_credentialed(tmp_path: Path):
cfg = load_config()
cache = CacheEngine(str(tmp_path / "network-cred-cache.db"))
authenticators = create_authenticators(cache, cfg)
fetchers = create_fetchers(cache, authenticators, cfg)
return fetchers, cfg
def _load_fixture(name: str) -> dict | list:
return json.loads((_FIXTURE_DIR / name).read_text(encoding="utf-8"))
def _assert_shape(actual: object, fixture: object) -> None:
"""Assert actual payload contains fixture structure recursively.
- dict: all fixture keys must exist with matching nested shape
- list: actual must contain at least fixture length and each indexed shape must match
- scalar: runtime type must match fixture type
"""
if isinstance(fixture, dict):
assert isinstance(actual, dict)
for key, value in fixture.items():
assert key in actual
_assert_shape(actual[key], value)
return
if isinstance(fixture, list):
assert isinstance(actual, list)
assert len(actual) >= len(fixture)
for idx, value in enumerate(fixture):
_assert_shape(actual[idx], value)
return
if fixture is None:
return
assert isinstance(actual, type(fixture))
def _fetch_with_method(
lrc_manager: LrcManager,
method: FetcherMethodType,
expect_fail: bool = False,
bypass_cache: bool = True,
) -> None:
result = lrc_manager.fetch_for_track(
SAMPLE_SPOTIFY_TRACK, force_method=method, bypass_cache=bypass_cache
*,
bypass_cache: bool = False,
):
return lrc_manager.fetch_for_track(
SAMPLE_TRACK,
force_method=method,
bypass_cache=bypass_cache,
)
if expect_fail:
# Cache-search fetcher behavior
def test_cache_search_no_cache_fails(lrc_manager: LrcManager):
result = _fetch_with_method(lrc_manager, "cache-search", bypass_cache=False)
assert result is None
else:
def test_cache_search_exact_hit(lrc_manager: LrcManager):
expected = "[00:00.01]lyrics"
lrc_manager.manual_insert(SAMPLE_TRACK, expected)
result = lrc_manager.fetch_for_track(
SAMPLE_TRACK,
force_method="cache-search",
bypass_cache=False,
)
assert result is not None
assert result.status == "SUCCESS_SYNCED"
assert result.lyrics is not None
def test_cache_search_fetcher_without_cache(lrc_manager: LrcManager):
_fetch_and_assert(lrc_manager, "cache-search", expect_fail=True, bypass_cache=False)
assert result.lyrics.to_text() == expected
@pytest.mark.parametrize(
"query_track",
[
pytest.param(SAMPLE_SPOTIFY_TRACK, id="exact_match"),
pytest.param(SAMPLE_SPOTIFY_TRACK_ARTIST_MODIFIED, id="artist_modified"),
pytest.param(SAMPLE_SPOTIFY_TRACK_ALBUM_MODIFIED, id="album_modified"),
pytest.param(
SAMPLE_SPOTIFY_TRACK_ALBUM_ARTIST_MODIFIED, id="album_artist_modified"
),
pytest.param(SAMPLE_TRACK_ARTIST_MODIFIED, id="artist_modified"),
pytest.param(SAMPLE_TRACK_ALBUM_MODIFIED, id="album_modified"),
pytest.param(SAMPLE_TRACK_ALBUM_ARTIST_MODIFIED, id="album_artist_modified"),
],
)
def test_cache_search_fetcher_with_fuzzy_metadata(
lrc_manager: LrcManager, query_track: TrackMeta
):
expected_lrc = "[00:00.01]lyrics"
lrc_manager.manual_insert(SAMPLE_SPOTIFY_TRACK, expected_lrc)
def test_cache_search_fuzzy_hit(lrc_manager: LrcManager, query_track: TrackMeta):
expected = "[00:00.01]lyrics"
lrc_manager.manual_insert(SAMPLE_TRACK, expected)
result = lrc_manager.fetch_for_track(
query_track, force_method="cache-search", bypass_cache=False
query_track,
force_method="cache-search",
bypass_cache=False,
)
assert result is not None
assert result.lyrics is not None
assert result.lyrics.to_text() == expected_lrc
assert result.lyrics.to_text() == expected
def test_cache_search_fetcher_prefer_better_match(lrc_manager: LrcManager):
def test_cache_search_prefer_better_match(lrc_manager: LrcManager):
lrc_manager.manual_insert(
SAMPLE_SPOTIFY_TRACK_ARTIST_MODIFIED, "[00:00.01]artist modified"
SAMPLE_TRACK_ARTIST_MODIFIED,
"[00:00.01]artist modified",
)
lrc_manager.manual_insert(
SAMPLE_SPOTIFY_TRACK_ALBUM_ARTIST_MODIFIED, "[00:00.01]artist+album modified"
SAMPLE_TRACK_ALBUM_ARTIST_MODIFIED,
"[00:00.01]artist+album modified",
)
result = lrc_manager.fetch_for_track(
SAMPLE_SPOTIFY_TRACK, force_method="cache-search", bypass_cache=False
SAMPLE_TRACK,
force_method="cache-search",
bypass_cache=False,
)
assert result is not None
@@ -109,52 +196,347 @@ def test_cache_search_fetcher_prefer_better_match(lrc_manager: LrcManager):
assert result.lyrics.to_text() == "[00:00.01]artist modified"
# API response format for every fetcher
@pytest.mark.network
@pytest.mark.parametrize(
"method, expect_fail",
[
("lrclib", False),
("lrclib-search", False),
("netease", False),
("spotify", True), # requires auth
("qqmusic", True), # requires api
],
)
def test_anonymous_remote_fetchers(
lrc_manager: LrcManager,
method: FetcherMethodType,
expect_fail: bool,
):
_fetch_and_assert(lrc_manager, method, expect_fail)
def test_api_lrclib_response_shape(fetcher_runtime_anonymous):
fetchers, _cfg = fetcher_runtime_anonymous
fetcher = fetchers["lrclib"]
assert isinstance(fetcher, LrclibFetcher)
async def _run() -> dict:
async with httpx.AsyncClient(timeout=_NETWORK_TIMEOUT) as client:
response = await fetcher._api_get(client, SAMPLE_TRACK)
assert response.status_code == 200
payload = response.json()
assert isinstance(payload, dict)
return payload
payload = asyncio.run(_run())
_assert_shape(payload, _load_fixture("lrclib_response.json"))
@pytest.mark.network
def test_api_lrclib_search_response_shape(fetcher_runtime_anonymous):
fetchers, _cfg = fetcher_runtime_anonymous
fetcher = fetchers["lrclib-search"]
assert isinstance(fetcher, LrclibSearchFetcher)
async def _run() -> list[dict]:
async with httpx.AsyncClient(timeout=_NETWORK_TIMEOUT) as client:
items, had_error = await fetcher._api_candidates(client, SAMPLE_TRACK)
assert had_error is False
return items
payload = asyncio.run(_run())
_assert_shape(payload, _load_fixture("lrclib_search_results.json"))
@pytest.mark.network
def test_api_netease_response_shape(fetcher_runtime_anonymous):
fetchers, _cfg = fetcher_runtime_anonymous
fetcher = fetchers["netease"]
assert isinstance(fetcher, NeteaseFetcher)
async def _run() -> tuple[dict, dict]:
async with httpx.AsyncClient(timeout=_NETWORK_TIMEOUT) as client:
search = await fetcher._api_search_track(client, SAMPLE_TRACK, 5)
lyric = await fetcher._api_lyric_track(client, SAMPLE_TRACK, 5)
assert isinstance(search, dict)
assert isinstance(lyric, dict)
return search, lyric
search_payload, lyric_payload = asyncio.run(_run())
_assert_shape(search_payload, _load_fixture("netease_search.json"))
_assert_shape(lyric_payload, _load_fixture("netease_lyrics.json"))
@pytest.mark.network
@requires_spotify
def test_spotify_fetcher(cred_lrc_manager: LrcManager):
_fetch_and_assert(cred_lrc_manager, "spotify")
def test_api_spotify_response_shape(fetcher_runtime_credentialed):
fetchers, _cfg = fetcher_runtime_credentialed
fetcher = fetchers["spotify"]
assert isinstance(fetcher, SpotifyFetcher)
async def _run() -> dict:
payload = await fetcher._api_lyrics(SAMPLE_TRACK)
assert isinstance(payload, dict)
return payload
payload = asyncio.run(_run())
_assert_shape(payload, _load_fixture("spotify_synced.json"))
@pytest.mark.network
@requires_qq_music
def test_qqmusic_fetcher(cred_lrc_manager: LrcManager):
_fetch_and_assert(cred_lrc_manager, "qqmusic")
def test_api_qqmusic_response_shape(fetcher_runtime_credentialed):
fetchers, _cfg = fetcher_runtime_credentialed
fetcher = fetchers["qqmusic"]
assert isinstance(fetcher, QQMusicFetcher)
async def _run() -> tuple[dict, dict]:
search = await fetcher._api_search(SAMPLE_TRACK, 10)
lyric = await fetcher._api_lyric_track(SAMPLE_TRACK, 10)
assert isinstance(search, dict)
assert isinstance(lyric, dict)
return search, lyric
search_payload, lyric_payload = asyncio.run(_run())
_assert_shape(search_payload, _load_fixture("qq_search.json"))
_assert_shape(lyric_payload, _load_fixture("qq_lyrics.json"))
@pytest.mark.network
def test_musixmatch_anonymous_fetcher(lrc_manager: LrcManager):
# These fetchers should be tested in a single test to share the same usertoken
# Otherwise the second may fail due to rate limits
_fetch_and_assert(lrc_manager, "musixmatch", expect_fail=False)
_fetch_and_assert(lrc_manager, "musixmatch-spotify", expect_fail=False)
def test_api_musixmatch_anonymous_response_shape(fetcher_runtime_anonymous):
"""Anonymous musixmatch calls must share one cache/auth context in this test."""
fetchers, _cfg = fetcher_runtime_anonymous
search_fetcher = fetchers["musixmatch"]
spotify_fetcher = fetchers["musixmatch-spotify"]
assert isinstance(search_fetcher, MusixmatchFetcher)
assert isinstance(spotify_fetcher, MusixmatchSpotifyFetcher)
async def _run() -> tuple[dict, dict, dict]:
search = await search_fetcher._api_search_track(SAMPLE_TRACK)
macro_from_search = await search_fetcher._api_macro_track(SAMPLE_TRACK)
macro_from_spotify = await spotify_fetcher._api_macro_track(SAMPLE_TRACK)
assert isinstance(search, dict)
assert isinstance(macro_from_search, dict)
assert isinstance(macro_from_spotify, dict)
return search, macro_from_search, macro_from_spotify
search_payload, macro_payload, spotify_macro_payload = asyncio.run(_run())
_assert_shape(search_payload, _load_fixture("musixmatch_search.json"))
_assert_shape(macro_payload, _load_fixture("musixmatch_macro_richsync.json"))
_assert_shape(
spotify_macro_payload, _load_fixture("musixmatch_macro_richsync.json")
)
@pytest.mark.network
@requires_musixmatch_token
def test_musixmatch_fetcher(cred_lrc_manager: LrcManager):
_fetch_and_assert(cred_lrc_manager, "musixmatch")
_fetch_and_assert(cred_lrc_manager, "musixmatch-spotify")
def test_api_musixmatch_token_response_shape(fetcher_runtime_credentialed):
fetchers, _cfg = fetcher_runtime_credentialed
search_fetcher = fetchers["musixmatch"]
spotify_fetcher = fetchers["musixmatch-spotify"]
assert isinstance(search_fetcher, MusixmatchFetcher)
assert isinstance(spotify_fetcher, MusixmatchSpotifyFetcher)
async def _run() -> tuple[dict, dict, dict]:
search = await search_fetcher._api_search_track(SAMPLE_TRACK)
macro_from_search = await search_fetcher._api_macro_track(SAMPLE_TRACK)
macro_from_spotify = await spotify_fetcher._api_macro_track(SAMPLE_TRACK)
assert isinstance(search, dict)
assert isinstance(macro_from_search, dict)
assert isinstance(macro_from_spotify, dict)
return search, macro_from_search, macro_from_spotify
search_payload, macro_payload, spotify_macro_payload = asyncio.run(_run())
_assert_shape(search_payload, _load_fixture("musixmatch_search.json"))
_assert_shape(macro_payload, _load_fixture("musixmatch_macro_richsync.json"))
_assert_shape(
spotify_macro_payload, _load_fixture("musixmatch_macro_richsync.json")
)
def test_local_fetcher(lrc_manager: LrcManager):
# Since this not a local track
_fetch_and_assert(lrc_manager, "local", True)
# Parse fixture JSON into real data structures
@pytest.mark.parametrize(
"fixture_name,parser,expected_status",
[
pytest.param(
"spotify_synced.json",
_parse_spotify_lyrics,
"SUCCESS_SYNCED",
id="spotify-synced",
),
pytest.param(
"spotify_unsynced.json",
_parse_spotify_lyrics,
"SUCCESS_UNSYNCED",
id="spotify-unsynced",
),
],
)
def test_parse_spotify_fixture(
fixture_name: str,
parser: ParserFunc,
expected_status: str,
):
payload = _load_fixture(fixture_name)
assert isinstance(payload, dict)
parsed = parser(payload)
assert parsed is not None
assert parsed.detect_sync_status().value == expected_status
if expected_status == "SUCCESS_SYNCED":
assert parsed.to_text() == "[00:01.00]hello\n[00:02.50]world"
else:
assert parsed.to_text() == "[00:00.00]plain one\n[00:00.00]plain two"
def test_parse_qq_search_fixture() -> None:
payload = _load_fixture("qq_search.json")
assert isinstance(payload, dict)
parsed = _parse_qq_search(payload)
assert len(parsed) == 2
assert parsed[0].item == "mid1"
assert parsed[0].title == "My Love"
assert parsed[0].artist == "Westlife"
assert parsed[0].duration_ms == 232000.0
assert parsed[0].album == "Coast To Coast"
assert parsed[1].item == "mid2"
assert parsed[1].title == "My Love (Album Version)"
assert parsed[1].artist == "Little Texas"
assert parsed[1].duration_ms == 248000.0
assert parsed[1].album == "Greatest Hits"
def test_parse_qq_lyrics_fixture() -> None:
payload = _load_fixture("qq_lyrics.json")
assert isinstance(payload, dict)
parsed = _parse_qq_lyrics(payload)
assert parsed is not None
assert len(parsed) == 2
assert parsed.detect_sync_status() == CacheStatus.SUCCESS_SYNCED
def test_parse_lrclib_response_fixture() -> None:
payload = _load_fixture("lrclib_response.json")
assert isinstance(payload, dict)
parsed = _parse_lrclib_response(payload)
assert parsed.synced is not None and parsed.synced.lyrics is not None
assert parsed.unsynced is not None and parsed.unsynced.lyrics is not None
assert parsed.synced.status == CacheStatus.SUCCESS_SYNCED
assert parsed.unsynced.status == CacheStatus.SUCCESS_UNSYNCED
assert parsed.synced.lyrics.to_text() == "[00:01.00]s1\n[00:02.00]s2"
assert parsed.unsynced.lyrics.to_text() == "[00:00.00]p1\n[00:00.00]p2"
def test_parse_lrclib_search_results_fixture() -> None:
payload = _load_fixture("lrclib_search_results.json")
assert isinstance(payload, list)
parsed = _parse_lrclib_search_results(payload)
assert len(parsed) == 2
assert parsed[0].item.get("id") == 1
assert parsed[0].duration_ms == 231847.0
assert parsed[0].is_synced is True
assert parsed[0].title == "My Love"
assert parsed[0].artist == "Westlife"
assert parsed[0].album == "Coast To Coast"
assert parsed[1].item.get("id") == 2
assert parsed[1].duration_ms == 262000.0
assert parsed[1].is_synced is False
assert parsed[1].title == "My Love (Live)"
assert parsed[1].artist == "Westlife"
assert parsed[1].album == "Live"
def test_parse_netease_search_fixture() -> None:
payload = _load_fixture("netease_search.json")
assert isinstance(payload, dict)
parsed = _parse_netease_search(payload)
assert len(parsed) == 2
assert parsed[0].item == 2080607
assert parsed[0].title == "My Love"
assert parsed[0].artist == "Westlife"
assert parsed[0].duration_ms == 231941.0
assert parsed[0].album == "Unbreakable"
assert parsed[1].item == 572412968
assert parsed[1].artist == "Westlife"
assert parsed[1].duration_ms == 231000.0
def test_parse_netease_lyrics_fixture() -> None:
payload = _load_fixture("netease_lyrics.json")
assert isinstance(payload, dict)
parsed = _parse_netease_lyrics(payload)
assert parsed is not None
assert len(parsed) == 2
assert parsed.detect_sync_status() == CacheStatus.SUCCESS_SYNCED
assert parsed.to_text() == "[00:01.00]line1\n[00:02.00]line2"
def test_parse_musixmatch_search_fixture() -> None:
payload = _load_fixture("musixmatch_search.json")
assert isinstance(payload, dict)
parsed = _parse_mxm_search(payload)
assert len(parsed) == 1
assert parsed[0].item == 123
assert parsed[0].is_synced is True
assert parsed[0].title == "My Love"
assert parsed[0].artist == "Westlife"
assert parsed[0].duration_ms == 232000.0
assert parsed[0].album == "Coast To Coast"
def test_parse_musixmatch_macro_fixture() -> None:
payload = _load_fixture("musixmatch_macro_richsync.json")
assert isinstance(payload, dict)
parsed = _parse_mxm_macro(payload)
assert parsed is not None
assert len(parsed) == 2
assert parsed.detect_sync_status() == CacheStatus.SUCCESS_SYNCED
def test_parse_musixmatch_macro_subtitle_fallback_fixture() -> None:
payload = _load_fixture("musixmatch_macro_subtitle.json")
assert isinstance(payload, dict)
parsed = _parse_mxm_macro(payload)
assert parsed is not None
assert len(parsed) == 2
assert parsed.detect_sync_status() == CacheStatus.SUCCESS_SYNCED
assert parsed.to_text() == "[00:01.10]hello\n[00:02.22]world"
# Empty / partial-error response handling
def test_parse_spotify_empty_or_invalid() -> None:
assert _parse_spotify_lyrics({}) is None
assert _parse_spotify_lyrics({"lyrics": {"lines": []}}) is None
def test_parse_qq_search_empty_or_error() -> None:
assert _parse_qq_search({}) == []
assert _parse_qq_search({"code": 1}) == []
assert _parse_qq_search({"code": 0, "data": {"list": []}}) == []
def test_parse_qq_lyrics_empty_or_error() -> None:
assert _parse_qq_lyrics({}) is None
assert _parse_qq_lyrics({"code": 1}) is None
assert _parse_qq_lyrics({"code": 0, "data": {"lyric": ""}}) is None
def test_parse_lrclib_response_empty_or_partial() -> None:
parsed = _parse_lrclib_response({})
assert parsed.synced is not None
assert parsed.unsynced is not None
assert parsed.synced.lyrics is None
assert parsed.unsynced.lyrics is None
parsed_partial = _parse_lrclib_response({"syncedLyrics": "[00:01.00]line"})
assert (
parsed_partial.synced is not None and parsed_partial.synced.lyrics is not None
)
assert parsed_partial.unsynced is not None
def test_parse_netease_empty_or_partial() -> None:
assert _parse_netease_search({}) == []
assert _parse_netease_search({"result": {"songs": []}}) == []
assert _parse_netease_lyrics({}) is None
assert _parse_netease_lyrics({"lrc": {"lyric": ""}}) is None
def test_parse_musixmatch_empty_or_partial() -> None:
assert _parse_mxm_search({}) == []
assert _parse_mxm_search({"message": {"body": {"track_list": []}}}) == []
assert _parse_mxm_macro({}) is None
assert _parse_mxm_macro({"message": {"body": []}}) is None
uv.lock
Generated
+1 -1
View File
@@ -153,7 +153,7 @@ wheels = [
[[package]]
name = "lrx-cli"
version = "0.7.3"
version = "0.7.4"
source = { editable = "." }
dependencies = [
{ name = "cyclopts" },