From 35455e7dd6db60c94579de54bb7b92d1032df477 Mon Sep 17 00:00:00 2001 From: dgtlmoon Date: Mon, 17 Mar 2025 18:51:01 +0100 Subject: [PATCH] plugins refactor --- .../processors/processor_registry.py | 245 ++++++++++-------- changedetectionio/processors/whois_plugin.py | 28 +- 2 files changed, 134 insertions(+), 139 deletions(-) diff --git a/changedetectionio/processors/processor_registry.py b/changedetectionio/processors/processor_registry.py index 1a7bc6136..5b123e6ee 100644 --- a/changedetectionio/processors/processor_registry.py +++ b/changedetectionio/processors/processor_registry.py @@ -1,6 +1,8 @@ from loguru import logger from changedetectionio.model import Watch from .pluggy_interface import plugin_manager +from typing import Dict, Any, List, Tuple, Optional, TypeVar, Type +import functools # Register the WHOIS plugin from . import whois_plugin @@ -9,158 +11,167 @@ plugin_manager.register(whois_plugin) # Load any setuptools entrypoints plugin_manager.load_setuptools_entrypoints("changedetectionio_processors") +# Type definitions for better type hinting +T = TypeVar('T') +ProcessorClass = TypeVar('ProcessorClass') +ProcessorForm = TypeVar('ProcessorForm') +ProcessorWatchModel = TypeVar('ProcessorWatchModel') +ProcessorInstance = TypeVar('ProcessorInstance') + +# Cache for plugin name mapping to improve performance +# This will be populated after the first call to _get_plugin_name_map +_plugin_name_map: Dict[str, Any] = {} + def register_plugin(plugin_module): """Register a processor plugin""" plugin_manager.register(plugin_module) + # Clear the plugin name map cache when a new plugin is registered + global _plugin_name_map + _plugin_name_map = {} -def get_all_processors(): +def _get_plugin_name_map() -> Dict[str, Any]: + """Get a mapping of processor names to plugins + :return: Dictionary mapping processor names to plugins + """ + global _plugin_name_map + + # Return cached map if available + if _plugin_name_map: + return _plugin_name_map + + # Build the map + result = {} + for plugin in plugin_manager.get_plugins(): + try: + # Get the processor name from this plugin + name_results = plugin_manager.hook.get_processor_name(plugin=plugin) + if name_results: + # Each plugin should return exactly one name + result[name_results[0]] = plugin + except Exception as e: + logger.error(f"Error getting processor name from plugin: {str(e)}") + + # Cache the map + _plugin_name_map = result + return result + +def _get_plugin_by_name(processor_name: str) -> Optional[Any]: + """Get a plugin by its processor name + :param processor_name: Name of the processor + :return: Plugin object or None + """ + return _get_plugin_name_map().get(processor_name) + +def _call_hook_for_plugin(plugin: Any, hook_name: str, default_value: T = None, **kwargs) -> Optional[T]: + """Call a hook for a specific plugin and handle exceptions + :param plugin: The plugin to call the hook for + :param hook_name: Name of the hook to call + :param default_value: Default value to return if the hook call fails + :param kwargs: Additional arguments to pass to the hook + :return: Result of the hook call or default value + """ + if not plugin: + return default_value + + try: + hook = getattr(plugin_manager.hook, hook_name) + results = hook(plugin=plugin, **kwargs) + if results: + return results[0] + except Exception as e: + logger.error(f"Error calling {hook_name} for plugin: {str(e)}") + + return default_value + +def get_all_processors() -> List[Tuple[str, str]]: """Get all processors :return: List of tuples (processor_name, processor_description) """ processors = [] - for plugin in plugin_manager.get_plugins(): + for processor_name, plugin in _get_plugin_name_map().items(): try: - # Get the processor name from this plugin - name = plugin_manager.hook.get_processor_name(plugin=plugin) - # Get the processor description from this plugin - description = plugin_manager.hook.get_processor_description(plugin=plugin) + description = _call_hook_for_plugin(plugin, 'get_processor_description') # Check if both name and description were returned - if name and description: - # Each plugin should return exactly one name and one description - processors.append((name[0], description[0])) + if description: + processors.append((processor_name, description)) except Exception as e: - logger.error(f"Error getting processor info from plugin: {str(e)}") + logger.error(f"Error getting processor info for {processor_name}: {str(e)}") return processors -def get_processor_class(processor_name): +def get_processor_class(processor_name: str) -> Optional[Type[ProcessorClass]]: """Get processor class by name :param processor_name: Name of the processor :return: Processor class or None """ - for plugin in plugin_manager.get_plugins(): - try: - # Get the processor name from this plugin - name = plugin_manager.hook.get_processor_name(plugin=plugin) - - # If this plugin's processor name matches what we're looking for - if name and name[0] == processor_name: - # Get the processor class - processor_class = plugin_manager.hook.get_processor_class(plugin=plugin) - if processor_class: - return processor_class[0] - except Exception as e: - logger.error(f"Error getting processor class from plugin: {str(e)}") - - return None + plugin = _get_plugin_by_name(processor_name) + return _call_hook_for_plugin(plugin, 'get_processor_class') -def get_processor_form(processor_name): +def get_processor_form(processor_name: str) -> Optional[Type[ProcessorForm]]: """Get processor form by name :param processor_name: Name of the processor :return: Processor form class or None """ - for plugin in plugin_manager.get_plugins(): - try: - # Get the processor name from this plugin - name = plugin_manager.hook.get_processor_name(plugin=plugin) - - # If this plugin's processor name matches what we're looking for - if name and name[0] == processor_name: - # Get the processor form - processor_form = plugin_manager.hook.get_processor_form(plugin=plugin) - if processor_form: - return processor_form[0] - except Exception as e: - logger.error(f"Error getting processor form from plugin: {str(e)}") - - return None + plugin = _get_plugin_by_name(processor_name) + return _call_hook_for_plugin(plugin, 'get_processor_form') -def get_processor_watch_model(processor_name): +def get_processor_watch_model(processor_name: str) -> Type[ProcessorWatchModel]: """Get processor watch model by name :param processor_name: Name of the processor :return: Watch model class or default Watch model """ - for plugin in plugin_manager.get_plugins(): - try: - # Get the processor name from this plugin - name = plugin_manager.hook.get_processor_name(plugin=plugin) - - # If this plugin's processor name matches what we're looking for - if name and name[0] == processor_name: - # Get the processor watch model - processor_watch_model = plugin_manager.hook.get_processor_watch_model(plugin=plugin) - if processor_watch_model and processor_watch_model[0]: - return processor_watch_model[0] - except Exception as e: - logger.error(f"Error getting processor watch model from plugin: {str(e)}") - - return Watch.model + plugin = _get_plugin_by_name(processor_name) + return _call_hook_for_plugin(plugin, 'get_processor_watch_model', default_value=Watch.model) - -def get_processor_site_check(processor_name, datastore, watch_uuid): +def get_processor_site_check(processor_name: str, datastore: Any, watch_uuid: str) -> Optional[ProcessorInstance]: """Get a processor instance ready to perform site check :param processor_name: Name of the processor :param datastore: The application datastore :param watch_uuid: The UUID of the watch to check :return: A processor instance ready to perform site check, or None """ - for plugin in plugin_manager.get_plugins(): - try: - # Get the processor name from this plugin - name = plugin_manager.hook.get_processor_name(plugin=plugin) - - # If this plugin's processor name matches what we're looking for - if name and name[0] == processor_name: - # Try to get the perform_site_check implementation - perform_site_check_impls = plugin_manager.hook.perform_site_check( - plugin=plugin, - datastore=datastore, - watch_uuid=watch_uuid - ) - if perform_site_check_impls: - return perform_site_check_impls[0] - - # If no perform_site_check hook implementation, try getting the class and instantiating it - processor_class = plugin_manager.hook.get_processor_class(plugin=plugin) - if processor_class: - return processor_class[0](datastore=datastore, watch_uuid=watch_uuid) - except Exception as e: - logger.error(f"Error getting processor site check from plugin: {str(e)}") + plugin = _get_plugin_by_name(processor_name) + if not plugin: + return None + + # Try to get the perform_site_check implementation + try: + processor = _call_hook_for_plugin( + plugin, + 'perform_site_check', + datastore=datastore, + watch_uuid=watch_uuid + ) + if processor: + return processor + + # If no perform_site_check hook implementation, try getting the class and instantiating it + processor_class = _call_hook_for_plugin(plugin, 'get_processor_class') + if processor_class: + return processor_class(datastore=datastore, watch_uuid=watch_uuid) + except Exception as e: + logger.error(f"Error getting processor site check for {processor_name}: {str(e)}") return None - -def get_display_link(url, processor_name): +def get_display_link(url: str, processor_name: str) -> Optional[str]: """Get a custom display link for the given processor :param url: The original URL from the watch :param processor_name: Name of the processor :return: A string with the custom display link or None to use the default """ - for plugin in plugin_manager.get_plugins(): - try: - # Get the processor name from this plugin - name = plugin_manager.hook.get_processor_name(plugin=plugin) - - # If this plugin's processor name matches what we're looking for - if name and name[0] == processor_name: - # Try to get the get_display_link implementation - display_links = plugin_manager.hook.get_display_link( - plugin=plugin, - url=url, - processor_name=processor_name - ) - if display_links and display_links[0]: - return display_links[0] - except Exception as e: - logger.error(f"Error getting display link from plugin: {str(e)}") - - return None + plugin = _get_plugin_by_name(processor_name) + return _call_hook_for_plugin( + plugin, + 'get_display_link', + url=url, + processor_name=processor_name + ) - -def get_plugin_processor_modules(): +def get_plugin_processor_modules() -> List[Tuple[Any, str]]: """Get processor modules for all plugins that can be used with the find_processors function This function adapts pluggy plugins to be compatible with the traditional find_processors system @@ -169,19 +180,25 @@ def get_plugin_processor_modules(): """ result = [] + # Import base modules once to avoid repeated imports + from changedetectionio.processors.text_json_diff import processor as text_json_diff_processor + # For each plugin, create a fake module that can be used with find_processors - for plugin in plugin_manager.get_plugins(): + for processor_name, plugin in _get_plugin_name_map().items(): try: - # Get the processor name from this plugin - name = plugin_manager.hook.get_processor_name(plugin=plugin) + # Get the processor class for this plugin + processor_class = _call_hook_for_plugin(plugin, 'get_processor_class') - if name: - processor_name = name[0] - # For the WHOIS processor specifically, use text_json_diff as the base module - if processor_name == 'whois': - from changedetectionio.processors.text_json_diff import processor as text_json_diff_processor + if processor_class: + # Check if this processor inherits from TextJsonDiffProcessor + from changedetectionio.processors.text_json_diff.processor import TextJsonDiffProcessor + if issubclass(processor_class, TextJsonDiffProcessor) or 'TextJsonDiffProcessor' in str(processor_class.__bases__): result.append((text_json_diff_processor, processor_name)) + else: + # For non-inherited processors, could create a mapping to their base module + # Future enhancement: dynamically determine base module based on inheritance + logger.debug(f"Processor {processor_name} does not inherit from TextJsonDiffProcessor") except Exception as e: - logger.error(f"Error getting processor module from plugin: {str(e)}") + logger.error(f"Error determining processor module for {processor_name}: {str(e)}") return result \ No newline at end of file diff --git a/changedetectionio/processors/whois_plugin.py b/changedetectionio/processors/whois_plugin.py index 51144d6a3..bc86814ca 100644 --- a/changedetectionio/processors/whois_plugin.py +++ b/changedetectionio/processors/whois_plugin.py @@ -44,7 +44,7 @@ class WhoisProcessor(TextJsonDiffProcessor): # Ensure we have a valid domain if not domain: - error_msg = f"Could not extract domain from URL: {url}" + error_msg = f"Could not extract domain from URL: '{url}'" self.fetcher.content = error_msg self.fetcher.status_code = 400 logger.error(error_msg) @@ -107,6 +107,8 @@ class WhoisProcessor(TextJsonDiffProcessor): self.fetcher.get_last_status_code = lambda: self.fetcher.status_code self.fetcher.quit = lambda: None logger.error(error_msg) + + return def run_changedetection(self, watch): """Use the parent's run_changedetection which will use our overridden call_browser method""" @@ -127,30 +129,6 @@ class WhoisProcessor(TextJsonDiffProcessor): processor = WhoisProcessor(datastore=datastore, watch_uuid=watch_uuid) return processor -@hookimpl -def get_display_link(url, processor_name): - """Return a custom display link for WHOIS processor - - Extract the domain from the URL and return a formatted link that shows - this is a WHOIS lookup rather than a regular web page. - """ - if processor_name == 'whois': - try: - # Extract domain from URL - parsed_url = urllib.parse.urlparse(url) - domain = parsed_url.netloc - - # Remove www. prefix if present - domain = re.sub(r'^www\.', '', domain) - - if domain: - return f"WHOIS - {domain}" - except Exception as e: - logger.error(f"Error generating WHOIS display link: {str(e)}") - return url - - return None - @hookimpl def perform_site_check(datastore, watch_uuid): """Create and return a processor instance ready to perform site check"""