feat: add ytmusic song lyrics support

Add song lyrics support using timestamped YTMusic lyrics, convert lyrics to LRC, and isolate lyrics parsing into a dedicated module.

Author: cosven

fuo_ytmusic/lyrics.py

@@ -0,0 +1,64 @@
+from typing import Optional
+
+WATCH_NEXT_RENDERER_PATH = (
+    "contents",
+    "singleColumnMusicWatchNextResultsRenderer",
+    "tabbedRenderer",
+    "watchNextTabbedResultsRenderer",
+)
+
+
+def build_watch_playlist_body(video_id: str) -> dict:
+    return {
+        "enablePersistentPlaylistPanel": True,
+        "isAudioOnly": True,
+        "tunerSettingValue": "AUTOMIX_SETTING_NORMAL",
+        "videoId": video_id,
+        "playlistId": f"RDAMVM{video_id}",
+        "watchEndpointMusicSupportedConfigs": {
+            "watchEndpointMusicConfig": {
+                "hasPersistentPlaylistPanel": True,
+                "musicVideoType": "MUSIC_VIDEO_TYPE_ATV",
+            }
+        },
+    }
+
+
+def extract_lyrics_browse_id(watch_response) -> Optional[str]:
+    tabs = watch_response
+    for key in WATCH_NEXT_RENDERER_PATH:
+        tabs = tabs[key]
+    tabs = tabs["tabs"]
+
+    try:
+        browse_id = tabs[1]["tabRenderer"]["endpoint"]["browseEndpoint"]["browseId"]
+    except (IndexError, KeyError, TypeError):
+        return None
+    if not isinstance(browse_id, str):
+        raise TypeError(f"unexpected lyrics browse id type: {type(browse_id)}")
+    return browse_id
+
+
+def timestamped_lyrics_to_lrc(lyrics_payload) -> Optional[str]:
+    lyrics = lyrics_payload["lyrics"]
+    if not isinstance(lyrics, list):
+        return None
+
+    lines = [format_lyric_line(line) for line in lyrics]
+    if not lines:
+        return None
+    return "\n".join(lines)
+
+
+def format_lyric_line(line) -> str:
+    return f"{format_lrc_timestamp(line.start_time)}{line.text}"
+
+
+def format_lrc_timestamp(milliseconds: int) -> str:
+    if not isinstance(milliseconds, int):
+        raise TypeError(f"unexpected lyric start_time type: {type(milliseconds)}")
+    total_ms = max(0, milliseconds)
+    minutes = total_ms // 60000
+    seconds = (total_ms % 60000) // 1000
+    centiseconds = (total_ms % 1000) // 10
+    return f"[{minutes:02d}:{seconds:02d}.{centiseconds:02d}]"

fuo_ytmusic/provider.py

@@ -10,6 +10,7 @@
     BriefVideoModel,
     Collection,
     CollectionType,
+    LyricModel,
     ModelNotFound,
     PlaylistModel,
     ProviderV2,
@@ -118,7 +119,9 @@ def _get_ytdlp_cookiefile_path(self) -> str:
             return ""
         if not self._should_use_ytdlp_cookiefile():
             return ""
-        headerfile_path = getattr(self.service.api, "headerfile_path", None) or HEADER_FILE
+        headerfile_path = (
+            getattr(self.service.api, "headerfile_path", None) or HEADER_FILE
+        )
         cookiefile_path = YtdlpCookiefileManager(headerfile_path).cookiefile_path
         return "" if cookiefile_path is None else str(cookiefile_path)
 
@@ -472,13 +475,29 @@ def song_get(self, identifier):
         # hack(cosven): we use get_watch_playlist to try to get song detail.
         # It works for song like '如愿-王菲'.
         result = self.service.api.get_watch_playlist(identifier)
-        songs = [YtmusicWatchPlaylistSong(**track).v2_model() for track in result["tracks"]]
+        songs = [
+            YtmusicWatchPlaylistSong(**track).v2_model() for track in result["tracks"]
+        ]
         for song in songs:
             if song.identifier == identifier:
                 return song
         # I think this branch should not be reached (in most cases).
         return ModelNotFound(f"song:{identifier} not found")
 
+    def song_get_lyric(self, song: BriefSongProtocol) -> Optional[LyricModel]:
+        try:
+            content = self.service.song_lyrics(song.identifier)
+        except Exception as e:
+            logger.warning("fetch ytmusic lyrics failed for %s: %s", song.identifier, e)
+            raise ProviderIOError(f"get song lyric failed: {e}", provider=self)
+        if not content:
+            return None
+        return LyricModel(
+            identifier=song.identifier,
+            source=self.meta.identifier,
+            content=content,
+        )
+
     def song_list_similar(self, song):
         result = self.service.api.get_watch_playlist(song.identifier)
         songs = [

fuo_ytmusic/service.py

@@ -20,6 +20,11 @@
     update_headerfile_cookie,
 )
 from fuo_ytmusic.helpers import Singleton
+from fuo_ytmusic.lyrics import (
+    build_watch_playlist_body,
+    extract_lyrics_browse_id,
+    timestamped_lyrics_to_lrc,
+)
 from fuo_ytmusic.models import (
     AlbumInfo,
     ArtistInfo,
@@ -216,10 +221,13 @@ class YtmusicService(metaclass=Singleton):
     def __init__(self):
         self._session = requests.Session()
         self._api: Optional[YTMusic] = None
+        self._anonymous_lyrics_api: Optional[YTMusic] = None
+        self._timeout: Optional[int] = None
         self._session.hooks["response"].append(self._do_logging)
 
         self._signature_timestamp = 0
         self._api_lock = threading.Lock()
+        self._anonymous_lyrics_api_lock = threading.Lock()
         self._profile_manager = YtmusicProfileManager(self)
         self._language: Optional[str] = None
 
@@ -245,9 +253,12 @@ def get_signature_timestamp(self):
         return 0
 
     def reinitialize_by_headerfile(self, headerfile=None):
+        self._api = self._create_api(headerfile, self._session)
+
+    def _create_api(self, headerfile, session: requests.Session) -> YTMusic:
         language = self._language or "zh_CN"
         options = dict(
-            requests_session=self._session,
+            requests_session=session,
             language=language,
             oauth_credentials=OAuthCredentials(
                 # In the new version of ytmusicapi, client_id and client_secret
@@ -259,35 +270,62 @@ def reinitialize_by_headerfile(self, headerfile=None):
                     ".apps.googleusercontent.com"
                 ),
                 client_secret="SboVhoG9s0rNafixCSGGKXAT",
-                session=self._session,
+                session=session,
             ),
         )
         # Due to https://github.com/sigma67/ytmusicapi/issues/676,
         # YTMusic does not work in specific cases when auth file is provided.
         # So initialize without auth file when 400 is returned.
         if headerfile is not None and headerfile.exists():
             logger.info("Initializing ytmusic api with headerfile.")
-            self._api = YTMusic(str(headerfile), **options)
-            self._api.set_headerfile_path(headerfile)
+            api = YTMusic(str(headerfile), **options)
+            api.set_headerfile_path(headerfile)
         else:
             logger.info("Initializing ytmusic api with no headerfile.")
-            self._api = YTMusic(**options)
+            api = YTMusic(**options)
+        return api
+
+    def _get_anonymous_lyrics_api(self) -> YTMusic:
+        if self._anonymous_lyrics_api is None:
+            with self._anonymous_lyrics_api_lock:
+                if self._anonymous_lyrics_api is None:
+                    # Headerfile-authenticated clients can get HTTP 400 when
+                    # requesting timestamped lyrics, while anonymous requests
+                    # for the same lyrics browse id succeed.
+                    session = requests.Session()
+                    session.hooks["response"].append(self._do_logging)
+                    session.proxies = dict(self._session.proxies)
+                    if self._timeout is not None:
+                        session.request = partial(
+                            session.request,
+                            timeout=self._timeout,
+                        )
+                    self._anonymous_lyrics_api = self._create_api(None, session)
+        return self._anonymous_lyrics_api
+
+    def _clear_anonymous_lyrics_api(self):
+        with self._anonymous_lyrics_api_lock:
+            self._anonymous_lyrics_api = None
 
     def setup_language(self, language: str):
         self._language = language
+        self._clear_anonymous_lyrics_api()
 
     def setup_http_proxy(self, http_proxy):
         self._session.proxies = {
             "http": http_proxy,
             "https": http_proxy,
         }
+        self._clear_anonymous_lyrics_api()
 
     def setup_timeout(self, timeout):
+        self._timeout = timeout
         if isinstance(self._session.request, partial):
             request = self._session.request.func
         else:
             request = self._session.request
         self._session.request = partial(request, timeout=timeout)
+        self._clear_anonymous_lyrics_api()
 
     def search(
         self,
@@ -365,6 +403,20 @@ def album_info(self, browse_id: str) -> AlbumInfo:
     def song_info(self, video_id: str) -> SongInfo:
         return SongInfo(**self.api.get_song(video_id, self.get_signature_timestamp()))
 
+    def _song_lyrics_browse_id(self, video_id: str, api=None) -> Optional[str]:
+        if api is None:
+            api = self.api
+        response = api.send_api_request("next", build_watch_playlist_body(video_id))
+        return extract_lyrics_browse_id(response)
+
+    def song_lyrics(self, video_id: str) -> Optional[str]:
+        lyrics_api = self._get_anonymous_lyrics_api()
+        lyrics_browse_id = self._song_lyrics_browse_id(video_id, api=lyrics_api)
+        if lyrics_browse_id is None:
+            return None
+        lyrics_payload = lyrics_api.get_lyrics(lyrics_browse_id, timestamps=True)
+        return timestamped_lyrics_to_lrc(lyrics_payload)
+
     @ttl_cache(maxsize=CACHE_SIZE, ttl=CACHE_TTL)
     def categories(self) -> List[Categories]:
         return [

manual_tests/lyrics_e2e_test.py

@@ -0,0 +1,103 @@
+"""Manual end-to-end test for YTMusic timestamped lyrics.
+
+Run with:
+  uv run pytest manual_tests/lyrics_e2e_test.py -s --run-manual-tests
+
+Optional environment variables:
+  YTMUSIC_MANUAL_PROXY          HTTP proxy url, e.g. http://127.0.0.1:7890
+  YTMUSIC_MANUAL_TIMEOUT        socket timeout in seconds (default: 8)
+  YTMUSIC_MANUAL_LYRIC_SONG_IDS comma-separated song ids (default: tn7rzN8ABuo)
+  YTMUSIC_MANUAL_USE_HEADERFILE set to 1 to authenticate with ytmusic_header.json
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import socket
+from types import SimpleNamespace
+
+import pytest
+from feeluown.player.lyric import parse_lyric_text
+
+from fuo_ytmusic.consts import HEADER_FILE
+from fuo_ytmusic.provider import provider
+
+
+def _env_int(name: str, default: int) -> int:
+    value = os.getenv(name, "").strip()
+    if not value:
+        return default
+    try:
+        return int(value)
+    except ValueError:
+        return default
+
+
+def _env_bool(name: str) -> bool:
+    return os.getenv(name, "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _env_song_ids() -> list[str]:
+    value = os.getenv("YTMUSIC_MANUAL_LYRIC_SONG_IDS", "").strip()
+    if not value:
+        return ["tn7rzN8ABuo"]
+    return [part.strip() for part in value.split(",") if part.strip()]
+
+
+def _env_proxy() -> str:
+    return (
+        os.getenv("YTMUSIC_MANUAL_PROXY", "").strip()
+        or os.getenv("HTTP_PROXY", "").strip()
+        or os.getenv("http_proxy", "").strip()
+    )
+
+
+def _setup_provider(timeout: int, proxy: str):
+    socket.setdefaulttimeout(timeout)
+    if proxy:
+        provider.setup_http_proxy(proxy)
+    provider.setup_http_timeout(timeout)
+
+    if not _env_bool("YTMUSIC_MANUAL_USE_HEADERFILE"):
+        print("running anonymously")
+        return
+
+    if not HEADER_FILE.exists():
+        print(f"headerfile not found, running anonymously: {HEADER_FILE}")
+        return
+
+    user = provider.try_get_user_with_headerfile()
+    if user is None:
+        print("auto login failed, running anonymously")
+        return
+    provider.auth(user)
+
+
+@pytest.mark.manual
+def test_song_get_lyric_end_to_end():
+    timeout = _env_int("YTMUSIC_MANUAL_TIMEOUT", 8)
+    proxy = _env_proxy()
+    song_ids = _env_song_ids()
+
+    _setup_provider(timeout, proxy)
+    print(
+        f"manual lyrics config: timeout={timeout}, proxy={'set' if proxy else 'unset'}"
+    )
+
+    for song_id in song_ids:
+        print(f"\n=== song={song_id} ===")
+        lyric = provider.song_get_lyric(SimpleNamespace(identifier=song_id))
+        if lyric is None:
+            pytest.fail(f"no timestamped lyrics returned for {song_id}")
+
+        parsed = parse_lyric_text(lyric.content)
+        assert parsed, "lyric content should be parseable by FeelUOwn as LRC"
+        print(f"timestamped lines: {len(parsed)}")
+        summaries = []
+        for line in lyric.content.splitlines()[:10]:
+            timestamp = line.split("]", 1)[0] + "]" if "]" in line else ""
+            digest = hashlib.sha1(line.encode()).hexdigest()[:10]
+            summaries.append((timestamp, len(line), digest))
+        print(f"first 10 line summaries: {summaries}")
+        print(f"first 10 parsed timestamps: {list(parsed.keys())[:10]}")