# -*- coding: utf-8 -*- # Copyright © 2012-2024 Chris Warrick and others. # Permission is hereby granted, free of charge, to any # person obtaining a copy of this software and associated # documentation files (the "Software"), to deal in the # Software without restriction, including without limitation # the rights to use, copy, modify, merge, publish, # distribute, sublicense, and/or sell copies of the # Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice # shall be included in all copies or substantial portions of # the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY # KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE # WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR # PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS # OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR # OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR # OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE # SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """The Nikola plugin manager. Inspired by yapsy.""" import configparser import importlib import importlib.util import time import sys from dataclasses import dataclass from pathlib import Path from typing import Dict, List, Optional, Type, TYPE_CHECKING, Set from .plugin_categories import BasePlugin, CATEGORIES from .utils import get_logger if TYPE_CHECKING: import logging LEGACY_PLUGIN_NAMES: Dict[str, str] = { "Compiler": "PageCompiler", "Shortcode": "ShortcodePlugin", "Template": "TemplateSystem", } CATEGORY_NAMES: Set[str] = set(CATEGORIES.keys()) CATEGORY_TYPES: Set[Type[BasePlugin]] = set(CATEGORIES.values()) @dataclass(frozen=True) class PluginCandidate: """A candidate plugin that was located but not yet loaded (imported).""" name: str description: Optional[str] plugin_id: str category: str compiler: Optional[str] source_dir: Path module_name: str @dataclass(frozen=True) class PluginInfo: """A plugin that was loaded (imported).""" name: str description: Optional[str] plugin_id: str category: str compiler: Optional[str] source_dir: Path module_name: str module_object: object plugin_object: BasePlugin class PluginManager: """The Nikola plugin manager.""" categories_filter: Dict[str, Type[BasePlugin]] plugin_places: List[Path] logger: "logging.Logger" candidates: List[PluginCandidate] plugins: List[PluginInfo] _plugins_by_category: Dict[str, List[PluginInfo]] has_warnings: bool = False def __init__(self, plugin_places: List[Path]): """Initialize the plugin manager.""" self.plugin_places = plugin_places self.candidates = [] self.plugins = [] self._plugins_by_category = {} self.logger = get_logger("PluginManager") def locate_plugins(self) -> List[PluginCandidate]: """Locate plugins in plugin_places.""" self.candidates = [] plugin_files: List[Path] = [] for place in self.plugin_places: plugin_files += place.rglob("*.plugin") for plugin_file in plugin_files: source_dir = plugin_file.parent config = configparser.ConfigParser() config.read(plugin_file) name = config["Core"]["name"] module_name = config["Core"]["module"] plugin_id = f"Plugin {name} from {plugin_file}" description = None if "Documentation" in config: description = config["Documentation"].get("Description") if "Nikola" not in config: self.logger.warning(f"{plugin_id} does not specify Nikola configuration - plugin will not be loaded") self.logger.warning("Please add a [Nikola] section to the {plugin_file} file with a PluginCategory entry") self.has_warnings = True continue category = config["Nikola"].get("PluginCategory") compiler = config["Nikola"].get("Compiler") if not category: self.logger.warning(f"{plugin_id} does not specify any category (Nikola.PluginCategory is missing in .plugin file) - plugin will not be loaded") self.has_warnings = True continue if category in LEGACY_PLUGIN_NAMES: category = LEGACY_PLUGIN_NAMES[category] if category not in CATEGORY_NAMES: self.logger.warning(f"{plugin_id} specifies invalid category '{category}' in the .plugin file - plugin will not be loaded") self.has_warnings = True continue self.logger.debug(f"Discovered {plugin_id}") self.candidates.append( PluginCandidate( name=name, description=description, plugin_id=plugin_id, category=category, compiler=compiler, source_dir=source_dir, module_name=module_name, ) ) return self.candidates def load_plugins(self, candidates: List[PluginCandidate]) -> None: """Load selected candidate plugins.""" plugins_root = Path(__file__).parent.parent for candidate in candidates: name = candidate.name module_name = candidate.module_name source_dir = candidate.source_dir py_file_location = source_dir / f"{module_name}.py" plugin_id = candidate.plugin_id if not py_file_location.exists(): py_file_location = source_dir / module_name / "__init__.py" if not py_file_location.exists(): self.logger.warning(f"{plugin_id} could not be loaded (no valid module detected)") self.has_warnings = True continue plugin_id += f" ({py_file_location})" full_module_name = module_name try: name_parts = list(py_file_location.relative_to(plugins_root).parts) if name_parts[-1] == "__init__.py": name_parts.pop(-1) elif name_parts[-1].endswith(".py"): name_parts[-1] = name_parts[-1][:-3] full_module_name = ".".join(name_parts) except ValueError: pass try: spec = importlib.util.spec_from_file_location(full_module_name, py_file_location) module_object = importlib.util.module_from_spec(spec) if full_module_name not in sys.modules: sys.modules[full_module_name] = module_object spec.loader.exec_module(module_object) except Exception: self.logger.exception(f"{plugin_id} threw an exception while loading") self.has_warnings = True continue plugin_classes = [ c for c in vars(module_object).values() if isinstance(c, type) and issubclass(c, BasePlugin) and c not in CATEGORY_TYPES ] if len(plugin_classes) == 0: self.logger.warning(f"{plugin_id} does not have any plugin classes - plugin will not be loaded") self.has_warnings = True continue elif len(plugin_classes) > 1: self.logger.warning(f"{plugin_id} has multiple plugin classes; this is not supported - plugin will not be loaded") self.has_warnings = True continue plugin_class = plugin_classes[0] if not issubclass(plugin_class, CATEGORIES[candidate.category]): self.logger.warning(f"{plugin_id} has category '{candidate.category}' in the .plugin file, but the implementation class {plugin_class} does not inherit from this category - plugin will not be loaded") self.has_warnings = True continue try: plugin_object = plugin_class() except Exception: self.logger.exception(f"{plugin_id} threw an exception while creating the instance") self.has_warnings = True continue self.logger.debug(f"Loaded {plugin_id}") info = PluginInfo( name=name, description=candidate.description, plugin_id=candidate.plugin_id, category=candidate.category, compiler=candidate.compiler, source_dir=source_dir, module_name=module_name, module_object=module_object, plugin_object=plugin_object, ) self.plugins.append(info) self._plugins_by_category = {category: [] for category in CATEGORY_NAMES} for plugin_info in self.plugins: self._plugins_by_category[plugin_info.category].append(plugin_info) if self.has_warnings: self.logger.warning("Some plugins failed to load. Please review the above warning messages.") # TODO remove following messages and delay in v8.3.1 self.logger.warning("You may need to update some plugins (from plugins.getnikola.com) or to fix their .plugin files.") self.logger.warning("Waiting 2 seconds before continuing.") time.sleep(2) def get_plugins_of_category(self, category: str) -> List[PluginInfo]: """Get loaded plugins of a given category.""" return self._plugins_by_category.get(category, []) def get_plugin_by_name(self, name: str, category: Optional[str] = None) -> Optional[PluginInfo]: """Get a loaded plugin by name and optionally by category. Returns None if no such plugin is loaded.""" for p in self.plugins: if p.name == name and (category is None or p.category == category): return p # Aliases for Yapsy compatibility # TODO: remove in v9 def getPluginsOfCategory(self, category: str) -> List[PluginInfo]: """Get loaded plugins of a given category.""" self.logger.warning("Legacy getPluginsOfCategory method was used, it may be removed in the future. Please change it to get_plugins_of_category.") return self._plugins_by_category.get(category, []) # TODO: remove in v9 def getPluginByName(self, name: str, category: Optional[str] = None) -> Optional[PluginInfo]: """Get a loaded plugin by name and optionally by category. Returns None if no such plugin is loaded.""" self.logger.warning("Legacy getPluginByName method was used, it may be removed in the future. Please change it to get_plugin_by_name.") return self.get_plugin_by_name(name, category)