Files
changedetection.io/changedetectionio/processors/pluggy_interface.py
T
2025-03-21 10:42:25 +01:00

85 lines
2.9 KiB
Python

import pluggy
from loguru import logger
# Ensure that the namespace in HookspecMarker matches PluginManager
PLUGIN_NAMESPACE = "changedetectionio_processors"
hookspec = pluggy.HookspecMarker(PLUGIN_NAMESPACE)
hookimpl = pluggy.HookimplMarker(PLUGIN_NAMESPACE)
UI_tags = {}
class ProcessorSpec:
"""Hook specifications for difference detection processors."""
@hookspec
def get_processor_name():
"""Return the processor name for selection in the UI."""
pass
@hookspec
def get_processor_description():
"""Return a human-readable description of the processor."""
pass
@hookspec
def get_processor_version():
"""Return the processor plugin version."""
pass
@hookspec
def get_processor_ui_tag():
"""Return the UI tag for the processor (used for categorization in UI)."""
pass
@hookspec
def perform_site_check(datastore, watch_uuid):
"""Return the processor handler class or None if not applicable.
Each plugin should check if it's the right processor for this watch
and return None if it's not.
Should return an instance of a class that implements:
- call_browser(preferred_proxy_id=None): Fetch the content
- run_changedetection(watch): Analyze for changes and return tuple of (changed_detected, update_obj, contents)
"""
pass
@hookspec
def get_form_class(processor_name):
"""Return the WTForms form class for the processor settings or None if not applicable.
Each plugin should check if it's the right processor and return None if not.
"""
pass
@hookspec
def get_watch_model_class(processor_name):
"""Return a custom Watch model class if needed or None if not applicable.
Each plugin should check if it's the right processor and return None if not.
"""
pass
# Set up Pluggy Plugin Manager
plugin_manager = pluggy.PluginManager(PLUGIN_NAMESPACE)
# Register hookspecs
plugin_manager.add_hookspecs(ProcessorSpec)
# Initialize by loading plugins and building UI_tags dictionary
try:
# Discover installed plugins from external packages (if any)
plugin_manager.load_setuptools_entrypoints(PLUGIN_NAMESPACE)
logger.info(f"Loaded plugins: {plugin_manager.get_plugins()}")
# Build UI_tags dictionary from all plugins
for plugin in plugin_manager.get_plugins():
if hasattr(plugin, "get_processor_name") and hasattr(plugin, "get_processor_ui_tag"):
plugin_name = plugin.get_processor_name()
ui_tag = plugin.get_processor_ui_tag()
if plugin_name and ui_tag:
UI_tags[plugin_name] = ui_tag
logger.info(f"Found UI tag for plugin {plugin_name}: {ui_tag}")
except Exception as e:
logger.critical(f"Error loading plugins: {str(e)}")