diff --git a/yt_dlp/extractor/youtube/pot/_builtin/webpo_cachespec.py b/yt_dlp/extractor/youtube/pot/_builtin/webpo_cachespec.py index cfd42ad8d1..aac30add09 100644 --- a/yt_dlp/extractor/youtube/pot/_builtin/webpo_cachespec.py +++ b/yt_dlp/extractor/youtube/pot/_builtin/webpo_cachespec.py @@ -23,7 +23,7 @@ class WebPoPCSP(PoTokenCacheSpecProvider, BuiltInIEContentProvider): content_binding, content_binding_type = get_webpo_content_binding( request, bind_to_visitor_id=bind_to_visitor_id) - if not content_binding: + if not content_binding or not content_binding_type: return None write_policy = CacheProviderWritePolicy.WRITE_ALL diff --git a/yt_dlp/extractor/youtube/pot/_director.py b/yt_dlp/extractor/youtube/pot/_director.py index 0234d95070..2903634777 100644 --- a/yt_dlp/extractor/youtube/pot/_director.py +++ b/yt_dlp/extractor/youtube/pot/_director.py @@ -40,7 +40,8 @@ from yt_dlp.extractor.youtube.pot.provider import ( from yt_dlp.utils import ExtractorError, bug_reports_message, format_field, join_nonempty, traverse_obj if typing.TYPE_CHECKING: - from yt_dlp.extractor.youtube.pot.cache import PCPPreference + from yt_dlp.extractor.youtube.pot.cache import CacheProviderPreference + from yt_dlp.extractor.youtube.pot.provider import Preference class YoutubeIEContentProviderLogger(IEContentProviderLogger): @@ -81,12 +82,12 @@ class PoTokenCache: logger: IEContentProviderLogger, cache_providers: list[PoTokenCacheProvider], cache_spec_providers: list[PoTokenCacheSpecProvider], - cache_provider_preferences: list[PCPPreference] | None = None, + cache_provider_preferences: list[CacheProviderPreference] | None = None, ): self.cache_providers: dict[str, PoTokenCacheProvider] = { provider.PROVIDER_KEY: provider for provider in (cache_providers or []) } - self.cache_provider_preferences: list[PCPPreference] = cache_provider_preferences or [] + self.cache_provider_preferences: list[CacheProviderPreference] = cache_provider_preferences or [] self.cache_spec_providers: dict[str, PoTokenCacheSpecProvider] = { provider.PROVIDER_KEY: provider for provider in (cache_spec_providers or []) @@ -107,8 +108,8 @@ class PoTokenCache: f'{provider.PROVIDER_KEY}={pref}' for provider, pref in preferences.items()))) return ( - provider for provider in sorted(self.cache_providers.values(), key=preferences.get, reverse=True) if provider.is_available() - ) + provider for provider in sorted( + self.cache_providers.values(), key=preferences.get, reverse=True) if provider.is_available()) def _get_cache_spec(self, request: PoTokenRequest) -> PoTokenCacheSpec | None: for provider in self.cache_spec_providers.values(): @@ -129,14 +130,16 @@ class PoTokenCache: f'Error occurred with "{provider.PROVIDER_NAME}" PO Token cache spec provider: {e!r}{provider_bug_report_message(provider)}', ) continue + return None def _generate_key_bindings(self, spec: PoTokenCacheSpec) -> dict[str, str]: bindings_cleaned = { **{k: v for k, v in spec.key_bindings.items() if v is not None}, # Allow us to invalidate caches if such need arises '_yt': 'v1', - '_p': spec._provider.PROVIDER_KEY, } + if spec._provider: + bindings_cleaned['_p'] = spec._provider.PROVIDER_KEY self.logger.trace('Generate cache key bindings: {}'.format(', '.join(f'{k}={v}' for k, v in bindings_cleaned.items()))) return bindings_cleaned @@ -184,6 +187,7 @@ class PoTokenCache: f'Error occurred with "{provider.PROVIDER_NAME}" PO Token cache provider: {e!r}{provider_bug_report_message(provider)}', ) continue + return None def store(self, request: PoTokenRequest, response: PoTokenResponse, write_policy: CacheProviderWritePolicy | None = None): spec = self._get_cache_spec(request) @@ -232,15 +236,15 @@ class PoTokenCache: class PoTokenRequestDirector: def __init__(self, logger: IEContentProviderLogger, cache: PoTokenCache): - self.providers = {} - self.preferences = [] + self.providers: dict[str, PoTokenProvider] = {} + self.preferences: list[Preference] = [] self.cache = cache self.logger = logger def register_provider(self, provider: PoTokenProvider): self.providers[provider.PROVIDER_KEY] = provider - def register_preference(self, preference): + def register_preference(self, preference: Preference): self.preferences.append(preference) def _get_providers(self, request: PoTokenRequest) -> Iterable[PoTokenProvider]: @@ -407,7 +411,7 @@ def clean_pot(po_token: str): raise ValueError('Invalid PO Token') -def validate_response(response: PoTokenResponse): +def validate_response(response: PoTokenResponse | None): if ( not isinstance(response, PoTokenResponse) or not response.po_token diff --git a/yt_dlp/extractor/youtube/pot/cache.py b/yt_dlp/extractor/youtube/pot/cache.py index 795b0f57f6..6d69316adc 100644 --- a/yt_dlp/extractor/youtube/pot/cache.py +++ b/yt_dlp/extractor/youtube/pot/cache.py @@ -83,8 +83,8 @@ def register_spec(provider: type[PoTokenCacheSpecProvider]): ) -# XXX: I don't think the typing is correct, and that we need py3.10 to properly type this -def register_preference(*providers: type[PoTokenCacheProvider]) -> typing.Callable[[PCPPreference], PCPPreference]: +def register_preference( + *providers: type[PoTokenCacheProvider]) -> typing.Callable[[CacheProviderPreference], CacheProviderPreference]: """Register a preference for a PoTokenCacheProvider""" return register_preference_generic( PoTokenCacheProvider, @@ -94,4 +94,4 @@ def register_preference(*providers: type[PoTokenCacheProvider]) -> typing.Callab if typing.TYPE_CHECKING: - PCPPreference = typing.Callable[[PoTokenCacheProvider, PoTokenRequest, ...], int] + CacheProviderPreference = typing.Callable[[PoTokenCacheProvider, PoTokenRequest], int] diff --git a/yt_dlp/extractor/youtube/pot/provider.py b/yt_dlp/extractor/youtube/pot/provider.py index 1963e153f6..6649f71443 100644 --- a/yt_dlp/extractor/youtube/pot/provider.py +++ b/yt_dlp/extractor/youtube/pot/provider.py @@ -198,6 +198,7 @@ class PoTokenProvider(IEContentProvider, abc.ABC, suffix='PTP'): # Merge some ctx request settings into the request # Most of these will already be used by the configured ydl instance, # however, the YouTube extractor may override some. + # TODO: add _Request_webpage to IeContentProvider if pot_request is not None: req.headers = HTTPHeaderDict(pot_request.request_headers, req.headers) req.proxies = req.proxies or ({'all': pot_request.request_proxy} if pot_request.request_proxy else {}) @@ -229,7 +230,6 @@ def provider_bug_report_message(provider: IEContentProvider, before=';'): return (before + ' ' if before else '') + msg -# XXX: I don't think the typing is correct, and that we need py3.10 to properly type this def register_preference(*providers: type[PoTokenProvider]) -> typing.Callable[[Preference], Preference]: """Register a preference for a PoTokenProvider""" return register_preference_generic( @@ -240,7 +240,7 @@ def register_preference(*providers: type[PoTokenProvider]) -> typing.Callable[[P if typing.TYPE_CHECKING: - Preference = typing.Callable[[PoTokenProvider, PoTokenRequest, ...], int] + Preference = typing.Callable[[PoTokenProvider, PoTokenRequest], int] # Barebones innertube context. There may be more fields. class ClientInfo(typing.TypedDict, total=False): diff --git a/yt_dlp/extractor/youtube/pot/utils.py b/yt_dlp/extractor/youtube/pot/utils.py index 82070eda25..bd9a12b099 100644 --- a/yt_dlp/extractor/youtube/pot/utils.py +++ b/yt_dlp/extractor/youtube/pot/utils.py @@ -49,6 +49,8 @@ def get_webpo_content_binding(request: PoTokenRequest, webpo_clients=WEBPO_CLIEN elif request.context == PoTokenContext.PLAYER or client_name != 'WEB_REMIX': return request.video_id, ContentBindingType.VIDEO_ID + return None, None + def _extract_visitor_id(visitor_data): if not visitor_data: