Merge pull request #23 from feeluown/codex/image-cover-proxy

Refactor yt-dlp cookiefile handling and add stability tests

Author: cosven

fuo_ytmusic/headerfile.py

@@ -8,13 +8,68 @@
 # Store selected profile in headerfile using a custom key; this is not part of
 # YouTube headers but lets us persist profile selection without extra files.
 PROFILE_GAIA_KEY = "X-Feeluown-Profile-Gaia-Id"
-YTDLP_COOKIE_DOMAINS = (".youtube.com", ".google.com")
+# Include both root YouTube domain and Music subdomain explicitly to avoid
+# extractor differences when matching cookie scope.
+YTDLP_COOKIE_DOMAINS = (".youtube.com", ".music.youtube.com", ".google.com")
 
 # yt-dlp reads cookies from a Netscape cookiefile. We persist this file next to
 # ytmusic_header.json and keep it synchronized whenever header auth cookie is
 # written/updated, so playback can reuse the same long-lived auth lifecycle.
 
 
+class YtdlpCookiefileManager:
+    """Manage yt-dlp cookiefile generation for a specific headerfile path."""
+
+    def __init__(self, headerfile_path=HEADER_FILE):
+        self._headerfile_path = (
+            Path(headerfile_path) if headerfile_path is not None else None
+        )
+
+    @property
+    def headerfile_path(self) -> Optional[Path]:
+        return self._headerfile_path
+
+    @property
+    def cookiefile_path(self) -> Optional[Path]:
+        path = self._headerfile_path
+        if path is None or not path.exists():
+            return None
+        if path.suffix:
+            return path.with_suffix(".cookies.txt")
+        return Path(f"{path}.cookies.txt")
+
+    def write(self, cookie_value: str) -> Optional[Path]:
+        # Write/update the persistent cookiefile used by yt-dlp. Empty cookie
+        # means delete the file so stale auth is not reused.
+        cookiefile_path = self.cookiefile_path
+        if cookiefile_path is None:
+            return None
+
+        cookie_jar = SimpleCookie()
+        cookie_jar.load(cookie_value or "")
+        if not cookie_jar:
+            if cookiefile_path.exists():
+                cookiefile_path.unlink()
+            return None
+
+        lines = [
+            "# Netscape HTTP Cookie File",
+            "# This file is generated by FeelUOwn ytmusic plugin. Do not edit.",
+            "",
+        ]
+        for morsel in cookie_jar.values():
+            for domain in YTDLP_COOKIE_DOMAINS:
+                lines.append(
+                    f"{domain}\tTRUE\t/\tTRUE\t0\t{morsel.key}\t{morsel.value}"
+                )
+
+        cookiefile_path.parent.mkdir(parents=True, exist_ok=True)
+        with cookiefile_path.open("w", encoding="utf-8") as handle:
+            handle.write("\n".join(lines))
+            handle.write("\n")
+        return cookiefile_path
+
+
 def write_headerfile(auth: str, cookie: str, headerfile_path=HEADER_FILE) -> None:
     tpl = {
         "Accept": "*/*",
@@ -27,7 +82,7 @@ def write_headerfile(auth: str, cookie: str, headerfile_path=HEADER_FILE) -> Non
     with headerfile_path.open("w") as handle:
         json.dump(tpl, handle, indent=2, ensure_ascii=True)
     # Keep yt-dlp cookiefile in sync with freshly written headerfile auth.
-    write_ytdlp_cookiefile(cookie, headerfile_path)
+    YtdlpCookiefileManager(headerfile_path).write(cookie)
 
 
 def read_headerfile(headerfile_path=HEADER_FILE) -> dict:
@@ -72,49 +127,20 @@ def update_profile_gaia_id(gaia_id: Optional[str], headerfile_path=HEADER_FILE)
         return False
     with open(headerfile_path, "w") as handle:
         json.dump(data, handle, indent=2, ensure_ascii=True)
+    # Profile switch metadata does not change auth cookie, so no cookiefile sync.
     return True
 
 
 def get_ytdlp_cookiefile_path(headerfile_path=HEADER_FILE) -> Optional[Path]:
-    # Read-only helper: resolve the stable cookiefile path for a headerfile.
-    if headerfile_path is None:
-        return None
-    path = Path(headerfile_path)
-    if path.suffix:
-        return path.with_suffix(".cookies.txt")
-    return Path(f"{path}.cookies.txt")
+    # Backward-compatible helper. New code should use YtdlpCookiefileManager.
+    return YtdlpCookiefileManager(headerfile_path).cookiefile_path
 
 
 def write_ytdlp_cookiefile(
     cookie_value: str, headerfile_path=HEADER_FILE
 ) -> Optional[Path]:
-    # Write/update the persistent cookiefile used by yt-dlp. Empty cookie means
-    # delete the file so stale auth is not reused.
-    cookiefile_path = get_ytdlp_cookiefile_path(headerfile_path)
-    if cookiefile_path is None:
-        return None
-
-    cookie_jar = SimpleCookie()
-    cookie_jar.load(cookie_value or "")
-    if not cookie_jar:
-        if cookiefile_path.exists():
-            cookiefile_path.unlink()
-        return None
-
-    lines = [
-        "# Netscape HTTP Cookie File",
-        "# This file is generated by FeelUOwn ytmusic plugin. Do not edit.",
-        "",
-    ]
-    for morsel in cookie_jar.values():
-        for domain in YTDLP_COOKIE_DOMAINS:
-            lines.append(f"{domain}\tTRUE\t/\tTRUE\t0\t{morsel.key}\t{morsel.value}")
-
-    cookiefile_path.parent.mkdir(parents=True, exist_ok=True)
-    with cookiefile_path.open("w", encoding="utf-8") as handle:
-        handle.write("\n".join(lines))
-        handle.write("\n")
-    return cookiefile_path
+    # Backward-compatible helper. New code should use YtdlpCookiefileManager.
+    return YtdlpCookiefileManager(headerfile_path).write(cookie_value)
 
 
 def update_headerfile_cookie(cookie_value: str, headerfile_path=HEADER_FILE) -> bool:
@@ -136,5 +162,5 @@ def update_headerfile_cookie(cookie_value: str, headerfile_path=HEADER_FILE) ->
             json.dump(data, handle, indent=2, ensure_ascii=True)
 
     # Keep yt-dlp cookiefile lifecycle aligned with header cookie updates.
-    write_ytdlp_cookiefile(cookie_value, headerfile_path)
+    YtdlpCookiefileManager(headerfile_path).write(cookie_value)
     return changed

fuo_ytmusic/provider.py

@@ -1,5 +1,5 @@
 import logging
-from typing import List, Optional
+from typing import List, Optional, Tuple
 
 from feeluown.excs import NoUserLoggedIn, ProviderIOError
 from feeluown.library import (
@@ -23,9 +23,14 @@
 from feeluown.media import Media, MediaType, Quality, VideoAudioManifest
 from feeluown.utils.dispatch import Signal
 from yt_dlp import DownloadError, YoutubeDL
+from yt_dlp.version import __version__ as YTDLP_VERSION
 
 from fuo_ytmusic.consts import HEADER_FILE
-from fuo_ytmusic.headerfile import get_profile_gaia_id, update_profile_gaia_id
+from fuo_ytmusic.headerfile import (
+    YtdlpCookiefileManager,
+    get_profile_gaia_id,
+    update_profile_gaia_id,
+)
 from fuo_ytmusic.home_recommendation import YtmusicHomeRecommendationBuilder
 from fuo_ytmusic.models import (
     Categories,
@@ -36,8 +41,32 @@
 logger = logging.getLogger(__name__)
 
 
+class _NoCookieSaveYoutubeDL(YoutubeDL):
+    """Load cookies from file but skip yt-dlp cookiefile write-back."""
+
+    def save_cookies(self):
+        # yt-dlp persists cookie jar on close by default. Keep this hook for
+        # experiments where cookiefile must be treated as read-only input.
+        return
+
+
+def _parse_ytdlp_version(version: str) -> Optional[Tuple[int, int, int]]:
+    parts = version.split(".")
+    if len(parts) < 3:
+        return None
+    try:
+        return int(parts[0]), int(parts[1]), int(parts[2])
+    except ValueError:
+        return None
+
+
 class YtmusicProvider(AbstractProvider, ProviderV2):
     HOME_SECTION_LIMIT = 12
+    # NOTE:
+    # yt-dlp >= 2025.12.08 may fail with cookiefile in some YouTube Music
+    # sessions due to PO token restrictions. For these versions, keep running
+    # yt-dlp without cookiefile and rely on proxy/timeout only.
+    YTDLP_COOKIEFILE_DISABLED_FROM = (2025, 12, 8)
     # get_charts returns playlist sections under these keys; artists are omitted.
     TOPLIST_CHART_KEYS = ("daily", "weekly", "videos", "genres")
 
@@ -56,15 +85,6 @@ def __init__(self):
             "socket_timeout": 2,
             "extractor_retries": 0,  # reduce retry
         }
-        self._default_audio_ytdl_opts = {
-            # The following two options may be only valid for select_audio API.
-            # Remove these two options if needed.
-            "format": "m4a/bestaudio/best",
-            **self._default_ytdl_opts,
-        }
-        self._default_video_ytdl_opts = {
-            **self._default_ytdl_opts,
-        }
 
     def setup_http_proxy(self, http_proxy):
         self._http_proxy = http_proxy
@@ -74,6 +94,34 @@ def setup_http_timeout(self, timeout):
         self.service.setup_timeout(timeout)
         self._default_ytdl_opts["socket_timeout"] = timeout
 
+    def _build_audio_ytdl_opts(self):
+        return {
+            # The following option may be only valid for select_audio API.
+            "format": "m4a/bestaudio/best",
+            **self._default_ytdl_opts,
+        }
+
+    def _build_video_ytdl_opts(self):
+        return dict(self._default_ytdl_opts)
+
+    def _should_use_ytdlp_cookiefile(self) -> bool:
+        parsed = _parse_ytdlp_version(YTDLP_VERSION)
+        if parsed is None:
+            # Keep old behavior for unknown version strings.
+            return True
+        return parsed < self.YTDLP_COOKIEFILE_DISABLED_FROM
+
+    def _get_ytdlp_cookiefile_path(self) -> str:
+        # Only authenticated provider sessions should pass cookiefile to yt-dlp.
+        # For anonymous sessions, keep yt-dlp behavior unchanged.
+        if not self.has_current_user():
+            return ""
+        if not self._should_use_ytdlp_cookiefile():
+            return ""
+        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)
+
     def setup_language(self, language: str):
         self.service.setup_language(language)
 
@@ -373,25 +421,23 @@ def song_list_quality(self, song) -> List[Quality.Audio]:
         return song_.list_formats() if song_ is not None else []
 
     def _song_get_media_from_ytdlp(self, song: SongModel) -> Optional[Media]:
-        ytdl_opts = {}
-        ytdl_opts.update(self._default_audio_ytdl_opts)
+        ytdl_opts = self._build_audio_ytdl_opts()
         # Only set proxy if it is nonempty.
         # Ytdl can make use of system proxy when proxy is not set.
         if self._http_proxy:
             ytdl_opts["proxy"] = self._http_proxy
 
-        # service keeps cookiefile synced with headerfile lifecycle; provider
-        # only consumes the stable path to configure yt-dlp runtime options.
-        cookiefile_path = self.service.get_ytdlp_cookiefile_path()
+        cookiefile_path = self._get_ytdlp_cookiefile_path()
         if cookiefile_path:
             ytdl_opts["cookiefile"] = cookiefile_path
 
-        user_agent = self.service.get_user_agent()
-        if user_agent:
-            ytdl_opts["user_agent"] = user_agent
+        debug_opts = dict(ytdl_opts)
+        debug_opts.pop("logger", None)
+        print(f"song_get_media yt-dlp options for {song.identifier}: {debug_opts}")
 
         url = self.song_get_web_url(song)
-        with YoutubeDL(ytdl_opts) as inner:
+        print(f"song_get_media source url for {song.identifier}: {url}")
+        with _NoCookieSaveYoutubeDL(ytdl_opts) as inner:
             info = inner.extract_info(url, download=False)
         media_url = info.get("url")
         if not media_url:
@@ -521,17 +567,19 @@ def video_get_web_url(self, video) -> str:
         return f"https://youtube.com/watch?v={video.identifier}"
 
     def video_get_media(self, video, quality) -> Optional[Media]:
-        ytdl_opts = {}
-        ytdl_opts.update(self._default_video_ytdl_opts)
+        ytdl_opts = self._build_video_ytdl_opts()
         # Only set proxy if it is nonempty.
         # Ytdl can make use of system proxy when proxy is not set.
         if self._http_proxy:
             ytdl_opts["proxy"] = self._http_proxy
+        cookiefile_path = self._get_ytdlp_cookiefile_path()
+        if cookiefile_path:
+            ytdl_opts["cookiefile"] = cookiefile_path
         url = self.video_get_web_url(video)
 
         audio_candidates = []  # [(url, abr)]  abr: average bitrate
         video_candidates = []  # [(url, width)]
-        with YoutubeDL(ytdl_opts) as inner:
+        with _NoCookieSaveYoutubeDL(ytdl_opts) as inner:
             try:
                 info = inner.extract_info(url, download=False)
             except DownloadError as e:  # noqa

fuo_ytmusic/service.py

@@ -16,11 +16,8 @@
 from ytmusicapi import YTMusic as YTMusicBase
 from ytmusicapi.ytmusic import OAuthCredentials
 
-from fuo_ytmusic.consts import HEADER_FILE
 from fuo_ytmusic.headerfile import (
-    get_ytdlp_cookiefile_path,
     update_headerfile_cookie,
-    write_ytdlp_cookiefile,
 )
 from fuo_ytmusic.helpers import Singleton
 from fuo_ytmusic.models import (
@@ -132,22 +129,6 @@ def update_auth_cookie_from_response(self, response):
             self._auth_headers["cookie"] = new_cookie
             self._persist_cookie_to_headerfile(new_cookie)
 
-    def get_auth_cookie(self) -> str:
-        auth_headers = getattr(self, "_auth_headers", None)
-        if isinstance(auth_headers, dict):
-            cookie = auth_headers.get("cookie") or auth_headers.get("Cookie")
-            if isinstance(cookie, str):
-                return cookie.strip()
-        return ""
-
-    def get_user_agent(self) -> str:
-        headers = getattr(self, "headers", None)
-        if isinstance(headers, dict):
-            user_agent = headers.get("User-Agent") or headers.get("user-agent")
-            if isinstance(user_agent, str):
-                return user_agent.strip()
-        return ""
-
     def _persist_cookie_to_headerfile(self, cookie_value: str):
         update_headerfile_cookie(cookie_value, self.headerfile_path)
 
@@ -292,10 +273,6 @@ def reinitialize_by_headerfile(self, headerfile=None):
             logger.info("Initializing ytmusic api with no headerfile.")
             self._api = YTMusic(**options)
 
-        # Keep yt-dlp cookiefile synchronized with current auth cookie so its
-        # lifecycle matches headerfile auth state.
-        self.sync_ytdlp_cookiefile()
-
     def setup_language(self, language: str):
         self._language = language
 
@@ -312,26 +289,6 @@ def setup_timeout(self, timeout):
             request = self._session.request
         self._session.request = partial(request, timeout=timeout)
 
-    def get_auth_cookie(self) -> str:
-        return self.api.get_auth_cookie()
-
-    def get_user_agent(self) -> str:
-        return self.api.get_user_agent()
-
-    def get_ytdlp_cookiefile_path(self) -> str:
-        # Read-only accessor used by provider when building yt-dlp options.
-        headerfile_path = getattr(self.api, "headerfile_path", None) or HEADER_FILE
-        cookiefile_path = get_ytdlp_cookiefile_path(headerfile_path)
-        return "" if cookiefile_path is None else str(cookiefile_path)
-
-    def sync_ytdlp_cookiefile(self) -> str:
-        # Explicitly refresh persistent cookiefile from current auth cookie.
-        headerfile_path = getattr(self.api, "headerfile_path", None) or HEADER_FILE
-        cookiefile_path = write_ytdlp_cookiefile(
-            self.get_auth_cookie(), headerfile_path
-        )
-        return "" if cookiefile_path is None else str(cookiefile_path)
-
     def search(
         self,
         keywords: str,