From 0f1f5b418aa1ca14abfef9c2f79dcf0c953a643c Mon Sep 17 00:00:00 2001 From: Uyanide Date: Fri, 10 Apr 2026 08:45:24 +0200 Subject: [PATCH] test: add api fixtures --- README.md | 7 +- pyproject.toml | 2 +- src/lrx_cli/authenticators/qqmusic.py | 44 ++ src/lrx_cli/authenticators/spotify.py | 33 ++ src/lrx_cli/fetchers/lrclib.py | 91 +-- src/lrx_cli/fetchers/lrclib_search.py | 117 ++-- src/lrx_cli/fetchers/musixmatch.py | 166 ++++-- src/lrx_cli/fetchers/netease.py | 195 ++++--- src/lrx_cli/fetchers/qqmusic.py | 232 ++++---- src/lrx_cli/fetchers/spotify.py | 182 +++--- tests/fixtures/fetchers/lrclib_response.json | 4 + .../fetchers/lrclib_search_results.json | 20 + .../fetchers/musixmatch_macro_richsync.json | 28 + .../fetchers/musixmatch_macro_subtitle.json | 32 ++ .../fixtures/fetchers/musixmatch_search.json | 20 + tests/fixtures/fetchers/netease_lyrics.json | 5 + tests/fixtures/fetchers/netease_search.json | 32 ++ tests/fixtures/fetchers/qq_lyrics.json | 6 + tests/fixtures/fetchers/qq_search.json | 33 ++ tests/fixtures/fetchers/spotify_synced.json | 9 + tests/fixtures/fetchers/spotify_unsynced.json | 9 + tests/test_fetchers.py | 538 +++++++++++++++--- uv.lock | 2 +- 23 files changed, 1324 insertions(+), 483 deletions(-) create mode 100644 tests/fixtures/fetchers/lrclib_response.json create mode 100644 tests/fixtures/fetchers/lrclib_search_results.json create mode 100644 tests/fixtures/fetchers/musixmatch_macro_richsync.json create mode 100644 tests/fixtures/fetchers/musixmatch_macro_subtitle.json create mode 100644 tests/fixtures/fetchers/musixmatch_search.json create mode 100644 tests/fixtures/fetchers/netease_lyrics.json create mode 100644 tests/fixtures/fetchers/netease_search.json create mode 100644 tests/fixtures/fetchers/qq_lyrics.json create mode 100644 tests/fixtures/fetchers/qq_search.json create mode 100644 tests/fixtures/fetchers/spotify_synced.json create mode 100644 tests/fixtures/fetchers/spotify_unsynced.json diff --git a/README.md b/README.md index 98500f6..f29cc4f 100644 --- a/README.md +++ b/README.md @@ -133,7 +133,8 @@ socket_path = "" # Unix socket path; defaults to / 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 diff --git a/pyproject.toml b/pyproject.toml index ee879bd..5e698c7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" diff --git a/src/lrx_cli/authenticators/qqmusic.py b/src/lrx_cli/authenticators/qqmusic.py index e584da4..21caaa6 100644 --- a/src/lrx_cli/authenticators/qqmusic.py +++ b/src/lrx_cli/authenticators/qqmusic.py @@ -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 diff --git a/src/lrx_cli/authenticators/spotify.py b/src/lrx_cli/authenticators/spotify.py index ad41b05..c09617b 100644 --- a/src/lrx_cli/authenticators/spotify.py +++ b/src/lrx_cli/authenticators/spotify.py @@ -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 diff --git a/src/lrx_cli/fetchers/lrclib.py b/src/lrx_cli/fetchers/lrclib.py index 96dcc97..7d7f7d5 100644 --- a/src/lrx_cli/fetchers/lrclib.py +++ b/src/lrx_cli/fetchers/lrclib.py @@ -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 - ) - res_unsynced: LyricResult = LyricResult( - status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND - ) - - 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, + result = _parse_lrclib_response(data) + if result.synced and result.synced.lyrics: + logger.info( + f"LRCLIB: got synced lyrics ({len(result.synced.lyrics)} lines)" ) - - 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, + if result.unsynced and result.unsynced.lyrics: + logger.info( + f"LRCLIB: got unsynced lyrics ({len(result.unsynced.lyrics)} lines)" ) - - return FetchResult(synced=res_synced, unsynced=res_unsynced) + return result except httpx.HTTPError as e: logger.error(f"LRCLIB: HTTP error: {e}") diff --git a/src/lrx_cli/fetchers/lrclib_search.py b/src/lrx_cli/fetchers/lrclib_search.py index 0d54b5a..f8f0e68 100644 --- a/src/lrx_cli/fetchers/lrclib_search.py +++ b/src/lrx_cli/fetchers/lrclib_search.py @@ -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,49 +77,63 @@ class LrclibSearchFetcher(BaseFetcher): return queries + 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: + resp = await client.get(url, headers={"User-Agent": UA_LRX}) + except httpx.HTTPError as e: + logger.error(f"LRCLIB-search: HTTP error: {e}") + return [], True + if resp.status_code != 200: + logger.error(f"LRCLIB-search: API returned {resp.status_code}") + return [], True + data = resp.json() + if not isinstance(data, list): + return [], False + return [item for item in data if isinstance(item, dict)], False + + 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 + for item in items: + item_id = item.get("id") + if item_id is not None and item_id in seen_ids: + continue + 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() - 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]: - url = f"{_LRCLIB_SEARCH_URL}?{urlencode(params)}" - logger.debug(f"LRCLIB-search: query {params}") - try: - resp = await client.get(url, headers={"User-Agent": UA_LRX}) - except httpx.HTTPError as e: - logger.error(f"LRCLIB-search: HTTP error: {e}") - return [], True - if resp.status_code != 200: - logger.error(f"LRCLIB-search: API returned {resp.status_code}") - return [], True - data = resp.json() - if not isinstance(data, list): - return [], False - return [item for item in data if isinstance(item, dict)], False - - all_results = await asyncio.gather(*(_query(p) for p in queries)) - - for items, err in all_results: - if err: - had_error = True - for item in items: - item_id = item.get("id") - if item_id is not None and item_id in seen_ids: - continue - if item_id is not None: - seen_ids.add(item_id) - candidates.append(item) + 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, diff --git a/src/lrx_cli/fetchers/musixmatch.py b/src/lrx_cli/fetchers/musixmatch.py index 96efc7a..cda77b4 100644 --- a/src/lrx_cli/fetchers/musixmatch.py +++ b/src/lrx_cli/fetchers/musixmatch.py @@ -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: diff --git a/src/lrx_cli/fetchers/netease.py b/src/lrx_cli/fetchers/netease.py index 0621298..9f7b193 100644 --- a/src/lrx_cli/fetchers/netease.py +++ b/src/lrx_cli/fetchers/netease.py @@ -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} " diff --git a/src/lrx_cli/fetchers/qqmusic.py b/src/lrx_cli/fetchers/qqmusic.py index 04f7ceb..35d3a4d 100644 --- a/src/lrx_cli/fetchers/qqmusic.py +++ b/src/lrx_cli/fetchers/qqmusic.py @@ -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,119 +72,124 @@ 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: - 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) - ] - ranked = select_ranked( - candidates, - track.length, - title=track.title, - artist=track.artist, - album=track.album, - ) - if ranked: - logger.debug( - "QQMusic: top candidates: " - + ", ".join(f"mid={m} ({c:.0f})" for m, c in ranked) - ) - else: - logger.debug("QQMusic: no suitable candidate found") - return ranked - - except Exception as e: - logger.error(f"QQMusic: search failed: {e}") + candidates = _parse_qq_search(search_data) + if not candidates: + logger.debug("QQMusic: search returned 0 results") return [] + logger.debug(f"QQMusic: search returned {len(candidates)} candidates") + ranked = select_ranked( + candidates, + track.length, + title=track.title, + artist=track.artist, + album=track.album, + ) + if ranked: + logger.debug( + "QQMusic: top candidates: " + + ", ".join(f"mid={m} ({c:.0f})" for m, c in ranked) + ) + else: + logger.debug("QQMusic: no suitable candidate found") + return ranked + async def _get_lyric(self, mid: str, confidence: float = 0.0) -> FetchResult: logger.debug(f"QQMusic: fetching lyrics for mid={mid}") + data = await self._api_lyric(mid) + if data is None: + return FetchResult.from_network_error() - 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() + lrcdata = _parse_qq_lyrics(data) + if lrcdata is None: + logger.debug(f"QQMusic: empty lyrics for mid={mid}") + return FetchResult.from_not_found() - if data.get("code") != 0: - logger.error(f"QQMusic: lyric API error: {data}") - return FetchResult.from_network_error() - - lrc = data.get("data", {}).get("lyric", "") - if not isinstance(lrc, str) or not lrc.strip(): - 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)" - ) - not_found = LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND) - if status == CacheStatus.SUCCESS_SYNCED: - return FetchResult( - synced=LyricResult( - status=CacheStatus.SUCCESS_SYNCED, - lyrics=lrcdata, - source=self.source_name, - confidence=confidence, - ), - unsynced=not_found, - ) + status = lrcdata.detect_sync_status() + logger.info( + f"QQMusic: got {status.value} lyrics for mid={mid} ({len(lrcdata)} lines)" + ) + not_found = LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND) + if status == CacheStatus.SUCCESS_SYNCED: return FetchResult( - synced=not_found, - unsynced=LyricResult( - status=CacheStatus.SUCCESS_UNSYNCED, + synced=LyricResult( + status=CacheStatus.SUCCESS_SYNCED, lyrics=lrcdata, source=self.source_name, confidence=confidence, ), + unsynced=not_found, ) - - except Exception as e: - logger.error(f"QQMusic: lyric fetch failed for mid={mid}: {e}") - return FetchResult.from_network_error() + return FetchResult( + synced=not_found, + unsynced=LyricResult( + status=CacheStatus.SUCCESS_UNSYNCED, + lyrics=lrcdata, + source=self.source_name, + confidence=confidence, + ), + ) async def fetch(self, track: TrackMeta, bypass_cache: bool = False) -> FetchResult: if not self._auth.is_configured(): diff --git a/src/lrx_cli/fetchers/spotify.py b/src/lrx_cli/fetchers/spotify.py index efc032c..1397bdd 100644 --- a/src/lrx_cli/fetchers/spotify.py +++ b/src/lrx_cli/fetchers/spotify.py @@ -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,88 +95,33 @@ 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() + 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() - 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, - } + content = _parse_spotify_lyrics(data) + if content is None: + logger.debug("Spotify: response contained no parseable lyric lines") + return FetchResult.from_not_found() - 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}") - 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") - 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)") - not_found = LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND) - if is_synced: - return FetchResult( - synced=LyricResult( - status=CacheStatus.SUCCESS_SYNCED, - lyrics=content, - source=self.source_name, - ), - unsynced=not_found, - ) + 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 status == CacheStatus.SUCCESS_SYNCED: return FetchResult( - synced=not_found, - unsynced=LyricResult( - status=CacheStatus.SUCCESS_UNSYNCED, + synced=LyricResult( + status=CacheStatus.SUCCESS_SYNCED, lyrics=content, source=self.source_name, ), + unsynced=not_found, ) - - except Exception as e: - logger.error(f"Spotify: lyrics fetch failed: {e}") - return FetchResult.from_network_error() + return FetchResult( + synced=not_found, + unsynced=LyricResult( + status=CacheStatus.SUCCESS_UNSYNCED, + lyrics=content, + source=self.source_name, + ), + ) diff --git a/tests/fixtures/fetchers/lrclib_response.json b/tests/fixtures/fetchers/lrclib_response.json new file mode 100644 index 0000000..13845a7 --- /dev/null +++ b/tests/fixtures/fetchers/lrclib_response.json @@ -0,0 +1,4 @@ +{ + "syncedLyrics": "[00:01.00]s1\n[00:02.00]s2", + "plainLyrics": "p1\np2" +} diff --git a/tests/fixtures/fetchers/lrclib_search_results.json b/tests/fixtures/fetchers/lrclib_search_results.json new file mode 100644 index 0000000..1c2d57a --- /dev/null +++ b/tests/fixtures/fetchers/lrclib_search_results.json @@ -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" + } +] diff --git a/tests/fixtures/fetchers/musixmatch_macro_richsync.json b/tests/fixtures/fetchers/musixmatch_macro_richsync.json new file mode 100644 index 0000000..289908d --- /dev/null +++ b/tests/fixtures/fetchers/musixmatch_macro_richsync.json @@ -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": {} + } + } + } + } + } +} diff --git a/tests/fixtures/fetchers/musixmatch_macro_subtitle.json b/tests/fixtures/fetchers/musixmatch_macro_subtitle.json new file mode 100644 index 0000000..fdb4330 --- /dev/null +++ b/tests/fixtures/fetchers/musixmatch_macro_subtitle.json @@ -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}}]" + } + } + ] + } + } + } + } + } + } +} diff --git a/tests/fixtures/fetchers/musixmatch_search.json b/tests/fixtures/fetchers/musixmatch_search.json new file mode 100644 index 0000000..8b4f940 --- /dev/null +++ b/tests/fixtures/fetchers/musixmatch_search.json @@ -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 + } + } + ] + } + } +} diff --git a/tests/fixtures/fetchers/netease_lyrics.json b/tests/fixtures/fetchers/netease_lyrics.json new file mode 100644 index 0000000..ba5325d --- /dev/null +++ b/tests/fixtures/fetchers/netease_lyrics.json @@ -0,0 +1,5 @@ +{ + "lrc": { + "lyric": "[00:01.00]line1\n[00:02.00]line2" + } +} diff --git a/tests/fixtures/fetchers/netease_search.json b/tests/fixtures/fetchers/netease_search.json new file mode 100644 index 0000000..bca2c3d --- /dev/null +++ b/tests/fixtures/fetchers/netease_search.json @@ -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" + } + } + ] + } +} diff --git a/tests/fixtures/fetchers/qq_lyrics.json b/tests/fixtures/fetchers/qq_lyrics.json new file mode 100644 index 0000000..93a8744 --- /dev/null +++ b/tests/fixtures/fetchers/qq_lyrics.json @@ -0,0 +1,6 @@ +{ + "code": 0, + "data": { + "lyric": "[00:01.00]hello\n[00:02.00]world" + } +} diff --git a/tests/fixtures/fetchers/qq_search.json b/tests/fixtures/fetchers/qq_search.json new file mode 100644 index 0000000..ec62bce --- /dev/null +++ b/tests/fixtures/fetchers/qq_search.json @@ -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" + } + } + ] + } +} diff --git a/tests/fixtures/fetchers/spotify_synced.json b/tests/fixtures/fetchers/spotify_synced.json new file mode 100644 index 0000000..25864a5 --- /dev/null +++ b/tests/fixtures/fetchers/spotify_synced.json @@ -0,0 +1,9 @@ +{ + "lyrics": { + "syncType": "LINE_SYNCED", + "lines": [ + {"startTimeMs": "1000", "words": "hello"}, + {"startTimeMs": "2500", "words": "world"} + ] + } +} diff --git a/tests/fixtures/fetchers/spotify_unsynced.json b/tests/fixtures/fetchers/spotify_unsynced.json new file mode 100644 index 0000000..91b7e26 --- /dev/null +++ b/tests/fixtures/fetchers/spotify_unsynced.json @@ -0,0 +1,9 @@ +{ + "lyrics": { + "syncType": "UNSYNCED", + "lines": [ + {"startTimeMs": "0", "words": "plain one"}, + {"startTimeMs": "0", "words": "plain two"} + ] + } +} diff --git a/tests/test_fetchers.py b/tests/test_fetchers.py index 411aebd..2c9cdd8 100644 --- a/tests/test_fetchers.py +++ b/tests/test_fetchers.py @@ -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: - assert result is None - else: - 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) +# 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 + + +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.lyrics is not None + 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 diff --git a/uv.lock b/uv.lock index 1cf98bf..7fcb34d 100644 --- a/uv.lock +++ b/uv.lock @@ -153,7 +153,7 @@ wheels = [ [[package]] name = "lrx-cli" -version = "0.7.3" +version = "0.7.4" source = { editable = "." } dependencies = [ { name = "cyclopts" },