mirror of
https://github.com/yt-dlp/yt-dlp.git
synced 2025-11-04 08:35:12 +00:00
4445f37a7a
- Adds `--no-plugin-dirs` to disable plugin loading - `--plugin-dirs` now supports post-processors Authored by: coletdjnz, Grub4K, pukkandan
255 lines
8.5 KiB
Python
255 lines
8.5 KiB
Python
import contextlib
|
|
import dataclasses
|
|
import functools
|
|
import importlib
|
|
import importlib.abc
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import inspect
|
|
import itertools
|
|
import os
|
|
import pkgutil
|
|
import sys
|
|
import traceback
|
|
import zipimport
|
|
from pathlib import Path
|
|
from zipfile import ZipFile
|
|
|
|
from .globals import (
|
|
Indirect,
|
|
plugin_dirs,
|
|
all_plugins_loaded,
|
|
plugin_specs,
|
|
)
|
|
|
|
from .utils import (
|
|
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):
|
|
"""Dummy loader for virtual namespace packages"""
|
|
|
|
def exec_module(self, module):
|
|
return None
|
|
|
|
|
|
@functools.cache
|
|
def dirs_in_zip(archive):
|
|
try:
|
|
with ZipFile(archive) as zip_:
|
|
return set(itertools.chain.from_iterable(
|
|
Path(file).parents for file in zip_.namelist()))
|
|
except FileNotFoundError:
|
|
pass
|
|
except Exception as e:
|
|
write_string(f'WARNING: Could not read zip file {archive}: {e}\n')
|
|
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):
|
|
"""
|
|
This class provides one or multiple namespace packages.
|
|
It searches in sys.path and yt-dlp config folders for
|
|
the existing subdirectories from which the modules can be imported
|
|
"""
|
|
|
|
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))
|
|
|
|
def search_locations(self, fullname):
|
|
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):
|
|
candidate = path / parts
|
|
try:
|
|
if candidate.is_dir():
|
|
yield candidate
|
|
elif path.suffix in ('.zip', '.egg', '.whl') and path.is_file():
|
|
if parts in dirs_in_zip(path):
|
|
yield candidate
|
|
except PermissionError as e:
|
|
write_string(f'Permission error while accessing modules in "{e.filename}"\n')
|
|
|
|
def find_spec(self, fullname, path=None, target=None):
|
|
if fullname not in self.packages:
|
|
return None
|
|
|
|
search_locations = list(map(str, self.search_locations(fullname)))
|
|
if not search_locations:
|
|
# 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
|
|
return spec
|
|
|
|
def invalidate_caches(self):
|
|
dirs_in_zip.cache_clear()
|
|
for package in self.packages:
|
|
if package in sys.modules:
|
|
del sys.modules[package]
|
|
|
|
|
|
def directories():
|
|
with contextlib.suppress(ModuleNotFoundError):
|
|
if spec := importlib.util.find_spec(PACKAGE_NAME):
|
|
return list(spec.submodule_search_locations)
|
|
return []
|
|
|
|
|
|
def iter_modules(subpackage):
|
|
fullname = f'{PACKAGE_NAME}.{subpackage}'
|
|
with contextlib.suppress(ModuleNotFoundError):
|
|
pkg = importlib.import_module(fullname)
|
|
yield from pkgutil.iter_modules(path=pkg.__path__, prefix=f'{fullname}.')
|
|
|
|
|
|
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 getattr(obj, 'PLUGIN_NAME', None) is None
|
|
))
|
|
|
|
|
|
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('.')):
|
|
continue
|
|
try:
|
|
if sys.version_info < (3, 10) and isinstance(finder, zipimport.zipimporter):
|
|
# zipimporter.load_module() is deprecated in 3.10 and removed in 3.12
|
|
# The exec_module branch below is the replacement for >= 3.10
|
|
# See: https://docs.python.org/3/library/zipimport.html#zipimport.zipimporter.exec_module
|
|
module = finder.load_module(module_name)
|
|
else:
|
|
spec = finder.find_spec(module_name)
|
|
module = importlib.util.module_from_spec(spec)
|
|
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)}',
|
|
)
|
|
continue
|
|
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
|
|
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}'))
|