artem/yt-dlp
1
0
Fork 0
mirror of https://github.com/yt-dlp/yt-dlp.git synced 2026-04-07 01:13:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

typing cleanup

Browse Source
This commit is contained in:
coletdjnz 2025-05-16 20:13:25 +12:00
parent 8045fed380
commit fcdf2b732f
No known key found for this signature in database
GPG Key ID: 91984263BB39894A
5 changed files with 22 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
yt_dlp/extractor/youtube/pot/_builtin/webpo_cachespec.py
View File

@ -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

24
yt_dlp/extractor/youtube/pot/_director.py
View File

@ -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

6
yt_dlp/extractor/youtube/pot/cache.py
View File

@ -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]

4
yt_dlp/extractor/youtube/pot/provider.py
View File

@ -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):

2
yt_dlp/extractor/youtube/pot/utils.py
View File

@ -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: