Update yt_dlp/extractor/youtube/pot/README.md

test/test_pot/conftest.py

@@ -1,3 +1,5 @@
+import collections
+
 import pytest
 
 from yt_dlp import YoutubeDL
@@ -14,27 +16,22 @@ class MockLogger(IEContentProviderLogger):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.messages = {}
+        self.messages = collections.defaultdict(list)
 
     def trace(self, message: str):
-        self.messages.setdefault('trace', []).append(message)
-        pass
+        self.messages['trace'].append(message)
 
     def debug(self, message: str):
-        self.messages.setdefault('debug', []).append(message)
-        pass
+        self.messages['debug'].append(message)
 
     def info(self, message: str):
-        self.messages.setdefault('info', []).append(message)
-        pass
+        self.messages['info'].append(message)
 
     def warning(self, message: str, *, once=False):
-        self.messages.setdefault('warning', []).append(message)
-        pass
+        self.messages['warning'].append(message)
 
     def error(self, message: str):
-        self.messages.setdefault('error', []).append(message)
-        pass
+        self.messages['error'].append(message)
 
 
 @pytest.fixture

test/test_pot/test_pot_builtin_memorycache.py

@@ -2,7 +2,7 @@ import threading
 import time
 from collections import OrderedDict
 import pytest
-from yt_dlp.extractor.youtube.pot._provider import IEContentProvider, BuiltInIEContentProvider
+from yt_dlp.extractor.youtube.pot._provider import IEContentProvider, BuiltinIEContentProvider
 from yt_dlp.utils import bug_reports_message
 from yt_dlp.extractor.youtube.pot._builtin.memory_cache import MemoryLRUPCP, memorylru_preference, initialize_global_cache
 from yt_dlp.version import __version__
@@ -13,7 +13,7 @@ class TestMemoryLRUPCS:
 
     def test_base_type(self):
         assert issubclass(MemoryLRUPCP, IEContentProvider)
-        assert issubclass(MemoryLRUPCP, BuiltInIEContentProvider)
+        assert issubclass(MemoryLRUPCP, BuiltinIEContentProvider)
 
     @pytest.fixture
     def pcp(self, ie, logger) -> MemoryLRUPCP:

test/test_pot/test_pot_builtin_webpospec.py

@@ -1,6 +1,6 @@
 import pytest
 
-from yt_dlp.extractor.youtube.pot._provider import IEContentProvider, BuiltInIEContentProvider
+from yt_dlp.extractor.youtube.pot._provider import IEContentProvider, BuiltinIEContentProvider
 from yt_dlp.extractor.youtube.pot.cache import CacheProviderWritePolicy
 from yt_dlp.utils import bug_reports_message
 from yt_dlp.extractor.youtube.pot.provider import (
@@ -23,7 +23,7 @@ def pot_request(pot_request) -> PoTokenRequest:
 class TestWebPoPCSP:
     def test_base_type(self):
         assert issubclass(WebPoPCSP, IEContentProvider)
-        assert issubclass(WebPoPCSP, BuiltInIEContentProvider)
+        assert issubclass(WebPoPCSP, BuiltinIEContentProvider)
 
     def test_init(self, ie, logger):
         pcs = WebPoPCSP(ie=ie, logger=logger, settings={})

test/test_pot/test_pot_director.py

@@ -7,7 +7,7 @@ import json
 import time
 import pytest
 
-from yt_dlp.extractor.youtube.pot._provider import BuiltInIEContentProvider, IEContentProvider
+from yt_dlp.extractor.youtube.pot._provider import BuiltinIEContentProvider, IEContentProvider
 
 from yt_dlp.extractor.youtube.pot.provider import (
     PoTokenRequest,
@@ -578,7 +578,8 @@ class TestPoTokenCache:
         assert cache.get(pot_request) is None
         cache.store(pot_request, response)
         assert len(memorypcp.cache) == 1
-        assert hashlib.sha256(f'_dlp_cachev1_pExampleProviderv{pot_request.video_id}'.encode()).hexdigest() in memorypcp.cache
+        assert hashlib.sha256(
+            f"{{'_dlp_cache': 'v1', '_p': 'ExampleProvider', 'v': '{pot_request.video_id}'}}".encode()).hexdigest() in memorypcp.cache
 
         # The second spec provider returns the exact same key bindings as the first one,
         # however the PoTokenCache should use the provider key to differentiate between them
@@ -591,7 +592,8 @@ class TestPoTokenCache:
         assert cache.get(pot_request) is None
         cache.store(pot_request, response)
         assert len(memorypcp.cache) == 2
-        assert hashlib.sha256(f'_dlp_cachev1_pExampleProviderTwov{pot_request.video_id}'.encode()).hexdigest() in memorypcp.cache
+        assert hashlib.sha256(
+            f"{{'_dlp_cache': 'v1', '_p': 'ExampleProviderTwo', 'v': '{pot_request.video_id}'}}".encode()).hexdigest() in memorypcp.cache
 
     def test_cache_provider_preferences(self, pot_request, ie, logger):
         pcp_one = create_memory_pcp(ie, logger, provider_key='memory_pcp_one')
@@ -1478,11 +1480,11 @@ def test_validate_pot_response(response, expected):
 
 
 def test_built_in_provider(ie, logger):
-    class BuiltInProviderDefaultT(BuiltInIEContentProvider, suffix='T'):
+    class BuiltinProviderDefaultT(BuiltinIEContentProvider, suffix='T'):
         def is_available(self):
             return True
 
-    class BuiltInProviderCustomNameT(BuiltInIEContentProvider, suffix='T'):
+    class BuiltinProviderCustomNameT(BuiltinIEContentProvider, suffix='T'):
         PROVIDER_NAME = 'CustomName'
 
         def is_available(self):
@@ -1503,18 +1505,25 @@ def test_built_in_provider(ie, logger):
         def is_available(self) -> bool:
             return False
 
-    class BuiltInProviderUnavailableT(IEContentProvider, suffix='T'):
+    class BuiltinProviderUnavailableT(IEContentProvider, suffix='T'):
         def is_available(self) -> bool:
             return False
 
-    built_in_default = BuiltInProviderDefaultT(ie=ie, logger=logger, settings={})
-    built_in_custom_name = BuiltInProviderCustomNameT(ie=ie, logger=logger, settings={})
-    built_in_unavailable = BuiltInProviderUnavailableT(ie=ie, logger=logger, settings={})
+    built_in_default = BuiltinProviderDefaultT(ie=ie, logger=logger, settings={})
+    built_in_custom_name = BuiltinProviderCustomNameT(ie=ie, logger=logger, settings={})
+    built_in_unavailable = BuiltinProviderUnavailableT(ie=ie, logger=logger, settings={})
     external_default = ExternalProviderDefaultT(ie=ie, logger=logger, settings={})
     external_custom = ExternalProviderCustomT(ie=ie, logger=logger, settings={})
     external_unavailable = ExternalProviderUnavailableT(ie=ie, logger=logger, settings={})
 
     assert provider_display_list([]) == 'none'
-    assert provider_display_list([built_in_default]) == 'BuiltInProviderDefault'
+    assert provider_display_list([built_in_default]) == 'BuiltinProviderDefault'
     assert provider_display_list([external_unavailable]) == 'ExternalProviderUnavailable-0.0.0 (external, unavailable)'
-    assert provider_display_list([built_in_default, built_in_custom_name, external_default, external_custom, external_unavailable, built_in_unavailable]) == 'BuiltInProviderDefault, CustomName, ExternalProviderDefault-0.0.0 (external), custom-5.4b2 (external), ExternalProviderUnavailable-0.0.0 (external, unavailable), BuiltInProviderUnavailable-0.0.0 (external, unavailable)'
+    assert provider_display_list([
+        built_in_default,
+        built_in_custom_name,
+        external_default,
+        external_custom,
+        external_unavailable,
+        built_in_unavailable],
+    ) == 'BuiltinProviderDefault, CustomName, ExternalProviderDefault-0.0.0 (external), custom-5.4b2 (external), ExternalProviderUnavailable-0.0.0 (external, unavailable), BuiltinProviderUnavailable-0.0.0 (external, unavailable)'

yt_dlp/extractor/youtube/pot/README.md

@@ -4,6 +4,10 @@ As part of the YouTube extractor, we have a framework for providing PO Tokens pr
 
 Refer to the [PO Token Guide](https://github.com/yt-dlp/yt-dlp/wiki/PO-Token-Guide) for more information on PO Tokens.
 
+> [!TIP]
+> If publishing a PO Token Provider plugin to GitHub, add the [yt-dlp-pot-provider](https://github.com/topics/yt-dlp-pot-provider) topic to your repository to help users find it.
+
+
 ## Public APIs
 
 - `yt_dlp.extractor.youtube.pot.cache`
@@ -143,7 +147,7 @@ class MyPoTokenProviderPTP(PoTokenProvider):  # Provider class name must end wit
 
         return PoTokenResponse(
             po_token=po_token,
-            # Optional, add a custom expiration time for the token. Use for caching.
+            # Optional, add a custom expiration timestamp for the token. Use for caching.
             # By default, yt-dlp will use the default ttl from a registered cache spec (see below)
             # Set to 0 or -1 to not cache this response.
             expires_at=None,
@@ -168,7 +172,7 @@ def my_provider_preference(provider: PoTokenProvider, request: PoTokenRequest) -
 - Use `self.logger.debug` to log a message to the verbose output (`--verbose`).
   - For debugging information visible to users posting verbose logs.
   - Try to not log too much, prefer using trace logging for detailed debug messages.
-- Use `self.logger.trace` to log a message to the PO Token debug output (`--extractor-args "youtube:pot_debug=true"`). 
+- Use `self.logger.trace` to log a message to the PO Token debug output (`--extractor-args "youtube:pot_trace=true"`). 
   - Log as much as you like here as needed for debugging your provider.
 - Avoid logging PO Tokens or any sensitive information to debug or info output.
 

yt_dlp/extractor/youtube/pot/__init__.py

@@ -1,2 +1,3 @@
 # Trigger import of built-in providers
-from ._builtin import _trigger_import  # noqa: F401
+from ._builtin.memory_cache import MemoryLRUPCP as _MemoryLRUPCP  # noqa: F401
+from ._builtin.webpo_cachespec import WebPoPCSP as _WebPoPCSP  # noqa: F401

yt_dlp/extractor/youtube/pot/_builtin/__init__.py

@@ -1,4 +0,0 @@
-from .memory_cache import MemoryLRUPCP as _MemoryLRUPCP  # noqa: F401
-from .webpo_cachespec import WebPoPCSP as _WebPoPCSP  # noqa: F401
-
-_trigger_import = None  # noqa: F401

yt_dlp/extractor/youtube/pot/_builtin/memory_cache.py

@@ -2,10 +2,9 @@ from __future__ import annotations
 
 import datetime as dt
 import typing
-from collections import OrderedDict
 from threading import Lock
 
-from yt_dlp.extractor.youtube.pot._provider import BuiltInIEContentProvider
+from yt_dlp.extractor.youtube.pot._provider import BuiltinIEContentProvider
 from yt_dlp.extractor.youtube.pot._registry import _pot_memory_cache
 from yt_dlp.extractor.youtube.pot.cache import (
     PoTokenCacheProvider,
@@ -16,7 +15,7 @@ from yt_dlp.extractor.youtube.pot.cache import (
 
 def initialize_global_cache(max_size: int):
     if _pot_memory_cache.value.get('cache') is None:
-        _pot_memory_cache.value['cache'] = OrderedDict()
+        _pot_memory_cache.value['cache'] = {}
         _pot_memory_cache.value['lock'] = Lock()
         _pot_memory_cache.value['max_size'] = max_size
 
@@ -31,14 +30,14 @@ def initialize_global_cache(max_size: int):
 
 
 @register_provider
-class MemoryLRUPCP(PoTokenCacheProvider, BuiltInIEContentProvider):
+class MemoryLRUPCP(PoTokenCacheProvider, BuiltinIEContentProvider):
     PROVIDER_NAME = 'memory'
     DEFAULT_CACHE_SIZE = 25
 
     def __init__(
         self,
         *args,
-        initialize_cache: typing.Callable[[int], tuple[OrderedDict, Lock, int]] = initialize_global_cache,
+        initialize_cache: typing.Callable[[int], tuple[dict[str, tuple[str, int]], Lock, int]] = initialize_global_cache,
         **kwargs,
     ):
         super().__init__(*args, **kwargs)
@@ -65,7 +64,8 @@ class MemoryLRUPCP(PoTokenCacheProvider, BuiltInIEContentProvider):
                 self.cache.pop(key)
             self.cache[key] = (value, expires_at)
             if len(self.cache) > self.max_size:
-                self.cache.popitem(last=False)
+                oldest_key = next(iter(self.cache))
+                self.cache.pop(oldest_key)
 
     def delete(self, key: str):
         with self.lock:

yt_dlp/extractor/youtube/pot/_builtin/webpo_cachespec.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from yt_dlp.extractor.youtube.pot._provider import BuiltInIEContentProvider
+from yt_dlp.extractor.youtube.pot._provider import BuiltinIEContentProvider
 from yt_dlp.extractor.youtube.pot.cache import (
     CacheProviderWritePolicy,
     PoTokenCacheSpec,
@@ -15,7 +15,7 @@ from yt_dlp.utils import traverse_obj
 
 
 @register_spec
-class WebPoPCSP(PoTokenCacheSpecProvider, BuiltInIEContentProvider):
+class WebPoPCSP(PoTokenCacheSpecProvider, BuiltinIEContentProvider):
     PROVIDER_NAME = 'webpo'
 
     def generate_cache_spec(self, request: PoTokenRequest) -> PoTokenCacheSpec | None:

yt_dlp/extractor/youtube/pot/_director.py

@@ -11,7 +11,7 @@ import urllib.parse
 from collections.abc import Iterable
 
 from yt_dlp.extractor.youtube.pot._provider import (
-    BuiltInIEContentProvider,
+    BuiltinIEContentProvider,
     IEContentProvider,
     IEContentProviderLogger,
 )
@@ -139,12 +139,11 @@ class PoTokenCache:
         }
         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())))
+        self.logger.trace(f'Generated cache key bindings: {bindings_cleaned}')
         return bindings_cleaned
 
     def _generate_key(self, bindings: dict) -> str:
-        binding_string = ''.join(f'{k}{v}' for k, v in sorted(bindings.items()))
+        binding_string = ''.join(repr(dict(sorted(bindings.items()))))
         return hashlib.sha256(binding_string.encode()).hexdigest()
 
     def get(self, request: PoTokenRequest) -> PoTokenResponse | None:
@@ -415,9 +414,9 @@ def provider_display_list(providers: Iterable[IEContentProvider]):
     def provider_display_name(provider):
         display_str = join_nonempty(
             provider.PROVIDER_NAME,
-            provider.PROVIDER_VERSION if not isinstance(provider, BuiltInIEContentProvider) else None)
+            provider.PROVIDER_VERSION if not isinstance(provider, BuiltinIEContentProvider) else None)
         statuses = []
-        if not isinstance(provider, BuiltInIEContentProvider):
+        if not isinstance(provider, BuiltinIEContentProvider):
             statuses.append('external')
         if not provider.is_available():
             statuses.append('unavailable')
@@ -451,16 +450,10 @@ def validate_response(response: PoTokenResponse | None):
     except ValueError:
         return False
 
-    return (
-        response.expires_at is None
-        or (
-            isinstance(response.expires_at, int)
-            and (
-                response.expires_at <= 0
-                or response.expires_at > int(dt.datetime.now(dt.timezone.utc).timestamp())
-            )
-        )
-    )
+    if not isinstance(response.expires_at, int):
+        return response.expires_at is None
+
+    return response.expires_at <= 0 or response.expires_at > int(dt.datetime.now(dt.timezone.utc).timestamp())
 
 
 def validate_cache_spec(spec: PoTokenCacheSpec):

yt_dlp/extractor/youtube/pot/_provider.py

@@ -120,7 +120,7 @@ class IEContentProvider(abc.ABC):
         return list(val) if casesense else [x.lower() for x in val]
 
 
-class BuiltInIEContentProvider(IEContentProvider, abc.ABC):
+class BuiltinIEContentProvider(IEContentProvider, abc.ABC):
     PROVIDER_VERSION = __version__
     BUG_REPORT_MESSAGE = bug_reports_message(before='')
 

yt_dlp/extractor/youtube/pot/provider.py

@@ -244,7 +244,7 @@ def provider_bug_report_message(provider: IEContentProvider, before=';'):
     if not before or before.endswith(('.', '!', '?')):
         msg = msg[0].title() + msg[1:]
 
-    return (before + ' ' if before else '') + msg
+    return f'{before} {msg}' if before else msg
 
 
 def register_preference(*providers: type[PoTokenProvider]) -> typing.Callable[[Preference], Preference]: