[core] Load plugins on demand (#11305)

- Adds `--no-plugin-dirs` to disable plugin loading
- `--plugin-dirs` now supports post-processors

Authored by: coletdjnz, Grub4K, pukkandan

24 files changed, 532 insertions, 180 deletions

diff --git a/README.md b/README.md
index e8ef1980a..ca0d4dfb5 100644
--- a/README.md
+++ b/README.md
@@ -337,10 +337,11 @@ If you fork the project on GitHub, you can run your fork's [build workflow](.git
     --plugin-dirs PATH              Path to an additional directory to search
                                     for plugins. This option can be used
                                     multiple times to add multiple directories.
-                                    Note that this currently only works for
-                                    extractor plugins; postprocessor plugins can
-                                    only be loaded from the default plugin
-                                    directories
+                                    Use "default" to search the default plugin
+                                    directories (default)
+    --no-plugin-dirs                Clear plugin directories to search,
+                                    including defaults and those provided by
+                                    previous --plugin-dirs
     --flat-playlist                 Do not extract a playlist's URL result
                                     entries; some entry metadata may be missing
                                     and downloading may be bypassed
diff --git a/devscripts/make_lazy_extractors.py b/devscripts/make_lazy_extractors.py
index d288d8429..0ce773e82 100644
--- a/devscripts/make_lazy_extractors.py
+++ b/devscripts/make_lazy_extractors.py
@@ -10,6 +10,9 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 from inspect import getsource
 
 from devscripts.utils import get_filename_args, read_file, write_file
+from yt_dlp.extractor import import_extractors
+from yt_dlp.extractor.common import InfoExtractor, SearchInfoExtractor
+from yt_dlp.globals import extractors
 
 NO_ATTR = object()
 STATIC_CLASS_PROPERTIES = [
@@ -38,8 +41,7 @@ def main():
 
     lazy_extractors_filename = get_filename_args(default_outfile='yt_dlp/extractor/lazy_extractors.py')
 
-    from yt_dlp.extractor.extractors import _ALL_CLASSES
-    from yt_dlp.extractor.common import InfoExtractor, SearchInfoExtractor
+    import_extractors()
 
     DummyInfoExtractor = type('InfoExtractor', (InfoExtractor,), {'IE_NAME': NO_ATTR})
     module_src = '\n'.join((
@@ -47,7 +49,7 @@ def main():
         '    _module = None',
         *extra_ie_code(DummyInfoExtractor),
         '\nclass LazyLoadSearchExtractor(LazyLoadExtractor):\n    pass\n',
-        *build_ies(_ALL_CLASSES, (InfoExtractor, SearchInfoExtractor), DummyInfoExtractor),
+        *build_ies(list(extractors.value.values()), (InfoExtractor, SearchInfoExtractor), DummyInfoExtractor),
     ))
 
     write_file(lazy_extractors_filename, f'{module_src}\n')
@@ -73,7 +75,7 @@ def build_ies(ies, bases, attr_base):
         if ie in ies:
             names.append(ie.__name__)
 
-    yield f'\n_ALL_CLASSES = [{", ".join(names)}]'
+    yield '\n_CLASS_LOOKUP = {%s}' % ', '.join(f'{name!r}: {name}' for name in names)
 
 
 def sort_ies(ies, ignored_bases):
diff --git a/pyproject.toml b/pyproject.toml
index 5eb9a9644..2a0008a45 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -384,6 +384,7 @@ select = [
     "W391",
     "W504",
 ]
+exclude = "*/extractor/lazy_extractors.py,*venv*,*/test/testdata/sigs/player-*.js,.idea,.vscode"
 
 [tool.pytest.ini_options]
 addopts = "-ra -v --strict-markers"
diff --git a/test/test_YoutubeDL.py b/test/test_YoutubeDL.py
index 17e081bc6..708a04f92 100644
--- a/test/test_YoutubeDL.py
+++ b/test/test_YoutubeDL.py
@@ -6,6 +6,8 @@ import sys
 import unittest
 from unittest.mock import patch
 
+from yt_dlp.globals import all_plugins_loaded
+
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 
@@ -1427,6 +1429,12 @@ class TestYoutubeDL(unittest.TestCase):
         self.assertFalse(result.get('cookies'), msg='Cookies set in cookies field for wrong domain')
         self.assertFalse(ydl.cookiejar.get_cookie_header(fmt['url']), msg='Cookies set in cookiejar for wrong domain')
 
+    def test_load_plugins_compat(self):
+        # Should try to reload plugins if they haven't already been loaded
+        all_plugins_loaded.value = False
+        FakeYDL().close()
+        assert all_plugins_loaded.value
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/test/test_plugins.py b/test/test_plugins.py
index 77545d136..195726b18 100644
--- a/test/test_plugins.py
+++ b/test/test_plugins.py
@@ -10,22 +10,71 @@ TEST_DATA_DIR = Path(os.path.dirname(os.path.abspath(__file__)), 'testdata')
 sys.path.append(str(TEST_DATA_DIR))
 importlib.invalidate_caches()
 
-from yt_dlp.utils import Config
-from yt_dlp.plugins import PACKAGE_NAME, directories, load_plugins
+from yt_dlp.plugins import (
+    PACKAGE_NAME,
+    PluginSpec,
+    directories,
+    load_plugins,
+    load_all_plugins,
+    register_plugin_spec,
+)
+
+from yt_dlp.globals import (
+    extractors,
+    postprocessors,
+    plugin_dirs,
+    plugin_ies,
+    plugin_pps,
+    all_plugins_loaded,
+    plugin_specs,
+)
+
+
+EXTRACTOR_PLUGIN_SPEC = PluginSpec(
+    module_name='extractor',
+    suffix='IE',
+    destination=extractors,
+    plugin_destination=plugin_ies,
+)
+
+POSTPROCESSOR_PLUGIN_SPEC = PluginSpec(
+    module_name='postprocessor',
+    suffix='PP',
+    destination=postprocessors,
+    plugin_destination=plugin_pps,
+)
+
+
+def reset_plugins():
+    plugin_ies.value = {}
+    plugin_pps.value = {}
+    plugin_dirs.value = ['default']
+    plugin_specs.value = {}
+    all_plugins_loaded.value = False
+    # Clearing override plugins is probably difficult
+    for module_name in tuple(sys.modules):
+        for plugin_type in ('extractor', 'postprocessor'):
+            if module_name.startswith(f'{PACKAGE_NAME}.{plugin_type}.'):
+                del sys.modules[module_name]
+
+    importlib.invalidate_caches()
 
 
 class TestPlugins(unittest.TestCase):
 
     TEST_PLUGIN_DIR = TEST_DATA_DIR / PACKAGE_NAME
 
+    def setUp(self):
+        reset_plugins()
+
+    def tearDown(self):
+        reset_plugins()
+
     def test_directories_containing_plugins(self):
         self.assertIn(self.TEST_PLUGIN_DIR, map(Path, directories()))
 
     def test_extractor_classes(self):
-        for module_name in tuple(sys.modules):
-            if module_name.startswith(f'{PACKAGE_NAME}.extractor'):
-                del sys.modules[module_name]
-        plugins_ie = load_plugins('extractor', 'IE')
+        plugins_ie = load_plugins(EXTRACTOR_PLUGIN_SPEC)
 
         self.assertIn(f'{PACKAGE_NAME}.extractor.normal', sys.modules.keys())
         self.assertIn('NormalPluginIE', plugins_ie.keys())
@@ -35,17 +84,29 @@ class TestPlugins(unittest.TestCase):
             f'{PACKAGE_NAME}.extractor._ignore' in sys.modules,
             'loaded module beginning with underscore')
         self.assertNotIn('IgnorePluginIE', plugins_ie.keys())
+        self.assertNotIn('IgnorePluginIE', plugin_ies.value)
 
         # Don't load extractors with underscore prefix
         self.assertNotIn('_IgnoreUnderscorePluginIE', plugins_ie.keys())
+        self.assertNotIn('_IgnoreUnderscorePluginIE', plugin_ies.value)
 
         # Don't load extractors not specified in __all__ (if supplied)
         self.assertNotIn('IgnoreNotInAllPluginIE', plugins_ie.keys())
+        self.assertNotIn('IgnoreNotInAllPluginIE', plugin_ies.value)
         self.assertIn('InAllPluginIE', plugins_ie.keys())
+        self.assertIn('InAllPluginIE', plugin_ies.value)
+
+        # Don't load override extractors
+        self.assertNotIn('OverrideGenericIE', plugins_ie.keys())
+        self.assertNotIn('OverrideGenericIE', plugin_ies.value)
+        self.assertNotIn('_UnderscoreOverrideGenericIE', plugins_ie.keys())
+        self.assertNotIn('_UnderscoreOverrideGenericIE', plugin_ies.value)
 
     def test_postprocessor_classes(self):
-        plugins_pp = load_plugins('postprocessor', 'PP')
+        plugins_pp = load_plugins(POSTPROCESSOR_PLUGIN_SPEC)
         self.assertIn('NormalPluginPP', plugins_pp.keys())
+        self.assertIn(f'{PACKAGE_NAME}.postprocessor.normal', sys.modules.keys())
+        self.assertIn('NormalPluginPP', plugin_pps.value)
 
     def test_importing_zipped_module(self):
         zip_path = TEST_DATA_DIR / 'zipped_plugins.zip'
@@ -58,10 +119,10 @@ class TestPlugins(unittest.TestCase):
                 package = importlib.import_module(f'{PACKAGE_NAME}.{plugin_type}')
                 self.assertIn(zip_path / PACKAGE_NAME / plugin_type, map(Path, package.__path__))
 
-            plugins_ie = load_plugins('extractor', 'IE')
+            plugins_ie = load_plugins(EXTRACTOR_PLUGIN_SPEC)
             self.assertIn('ZippedPluginIE', plugins_ie.keys())
 
-            plugins_pp = load_plugins('postprocessor', 'PP')
+            plugins_pp = load_plugins(POSTPROCESSOR_PLUGIN_SPEC)
             self.assertIn('ZippedPluginPP', plugins_pp.keys())
 
         finally:
@@ -69,23 +130,116 @@ class TestPlugins(unittest.TestCase):
             os.remove(zip_path)
             importlib.invalidate_caches()  # reset the import caches
 
-    def test_plugin_dirs(self):
-        # Internal plugin dirs hack for CLI --plugin-dirs
-        # To be replaced with proper system later
-        custom_plugin_dir = TEST_DATA_DIR / 'plugin_packages'
-        Config._plugin_dirs = [str(custom_plugin_dir)]
-        importlib.invalidate_caches()  # reset the import caches
+    def test_reloading_plugins(self):
+        reload_plugins_path = TEST_DATA_DIR / 'reload_plugins'
+        load_plugins(EXTRACTOR_PLUGIN_SPEC)
+        load_plugins(POSTPROCESSOR_PLUGIN_SPEC)
 
+        # Remove default folder and add reload_plugin path
+        sys.path.remove(str(TEST_DATA_DIR))
+        sys.path.append(str(reload_plugins_path))
+        importlib.invalidate_caches()
         try:
-            package = importlib.import_module(f'{PACKAGE_NAME}.extractor')
-            self.assertIn(custom_plugin_dir / 'testpackage' / PACKAGE_NAME / 'extractor', map(Path, package.__path__))
-
-            plugins_ie = load_plugins('extractor', 'IE')
-            self.assertIn('PackagePluginIE', plugins_ie.keys())
+            for plugin_type in ('extractor', 'postprocessor'):
+                package = importlib.import_module(f'{PACKAGE_NAME}.{plugin_type}')
+                self.assertIn(reload_plugins_path / PACKAGE_NAME / plugin_type, map(Path, package.__path__))
+
+            plugins_ie = load_plugins(EXTRACTOR_PLUGIN_SPEC)
+            self.assertIn('NormalPluginIE', plugins_ie.keys())
+            self.assertTrue(
+                plugins_ie['NormalPluginIE'].REPLACED,
+                msg='Reloading has not replaced original extractor plugin')
+            self.assertTrue(
+                extractors.value['NormalPluginIE'].REPLACED,
+                msg='Reloading has not replaced original extractor plugin globally')
+
+            plugins_pp = load_plugins(POSTPROCESSOR_PLUGIN_SPEC)
+            self.assertIn('NormalPluginPP', plugins_pp.keys())
+            self.assertTrue(plugins_pp['NormalPluginPP'].REPLACED,
+                            msg='Reloading has not replaced original postprocessor plugin')
+            self.assertTrue(
+                postprocessors.value['NormalPluginPP'].REPLACED,
+                msg='Reloading has not replaced original postprocessor plugin globally')
 
         finally:
-            Config._plugin_dirs = []
-            importlib.invalidate_caches()  # reset the import caches
+            sys.path.remove(str(reload_plugins_path))
+            sys.path.append(str(TEST_DATA_DIR))
+            importlib.invalidate_caches()
+
+    def test_extractor_override_plugin(self):
+        load_plugins(EXTRACTOR_PLUGIN_SPEC)
+
+        from yt_dlp.extractor.generic import GenericIE
+
+        self.assertEqual(GenericIE.TEST_FIELD, 'override')
+        self.assertEqual(GenericIE.SECONDARY_TEST_FIELD, 'underscore-override')
+
+        self.assertEqual(GenericIE.IE_NAME, 'generic+override+underscore-override')
+        importlib.invalidate_caches()
+        #  test that loading a second time doesn't wrap a second time
+        load_plugins(EXTRACTOR_PLUGIN_SPEC)
+        from yt_dlp.extractor.generic import GenericIE
+        self.assertEqual(GenericIE.IE_NAME, 'generic+override+underscore-override')
+
+    def test_load_all_plugin_types(self):
+
+        # no plugin specs registered
+        load_all_plugins()
+
+        self.assertNotIn(f'{PACKAGE_NAME}.extractor.normal', sys.modules.keys())
+        self.assertNotIn(f'{PACKAGE_NAME}.postprocessor.normal', sys.modules.keys())
+
+        register_plugin_spec(EXTRACTOR_PLUGIN_SPEC)
+        register_plugin_spec(POSTPROCESSOR_PLUGIN_SPEC)
+        load_all_plugins()
+        self.assertTrue(all_plugins_loaded.value)
+
+        self.assertIn(f'{PACKAGE_NAME}.extractor.normal', sys.modules.keys())
+        self.assertIn(f'{PACKAGE_NAME}.postprocessor.normal', sys.modules.keys())
+
+    def test_no_plugin_dirs(self):
+        register_plugin_spec(EXTRACTOR_PLUGIN_SPEC)
+        register_plugin_spec(POSTPROCESSOR_PLUGIN_SPEC)
+
+        plugin_dirs.value = []
+        load_all_plugins()
+
+        self.assertNotIn(f'{PACKAGE_NAME}.extractor.normal', sys.modules.keys())
+        self.assertNotIn(f'{PACKAGE_NAME}.postprocessor.normal', sys.modules.keys())
+
+    def test_set_plugin_dirs(self):
+        custom_plugin_dir = str(TEST_DATA_DIR / 'plugin_packages')
+        plugin_dirs.value = [custom_plugin_dir]
+
+        load_plugins(EXTRACTOR_PLUGIN_SPEC)
+
+        self.assertIn(f'{PACKAGE_NAME}.extractor.package', sys.modules.keys())
+        self.assertIn('PackagePluginIE', plugin_ies.value)
+
+    def test_invalid_plugin_dir(self):
+        plugin_dirs.value = ['invalid_dir']
+        with self.assertRaises(ValueError):
+            load_plugins(EXTRACTOR_PLUGIN_SPEC)
+
+    def test_append_plugin_dirs(self):
+        custom_plugin_dir = str(TEST_DATA_DIR / 'plugin_packages')
+
+        self.assertEqual(plugin_dirs.value, ['default'])
+        plugin_dirs.value.append(custom_plugin_dir)
+        self.assertEqual(plugin_dirs.value, ['default', custom_plugin_dir])
+
+        load_plugins(EXTRACTOR_PLUGIN_SPEC)
+
+        self.assertIn(f'{PACKAGE_NAME}.extractor.package', sys.modules.keys())
+        self.assertIn('PackagePluginIE', plugin_ies.value)
+
+    def test_get_plugin_spec(self):
+        register_plugin_spec(EXTRACTOR_PLUGIN_SPEC)
+        register_plugin_spec(POSTPROCESSOR_PLUGIN_SPEC)
+
+        self.assertEqual(plugin_specs.value.get('extractor'), EXTRACTOR_PLUGIN_SPEC)
+        self.assertEqual(plugin_specs.value.get('postprocessor'), POSTPROCESSOR_PLUGIN_SPEC)
+        self.assertIsNone(plugin_specs.value.get('invalid'))
 
 
 if __name__ == '__main__':
diff --git a/test/testdata/plugin_packages/testpackage/yt_dlp_plugins/extractor/package.py b/test/testdata/plugin_packages/testpackage/yt_dlp_plugins/extractor/package.py
index b860300d8..39020fef9 100644
--- a/test/testdata/plugin_packages/testpackage/yt_dlp_plugins/extractor/package.py
+++ b/test/testdata/plugin_packages/testpackage/yt_dlp_plugins/extractor/package.py
@@ -2,4 +2,5 @@ from yt_dlp.extractor.common import InfoExtractor
 
 
 class PackagePluginIE(InfoExtractor):
+    _VALID_URL = 'package'
     pass
diff --git a/test/testdata/reload_plugins/yt_dlp_plugins/extractor/normal.py b/test/testdata/reload_plugins/yt_dlp_plugins/extractor/normal.py
new file mode 100644
index 000000000..6b927077f
--- /dev/null
+++ b/test/testdata/reload_plugins/yt_dlp_plugins/extractor/normal.py
@@ -0,0 +1,10 @@
+from yt_dlp.extractor.common import InfoExtractor
+
+
+class NormalPluginIE(InfoExtractor):
+    _VALID_URL = 'normal'
+    REPLACED = True
+
+
+class _IgnoreUnderscorePluginIE(InfoExtractor):
+    pass
diff --git a/test/testdata/reload_plugins/yt_dlp_plugins/postprocessor/normal.py b/test/testdata/reload_plugins/yt_dlp_plugins/postprocessor/normal.py
new file mode 100644
index 000000000..5e44ba2b5
--- /dev/null
+++ b/test/testdata/reload_plugins/yt_dlp_plugins/postprocessor/normal.py
@@ -0,0 +1,5 @@
+from yt_dlp.postprocessor.common import PostProcessor
+
+
+class NormalPluginPP(PostProcessor):
+    REPLACED = True
diff --git a/test/testdata/yt_dlp_plugins/extractor/ignore.py b/test/testdata/yt_dlp_plugins/extractor/ignore.py
index 816a16aa2..dca111a37 100644
--- a/test/testdata/yt_dlp_plugins/extractor/ignore.py
+++ b/test/testdata/yt_dlp_plugins/extractor/ignore.py
@@ -6,6 +6,7 @@ class IgnoreNotInAllPluginIE(InfoExtractor):
 
 
 class InAllPluginIE(InfoExtractor):
+    _VALID_URL = 'inallpluginie'
     pass
 
 
diff --git a/test/testdata/yt_dlp_plugins/extractor/normal.py b/test/testdata/yt_dlp_plugins/extractor/normal.py
index b09009bdc..996b2936f 100644
--- a/test/testdata/yt_dlp_plugins/extractor/normal.py
+++ b/test/testdata/yt_dlp_plugins/extractor/normal.py
@@ -2,8 +2,10 @@ from yt_dlp.extractor.common import InfoExtractor
 
 
 class NormalPluginIE(InfoExtractor):
-    pass
+    _VALID_URL = 'normalpluginie'
+    REPLACED = False
 
 
 class _IgnoreUnderscorePluginIE(InfoExtractor):
+    _VALID_URL = 'ignoreunderscorepluginie'
     pass
diff --git a/test/testdata/yt_dlp_plugins/extractor/override.py b/test/testdata/yt_dlp_plugins/extractor/override.py
new file mode 100644
index 000000000..766dc32e1
--- /dev/null
+++ b/test/testdata/yt_dlp_plugins/extractor/override.py
@@ -0,0 +1,5 @@
+from yt_dlp.extractor.generic import GenericIE
+
+
+class OverrideGenericIE(GenericIE, plugin_name='override'):
+    TEST_FIELD = 'override'
diff --git a/test/testdata/yt_dlp_plugins/extractor/overridetwo.py b/test/testdata/yt_dlp_plugins/extractor/overridetwo.py
new file mode 100644
index 000000000..826184c64
--- /dev/null
+++ b/test/testdata/yt_dlp_plugins/extractor/overridetwo.py
@@ -0,0 +1,5 @@
+from yt_dlp.extractor.generic import GenericIE
+
+
+class _UnderscoreOverrideGenericIE(GenericIE, plugin_name='underscore-override'):
+    SECONDARY_TEST_FIELD = 'underscore-override'
diff --git a/test/testdata/yt_dlp_plugins/postprocessor/normal.py b/test/testdata/yt_dlp_plugins/postprocessor/normal.py
index 315b85a48..1e94d7b8b 100644
--- a/test/testdata/yt_dlp_plugins/postprocessor/normal.py
+++ b/test/testdata/yt_dlp_plugins/postprocessor/normal.py
@@ -2,4 +2,4 @@ from yt_dlp.postprocessor.common import PostProcessor
 
 
 class NormalPluginPP(PostProcessor):
-    pass
+    REPLACED = False
diff --git a/test/testdata/zipped_plugins/yt_dlp_plugins/extractor/zipped.py b/test/testdata/zipped_plugins/yt_dlp_plugins/extractor/zipped.py
index 01542e0d8..c5140bb02 100644
--- a/test/testdata/zipped_plugins/yt_dlp_plugins/extractor/zipped.py
+++ b/test/testdata/zipped_plugins/yt_dlp_plugins/extractor/zipped.py
@@ -2,4 +2,5 @@ from yt_dlp.extractor.common import InfoExtractor
 
 
 class ZippedPluginIE(InfoExtractor):
+    _VALID_URL = 'zippedpluginie'
     pass
diff --git a/yt_dlp/YoutubeDL.py b/yt_dlp/YoutubeDL.py
index 7026822b6..8790b326b 100644
--- a/yt_dlp/YoutubeDL.py
+++ b/yt_dlp/YoutubeDL.py
@@ -30,9 +30,18 @@ from .compat import urllib_req_to_req
 from .cookies import CookieLoadError, LenientSimpleCookie, load_cookies
 from .downloader import FFmpegFD, get_suitable_downloader, shorten_protocol_name
 from .downloader.rtmp import rtmpdump_version
-from .extractor import gen_extractor_classes, get_info_extractor
+from .extractor import gen_extractor_classes, get_info_extractor, import_extractors
 from .extractor.common import UnsupportedURLIE
 from .extractor.openload import PhantomJSwrapper
+from .globals import (
+    IN_CLI,
+    LAZY_EXTRACTORS,
+    plugin_ies,
+    plugin_ies_overrides,
+    plugin_pps,
+    all_plugins_loaded,
+    plugin_dirs,
+)
 from .minicurses import format_text
 from .networking import HEADRequest, Request, RequestDirector
 from .networking.common import _REQUEST_HANDLERS, _RH_PREFERENCES
@@ -44,8 +53,7 @@ from .networking.exceptions import (
     network_exceptions,
 )
 from .networking.impersonate import ImpersonateRequestHandler
-from .plugins import directories as plugin_directories
-from .postprocessor import _PLUGIN_CLASSES as plugin_pps
+from .plugins import directories as plugin_directories, load_all_plugins
 from .postprocessor import (
     EmbedThumbnailPP,
     FFmpegFixupDuplicateMoovPP,
@@ -642,6 +650,10 @@ class YoutubeDL:
         self.cache = Cache(self)
         self.__header_cookies = []
 
+        # compat for API: load plugins if they have not already
+        if not all_plugins_loaded.value:
+            load_all_plugins()
+
         try:
             windows_enable_vt_mode()
         except Exception as e:
@@ -3995,15 +4007,6 @@ class YoutubeDL:
         if not self.params.get('verbose'):
             return
 
-        from . import _IN_CLI  # Must be delayed import
-
-        # These imports can be slow. So import them only as needed
-        from .extractor.extractors import _LAZY_LOADER
-        from .extractor.extractors import (
-            _PLUGIN_CLASSES as plugin_ies,
-            _PLUGIN_OVERRIDES as plugin_ie_overrides,
-        )
-
         def get_encoding(stream):
             ret = str(getattr(stream, 'encoding', f'missing ({type(stream).__name__})'))
             additional_info = []
@@ -4042,17 +4045,18 @@ class YoutubeDL:
             _make_label(ORIGIN, CHANNEL.partition('@')[2] or __version__, __version__),
             f'[{RELEASE_GIT_HEAD[:9]}]' if RELEASE_GIT_HEAD else '',
             '' if source == 'unknown' else f'({source})',
-            '' if _IN_CLI else 'API' if klass == YoutubeDL else f'API:{self.__module__}.{klass.__qualname__}',
+            '' if IN_CLI.value else 'API' if klass == YoutubeDL else f'API:{self.__module__}.{klass.__qualname__}',
             delim=' '))
 
-        if not _IN_CLI:
+        if not IN_CLI.value:
             write_debug(f'params: {self.params}')
 
-        if not _LAZY_LOADER:
-            if os.environ.get('YTDLP_NO_LAZY_EXTRACTORS'):
-                write_debug('Lazy loading extractors is forcibly disabled')
-            else:
-                write_debug('Lazy loading extractors is disabled')
+        import_extractors()
+        lazy_extractors = LAZY_EXTRACTORS.value
+        if lazy_extractors is None:
+            write_debug('Lazy loading extractors is disabled')
+        elif not lazy_extractors:
+            write_debug('Lazy loading extractors is forcibly disabled')
         if self.params['compat_opts']:
             write_debug('Compatibility options: {}'.format(', '.join(self.params['compat_opts'])))
 
@@ -4081,24 +4085,27 @@ class YoutubeDL:
 
         write_debug(f'Proxy map: {self.proxies}')
         write_debug(f'Request Handlers: {", ".join(rh.RH_NAME for rh in self._request_director.handlers.values())}')
-        if os.environ.get('YTDLP_NO_PLUGINS'):
-            write_debug('Plugins are forcibly disabled')
-            return
 
-        for plugin_type, plugins in {'Extractor': plugin_ies, 'Post-Processor': plugin_pps}.items():
-            display_list = ['{}{}'.format(
-                klass.__name__, '' if klass.__name__ == name else f' as {name}')
-                for name, klass in plugins.items()]
+        for plugin_type, plugins in (('Extractor', plugin_ies), ('Post-Processor', plugin_pps)):
+            display_list = [
+                klass.__name__ if klass.__name__ == name else f'{klass.__name__} as {name}'
+                for name, klass in plugins.value.items()]
             if plugin_type == 'Extractor':
                 display_list.extend(f'{plugins[-1].IE_NAME.partition("+")[2]} ({parent.__name__})'
-                                    for parent, plugins in plugin_ie_overrides.items())
+                                    for parent, plugins in plugin_ies_overrides.value.items())
             if not display_list:
                 continue
             write_debug(f'{plugin_type} Plugins: {", ".join(sorted(display_list))}')
 
-        plugin_dirs = plugin_directories()
-        if plugin_dirs:
-            write_debug(f'Plugin directories: {plugin_dirs}')
+        plugin_dirs_msg = 'none'
+        if not plugin_dirs.value:
+            plugin_dirs_msg = 'none (disabled)'
+        else:
+            found_plugin_directories = plugin_directories()
+            if found_plugin_directories:
+                plugin_dirs_msg = ', '.join(found_plugin_directories)
+
+        write_debug(f'Plugin directories: {plugin_dirs_msg}')
 
     @functools.cached_property
     def proxies(self):
diff --git a/yt_dlp/__init__.py b/yt_dlp/__init__.py
index 3819656e2..7d8f10047 100644
--- a/yt_dlp/__init__.py
+++ b/yt_dlp/__init__.py
@@ -19,7 +19,9 @@ from .downloader.external import get_external_downloader
 from .extractor import list_extractor_classes
 from .extractor.adobepass import MSO_INFO
 from .networking.impersonate import ImpersonateTarget
+from .globals import IN_CLI, plugin_dirs
 from .options import parseOpts
+from .plugins import load_all_plugins as _load_all_plugins
 from .postprocessor import (
     FFmpegExtractAudioPP,
     FFmpegMergerPP,
@@ -33,7 +35,6 @@ from .postprocessor import (
 )
 from .update import Updater
 from .utils import (
-    Config,
     NO_DEFAULT,
     POSTPROCESS_WHEN,
     DateRange,
@@ -66,8 +67,6 @@ from .utils.networking import std_headers
 from .utils._utils import _UnsafeExtensionError
 from .YoutubeDL import YoutubeDL
 
-_IN_CLI = False
-
 
 def _exit(status=0, *args):
     for msg in args:
@@ -433,6 +432,10 @@ def validate_options(opts):
     }
 
     # Other options
+    opts.plugin_dirs = opts.plugin_dirs
+    if opts.plugin_dirs is None:
+        opts.plugin_dirs = ['default']
+
     if opts.playlist_items is not None:
         try:
             tuple(PlaylistEntries.parse_playlist_items(opts.playlist_items))
@@ -973,11 +976,6 @@ def _real_main(argv=None):
 
     parser, opts, all_urls, ydl_opts = parse_options(argv)
 
-    # HACK: Set the plugin dirs early on
-    # TODO(coletdjnz): remove when plugin globals system is implemented
-    if opts.plugin_dirs is not None:
-        Config._plugin_dirs = list(map(expand_path, opts.plugin_dirs))
-
     # Dump user agent
     if opts.dump_user_agent:
         ua = traverse_obj(opts.headers, 'User-Agent', casesense=False, default=std_headers['User-Agent'])
@@ -992,6 +990,11 @@ def _real_main(argv=None):
     if opts.ffmpeg_location:
         FFmpegPostProcessor._ffmpeg_location.set(opts.ffmpeg_location)
 
+    # load all plugins into the global lookup
+    plugin_dirs.value = opts.plugin_dirs
+    if plugin_dirs.value:
+        _load_all_plugins()
+
     with YoutubeDL(ydl_opts) as ydl:
         pre_process = opts.update_self or opts.rm_cachedir
         actual_use = all_urls or opts.load_info_filename
@@ -1091,8 +1094,7 @@ def _real_main(argv=None):
 
 
 def main(argv=None):
-    global _IN_CLI
-    _IN_CLI = True
+    IN_CLI.value = True
     try:
         _exit(*variadic(_real_main(argv)))
     except (CookieLoadError, DownloadError):
diff --git a/yt_dlp/extractor/__init__.py b/yt_dlp/extractor/__init__.py
index 6bfa4bd7b..a090e942d 100644
--- a/yt_dlp/extractor/__init__.py
+++ b/yt_dlp/extractor/__init__.py
@@ -1,16 +1,25 @@
 from ..compat.compat_utils import passthrough_module
+from ..globals import extractors as _extractors_context
+from ..globals import plugin_ies as _plugin_ies_context
+from ..plugins import PluginSpec, register_plugin_spec
 
 passthrough_module(__name__, '.extractors')
 del passthrough_module
 
+register_plugin_spec(PluginSpec(
+    module_name='extractor',
+    suffix='IE',
+    destination=_extractors_context,
+    plugin_destination=_plugin_ies_context,
+))
+
 
 def gen_extractor_classes():
     """ Return a list of supported extractors.
     The order does matter; the first extractor matched is the one handling the URL.
     """
-    from .extractors import _ALL_CLASSES
-
-    return _ALL_CLASSES
+    import_extractors()
+    return list(_extractors_context.value.values())
 
 
 def gen_extractors():
@@ -37,6 +46,9 @@ def list_extractors(age_limit=None):
 
 def get_info_extractor(ie_name):
     """Returns the info extractor class with the given ie_name"""
-    from . import extractors
+    import_extractors()
+    return _extractors_context.value[f'{ie_name}IE']
+
 
-    return getattr(extractors, f'{ie_name}IE')
+def import_extractors():
+    from . import extractors  # noqa: F401
diff --git a/yt_dlp/extractor/common.py b/yt_dlp/extractor/common.py
index 8d199b353..b816d788f 100644
--- a/yt_dlp/extractor/common.py
+++ b/yt_dlp/extractor/common.py
@@ -29,6 +29,7 @@ from ..compat import (
 from ..cookies import LenientSimpleCookie
 from ..downloader.f4m import get_base_url, remove_encrypted_media
 from ..downloader.hls import HlsFD
+from ..globals import plugin_ies_overrides
 from ..networking import HEADRequest, Request
 from ..networking.exceptions import (
     HTTPError,
@@ -3954,14 +3955,18 @@ class InfoExtractor:
     def __init_subclass__(cls, *, plugin_name=None, **kwargs):
         if plugin_name:
             mro = inspect.getmro(cls)
-            super_class = cls.__wrapped__ = mro[mro.index(cls) + 1]
-            cls.PLUGIN_NAME, cls.ie_key = plugin_name, super_class.ie_key
-            cls.IE_NAME = f'{super_class.IE_NAME}+{plugin_name}'
+            next_mro_class = super_class = mro[mro.index(cls) + 1]
+
             while getattr(super_class, '__wrapped__', None):
                 super_class = super_class.__wrapped__
-            setattr(sys.modules[super_class.__module__], super_class.__name__, cls)
-            _PLUGIN_OVERRIDES[super_class].append(cls)
 
+            if not any(override.PLUGIN_NAME == plugin_name for override in plugin_ies_overrides.value[super_class]):
+                cls.__wrapped__ = next_mro_class
+                cls.PLUGIN_NAME, cls.ie_key = plugin_name, next_mro_class.ie_key
+                cls.IE_NAME = f'{next_mro_class.IE_NAME}+{plugin_name}'
+
+                setattr(sys.modules[super_class.__module__], super_class.__name__, cls)
+                plugin_ies_overrides.value[super_class].append(cls)
         return super().__init_subclass__(**kwargs)
 
 
@@ -4017,6 +4022,3 @@ class UnsupportedURLIE(InfoExtractor):
 
     def _real_extract(self, url):
         raise UnsupportedError(url)
-
-
-_PLUGIN_OVERRIDES = collections.defaultdict(list)
diff --git a/yt_dlp/extractor/extractors.py b/yt_dlp/extractor/extractors.py
index baa69d242..050bed2da 100644
--- a/yt_dlp/extractor/extractors.py
+++ b/yt_dlp/extractor/extractors.py
@@ -1,28 +1,35 @@
-import contextlib
+import inspect
 import os
 
-from ..plugins import load_plugins
+from ..globals import LAZY_EXTRACTORS
+from ..globals import extractors as _extractors_context
 
-# NB: Must be before other imports so that plugins can be correctly injected
-_PLUGIN_CLASSES = load_plugins('extractor', 'IE')
-
-_LAZY_LOADER = False
+_CLASS_LOOKUP = None
 if not os.environ.get('YTDLP_NO_LAZY_EXTRACTORS'):
-    with contextlib.suppress(ImportError):
-        from .lazy_extractors import *  # noqa: F403
-        from .lazy_extractors import _ALL_CLASSES
-        _LAZY_LOADER = True
+    try:
+        from .lazy_extractors import _CLASS_LOOKUP
+        LAZY_EXTRACTORS.value = True
+    except ImportError:
+        LAZY_EXTRACTORS.value = False
+
+if not _CLASS_LOOKUP:
+    from . import _extractors
 
-if not _LAZY_LOADER:
-    from ._extractors import *  # noqa: F403
-    _ALL_CLASSES = [  # noqa: F811
-        klass
-        for name, klass in globals().items()
+    _CLASS_LOOKUP = {
+        name: value
+        for name, value in inspect.getmembers(_extractors)
         if name.endswith('IE') and name != 'GenericIE'
-    ]
-    _ALL_CLASSES.append(GenericIE)  # noqa: F405
+    }
+    _CLASS_LOOKUP['GenericIE'] = _extractors.GenericIE
+
+# We want to append to the main lookup
+_current = _extractors_context.value
+for name, ie in _CLASS_LOOKUP.items():
+    _current.setdefault(name, ie)
 
-globals().update(_PLUGIN_CLASSES)
-_ALL_CLASSES[:0] = _PLUGIN_CLASSES.values()
 
-from .common import _PLUGIN_OVERRIDES  # noqa: F401
+def __getattr__(name):
+    value = _CLASS_LOOKUP.get(name)
+    if not value:
+        raise AttributeError(f'module {__name__} has no attribute {name}')
+    return value
diff --git a/yt_dlp/globals.py b/yt_dlp/globals.py
new file mode 100644
index 000000000..e1c189d5a
--- /dev/null
+++ b/yt_dlp/globals.py
@@ -0,0 +1,30 @@
+from collections import defaultdict
+
+# Please Note: Due to necessary changes and the complex nature involved in the plugin/globals system,
+# no backwards compatibility is guaranteed for the plugin system API.
+# However, we will still try our best.
+
+
+class Indirect:
+    def __init__(self, initial, /):
+        self.value = initial
+
+    def __repr__(self, /):
+        return f'{type(self).__name__}({self.value!r})'
+
+
+postprocessors = Indirect({})
+extractors = Indirect({})
+
+# Plugins
+all_plugins_loaded = Indirect(False)
+plugin_specs = Indirect({})
+plugin_dirs = Indirect(['default'])
+
+plugin_ies = Indirect({})
+plugin_pps = Indirect({})
+plugin_ies_overrides = Indirect(defaultdict(list))
+
+# Misc
+IN_CLI = Indirect(False)
+LAZY_EXTRACTORS = Indirect(False)  # `False`=force, `None`=disabled, `True`=enabled
diff --git a/yt_dlp/options.py b/yt_dlp/options.py
index 06b65e0ea..91c2635a7 100644
--- a/yt_dlp/options.py
+++ b/yt_dlp/options.py
@@ -398,7 +398,7 @@ def create_parser():
             '(Alias: --no-config)'))
     general.add_option(
         '--no-config-locations',
-        action='store_const', dest='config_locations', const=[],
+        action='store_const', dest='config_locations', const=None,
         help=(
             'Do not load any custom configuration files (default). When given inside a '
             'configuration file, ignore all previous --config-locations defined in the current file'))
@@ -410,12 +410,21 @@ def create_parser():
             '("-" for stdin). Can be used multiple times and inside other configuration files'))
     general.add_option(
         '--plugin-dirs',
-        dest='plugin_dirs', metavar='PATH', action='append',
+        metavar='PATH',
+        dest='plugin_dirs',
+        action='callback',
+        callback=_list_from_options_callback,
+        type='str',
+        callback_kwargs={'delim': None},
+        default=['default'],
         help=(
             'Path to an additional directory to search for plugins. '
             'This option can be used multiple times to add multiple directories. '
-            'Note that this currently only works for extractor plugins; '
-            'postprocessor plugins can only be loaded from the default plugin directories'))
+            'Use "default" to search the default plugin directories (default)'))
+    general.add_option(
+        '--no-plugin-dirs',
+        dest='plugin_dirs', action='store_const', const=[],
+        help='Clear plugin directories to search, including defaults and those provided by previous --plugin-dirs')
     general.add_option(
         '--flat-playlist',
         action='store_const', dest='extract_flat', const='in_playlist', default=False,
diff --git a/yt_dlp/plugins.py b/yt_dlp/plugins.py
index 94335a9a3..941709b21 100644
--- a/yt_dlp/plugins.py
+++ b/yt_dlp/plugins.py
@@ -1,4 +1,5 @@
 import contextlib
+import dataclasses
 import functools
 import importlib
 import importlib.abc
@@ -14,17 +15,48 @@ import zipimport
 from pathlib import Path
 from zipfile import ZipFile
 
+from .globals import (
+    Indirect,
+    plugin_dirs,
+    all_plugins_loaded,
+    plugin_specs,
+)
+
 from .utils import (
-    Config,
     get_executable_path,
     get_system_config_dirs,
     get_user_config_dirs,
+    merge_dicts,
     orderedSet,
     write_string,
 )
 
 PACKAGE_NAME = 'yt_dlp_plugins'
 COMPAT_PACKAGE_NAME = 'ytdlp_plugins'
+_BASE_PACKAGE_PATH = Path(__file__).parent
+
+
+# Please Note: Due to necessary changes and the complex nature involved,
+# no backwards compatibility is guaranteed for the plugin system API.
+# However, we will still try our best.
+
+__all__ = [
+    'COMPAT_PACKAGE_NAME',
+    'PACKAGE_NAME',
+    'PluginSpec',
+    'directories',
+    'load_all_plugins',
+    'load_plugins',
+    'register_plugin_spec',
+]
+
+
+@dataclasses.dataclass
+class PluginSpec:
+    module_name: str
+    suffix: str
+    destination: Indirect
+    plugin_destination: Indirect
 
 
 class PluginLoader(importlib.abc.Loader):
@@ -44,7 +76,42 @@ def dirs_in_zip(archive):
         pass
     except Exception as e:
         write_string(f'WARNING: Could not read zip file {archive}: {e}\n')
-    return set()
+    return ()
+
+
+def default_plugin_paths():
+    def _get_package_paths(*root_paths, containing_folder):
+        for config_dir in orderedSet(map(Path, root_paths), lazy=True):
+            # We need to filter the base path added when running __main__.py directly
+            if config_dir == _BASE_PACKAGE_PATH:
+                continue
+            with contextlib.suppress(OSError):
+                yield from (config_dir / containing_folder).iterdir()
+
+    # Load from yt-dlp config folders
+    yield from _get_package_paths(
+        *get_user_config_dirs('yt-dlp'),
+        *get_system_config_dirs('yt-dlp'),
+        containing_folder='plugins',
+    )
+
+    # Load from yt-dlp-plugins folders
+    yield from _get_package_paths(
+        get_executable_path(),
+        *get_user_config_dirs(''),
+        *get_system_config_dirs(''),
+        containing_folder='yt-dlp-plugins',
+    )
+
+    # Load from PYTHONPATH directories
+    yield from (path for path in map(Path, sys.path) if path != _BASE_PACKAGE_PATH)
+
+
+def candidate_plugin_paths(candidate):
+    candidate_path = Path(candidate)
+    if not candidate_path.is_dir():
+        raise ValueError(f'Invalid plugin directory: {candidate_path}')
+    yield from candidate_path.iterdir()
 
 
 class PluginFinder(importlib.abc.MetaPathFinder):
@@ -56,40 +123,16 @@ class PluginFinder(importlib.abc.MetaPathFinder):
 
     def __init__(self, *packages):
         self._zip_content_cache = {}
-        self.packages = set(itertools.chain.from_iterable(
-            itertools.accumulate(name.split('.'), lambda a, b: '.'.join((a, b)))
-            for name in packages))
+        self.packages = set(
+            itertools.chain.from_iterable(
+                itertools.accumulate(name.split('.'), lambda a, b: '.'.join((a, b)))
+                for name in packages))
 
     def search_locations(self, fullname):
-        candidate_locations = []
-
-        def _get_package_paths(*root_paths, containing_folder='plugins'):
-            for config_dir in orderedSet(map(Path, root_paths), lazy=True):
-                with contextlib.suppress(OSError):
-                    yield from (config_dir / containing_folder).iterdir()
-
-        # Load from yt-dlp config folders
-        candidate_locations.extend(_get_package_paths(
-            *get_user_config_dirs('yt-dlp'),
-            *get_system_config_dirs('yt-dlp'),
-            containing_folder='plugins'))
-
-        # Load from yt-dlp-plugins folders
-        candidate_locations.extend(_get_package_paths(
-            get_executable_path(),
-            *get_user_config_dirs(''),
-            *get_system_config_dirs(''),
-            containing_folder='yt-dlp-plugins'))
-
-        candidate_locations.extend(map(Path, sys.path))  # PYTHONPATH
-        with contextlib.suppress(ValueError):  # Added when running __main__.py directly
-            candidate_locations.remove(Path(__file__).parent)
-
-        # TODO(coletdjnz): remove when plugin globals system is implemented
-        if Config._plugin_dirs:
-            candidate_locations.extend(_get_package_paths(
-                *Config._plugin_dirs,
-                containing_folder=''))
+        candidate_locations = itertools.chain.from_iterable(
+            default_plugin_paths() if candidate == 'default' else candidate_plugin_paths(candidate)
+            for candidate in plugin_dirs.value
+        )
 
         parts = Path(*fullname.split('.'))
         for path in orderedSet(candidate_locations, lazy=True):
@@ -109,7 +152,8 @@ class PluginFinder(importlib.abc.MetaPathFinder):
 
         search_locations = list(map(str, self.search_locations(fullname)))
         if not search_locations:
-            return None
+            # Prevent using built-in meta finders for searching plugins.
+            raise ModuleNotFoundError(fullname)
 
         spec = importlib.machinery.ModuleSpec(fullname, PluginLoader(), is_package=True)
         spec.submodule_search_locations = search_locations
@@ -123,8 +167,10 @@ class PluginFinder(importlib.abc.MetaPathFinder):
 
 
 def directories():
-    spec = importlib.util.find_spec(PACKAGE_NAME)
-    return spec.submodule_search_locations if spec else []
+    with contextlib.suppress(ModuleNotFoundError):
+        if spec := importlib.util.find_spec(PACKAGE_NAME):
+            return list(spec.submodule_search_locations)
+    return []
 
 
 def iter_modules(subpackage):
@@ -134,19 +180,23 @@ def iter_modules(subpackage):
         yield from pkgutil.iter_modules(path=pkg.__path__, prefix=f'{fullname}.')
 
 
-def load_module(module, module_name, suffix):
+def get_regular_classes(module, module_name, suffix):
+    # Find standard public plugin classes (not overrides)
     return inspect.getmembers(module, lambda obj: (
         inspect.isclass(obj)
         and obj.__name__.endswith(suffix)
         and obj.__module__.startswith(module_name)
         and not obj.__name__.startswith('_')
-        and obj.__name__ in getattr(module, '__all__', [obj.__name__])))
+        and obj.__name__ in getattr(module, '__all__', [obj.__name__])
+        and getattr(obj, 'PLUGIN_NAME', None) is None
+    ))
 
 
-def load_plugins(name, suffix):
-    classes = {}
-    if os.environ.get('YTDLP_NO_PLUGINS'):
-        return classes
+def load_plugins(plugin_spec: PluginSpec):
+    name, suffix = plugin_spec.module_name, plugin_spec.suffix
+    regular_classes = {}
+    if os.environ.get('YTDLP_NO_PLUGINS') or not plugin_dirs.value:
+        return regular_classes
 
     for finder, module_name, _ in iter_modules(name):
         if any(x.startswith('_') for x in module_name.split('.')):
@@ -163,24 +213,42 @@ def load_plugins(name, suffix):
                 sys.modules[module_name] = module
                 spec.loader.exec_module(module)
         except Exception:
-            write_string(f'Error while importing module {module_name!r}\n{traceback.format_exc(limit=-1)}')
+            write_string(
+                f'Error while importing module {module_name!r}\n{traceback.format_exc(limit=-1)}',
+            )
             continue
-        classes.update(load_module(module, module_name, suffix))
+        regular_classes.update(get_regular_classes(module, module_name, suffix))
 
     # Compat: old plugin system using __init__.py
     # Note: plugins imported this way do not show up in directories()
     # nor are considered part of the yt_dlp_plugins namespace package
-    with contextlib.suppress(FileNotFoundError):
-        spec = importlib.util.spec_from_file_location(
-            name, Path(get_executable_path(), COMPAT_PACKAGE_NAME, name, '__init__.py'))
-        plugins = importlib.util.module_from_spec(spec)
-        sys.modules[spec.name] = plugins
-        spec.loader.exec_module(plugins)
-        classes.update(load_module(plugins, spec.name, suffix))
-
-    return classes
-
-
-sys.meta_path.insert(0, PluginFinder(f'{PACKAGE_NAME}.extractor', f'{PACKAGE_NAME}.postprocessor'))
-
-__all__ = ['COMPAT_PACKAGE_NAME', 'PACKAGE_NAME', 'directories', 'load_plugins']
+    if 'default' in plugin_dirs.value:
+        with contextlib.suppress(FileNotFoundError):
+            spec = importlib.util.spec_from_file_location(
+                name,
+                Path(get_executable_path(), COMPAT_PACKAGE_NAME, name, '__init__.py'),
+            )
+            plugins = importlib.util.module_from_spec(spec)
+            sys.modules[spec.name] = plugins
+            spec.loader.exec_module(plugins)
+            regular_classes.update(get_regular_classes(plugins, spec.name, suffix))
+
+    # Add the classes into the global plugin lookup for that type
+    plugin_spec.plugin_destination.value = regular_classes
+    # We want to prepend to the main lookup for that type
+    plugin_spec.destination.value = merge_dicts(regular_classes, plugin_spec.destination.value)
+
+    return regular_classes
+
+
+def load_all_plugins():
+    for plugin_spec in plugin_specs.value.values():
+        load_plugins(plugin_spec)
+    all_plugins_loaded.value = True
+
+
+def register_plugin_spec(plugin_spec: PluginSpec):
+    # If the plugin spec for a module is already registered, it will not be added again
+    if plugin_spec.module_name not in plugin_specs.value:
+        plugin_specs.value[plugin_spec.module_name] = plugin_spec
+        sys.meta_path.insert(0, PluginFinder(f'{PACKAGE_NAME}.{plugin_spec.module_name}'))
diff --git a/yt_dlp/postprocessor/__init__.py b/yt_dlp/postprocessor/__init__.py
index 7b1620544..20e8b14b2 100644
--- a/yt_dlp/postprocessor/__init__.py
+++ b/yt_dlp/postprocessor/__init__.py
@@ -33,15 +33,38 @@ from .movefilesafterdownload import MoveFilesAfterDownloadPP
 from .sponskrub import SponSkrubPP
 from .sponsorblock import SponsorBlockPP
 from .xattrpp import XAttrMetadataPP
-from ..plugins import load_plugins
+from ..globals import plugin_pps, postprocessors
+from ..plugins import PACKAGE_NAME, register_plugin_spec, PluginSpec
+from ..utils import deprecation_warning
 
-_PLUGIN_CLASSES = load_plugins('postprocessor', 'PP')
+
+def __getattr__(name):
+    lookup = plugin_pps.value
+    if name in lookup:
+        deprecation_warning(
+            f'Importing a plugin Post-Processor from {__name__} is deprecated. '
+            f'Please import {PACKAGE_NAME}.postprocessor.{name} instead.')
+        return lookup[name]
+
+    raise AttributeError(f'module {__name__!r} has no attribute {name!r}')
 
 
 def get_postprocessor(key):
-    return globals()[key + 'PP']
+    return postprocessors.value[key + 'PP']
+
+
+register_plugin_spec(PluginSpec(
+    module_name='postprocessor',
+    suffix='PP',
+    destination=postprocessors,
+    plugin_destination=plugin_pps,
+))
 
+_default_pps = {
+    name: value
+    for name, value in globals().items()
+    if name.endswith('PP') or name in ('FFmpegPostProcessor', 'PostProcessor')
+}
+postprocessors.value.update(_default_pps)
 
-globals().update(_PLUGIN_CLASSES)
-__all__ = [name for name in globals() if name.endswith('PP')]
-__all__.extend(('FFmpegPostProcessor', 'PostProcessor'))
+__all__ = list(_default_pps.values())
diff --git a/yt_dlp/utils/_utils.py b/yt_dlp/utils/_utils.py
index 3e7a375ee..4093c238c 100644
--- a/yt_dlp/utils/_utils.py
+++ b/yt_dlp/utils/_utils.py
@@ -52,6 +52,7 @@ from ..compat import (
     compat_HTMLParseError,
 )
 from ..dependencies import xattr
+from ..globals import IN_CLI
 
 __name__ = __name__.rsplit('.', 1)[0]  # noqa: A001: Pretend to be the parent module
 
@@ -1487,8 +1488,7 @@ def write_string(s, out=None, encoding=None):
 
 # TODO: Use global logger
 def deprecation_warning(msg, *, printer=None, stacklevel=0, **kwargs):
-    from .. import _IN_CLI
-    if _IN_CLI:
+    if IN_CLI.value:
         if msg in deprecation_warning._cache:
             return
         deprecation_warning._cache.add(msg)
@@ -4891,10 +4891,6 @@ class Config:
     filename = None
     __initialized = False
 
-    # Internal only, do not use! Hack to enable --plugin-dirs
-    # TODO(coletdjnz): remove when plugin globals system is implemented
-    _plugin_dirs = None
-
     def __init__(self, parser, label=None):
         self.parser, self.label = parser, label
         self._loaded_paths, self.configs = set(), []