Conditions levenshtein

changedetectionio/PLUGIN_README.md

@@ -0,0 +1,98 @@
+# Creating Plugins for changedetection.io
+
+This document describes how to create plugins for changedetection.io. Plugins can be used to extend the functionality of the application in various ways.
+
+## Plugin Types
+
+### UI Stats Tab Plugins
+
+These plugins can add content to the Stats tab in the Edit page. This is useful for adding custom statistics or visualizations about a watch.
+
+#### Creating a UI Stats Tab Plugin
+
+1. Create a Python file in a directory that will be loaded by the plugin system.
+
+2. Use the `global_hookimpl` decorator to implement the `ui_edit_stats_extras` hook:
+
+```python
+import pluggy
+from loguru import logger
+
+global_hookimpl = pluggy.HookimplMarker("changedetectionio")
+
+@global_hookimpl
+def ui_edit_stats_extras(watch):
+    """Add custom content to the stats tab"""
+    # Calculate or retrieve your stats
+    my_stat = calculate_something(watch)
+
+    # Return HTML content as a string
+    html = f"""
+    <div class="my-plugin-stats">
+        <h4>My Plugin Statistics</h4>
+        <p>My statistic: {my_stat}</p>
+    </div>
+    """
+    return html
+```
+
+3. The HTML you return will be included in the Stats tab.
+
+## Plugin Loading
+
+Plugins can be loaded from:
+
+1. Built-in plugin directories in the codebase
+2. External packages using setuptools entry points
+
+To add a new plugin directory, modify the `plugin_dirs` dictionary in `pluggy_interface.py`.
+
+## Example Plugin
+
+Here's a simple example of a plugin that adds a word count statistic to the Stats tab:
+
+```python
+import pluggy
+from loguru import logger
+
+global_hookimpl = pluggy.HookimplMarker("changedetectionio")
+
+def count_words_in_history(watch):
+    """Count words in the latest snapshot"""
+    try:
+        if not watch.history.keys():
+            return 0
+
+        latest_key = list(watch.history.keys())[-1]
+        latest_content = watch.get_history_snapshot(latest_key)
+        return len(latest_content.split())
+    except Exception as e:
+        logger.error(f"Error counting words: {str(e)}")
+        return 0
+
+@global_hookimpl
+def ui_edit_stats_extras(watch):
+    """Add word count to the Stats tab"""
+    word_count = count_words_in_history(watch)
+
+    html = f"""
+    <div class="word-count-stats">
+        <h4>Content Analysis</h4>
+        <table class="pure-table">
+            <tbody>
+                <tr>
+                    <td>Word count (latest snapshot)</td>
+                    <td>{word_count}</td>
+                </tr>
+            </tbody>
+        </table>
+    </div>
+    """
+    return html
+```
+
+## Testing Your Plugin
+
+1. Place your plugin in one of the directories scanned by the plugin system
+2. Restart changedetection.io
+3. Go to the Edit page of a watch and check the Stats tab to see your content

changedetectionio/blueprint/ui/edit.py

@@ -233,6 +233,9 @@ def edit_page(uuid):
 
             # Only works reliably with Playwright
 
+            # Import the global plugin system
+            from changedetectionio.pluggy_interface import collect_ui_edit_stats_extras
+
             template_args = {
                 'available_processors': processors.available_processors(),
                 'available_timezones': sorted(available_timezones()),
@@ -250,6 +253,7 @@ def edit_page(uuid):
                 'settings_application': datastore.data['settings']['application'],
                 'system_has_playwright_configured': os.getenv('PLAYWRIGHT_DRIVER_URL'),
                 'system_has_webdriver_configured': os.getenv('WEBDRIVER_URL'),
+                'ui_edit_stats_extras': collect_ui_edit_stats_extras(watch),
                 'visual_selector_data_ready': datastore.visualselector_data_is_ready(watch_uuid=uuid),
                 'timezone_default_config': datastore.data['settings']['application'].get('timezone'),
                 'using_global_webdriver_wait': not default['webdriver_delay'],

changedetectionio/conditions/__init__.py

@@ -102,12 +102,31 @@ def execute_ruleset_against_all_plugins(current_watch_uuid: str, application_dat
         if complete_rules:
             # Give all plugins a chance to update the data dict again (that we will test the conditions against)
             for plugin in plugin_manager.get_plugins():
-                new_execute_data = plugin.add_data(current_watch_uuid=current_watch_uuid,
-                                                   application_datastruct=application_datastruct,
-                                                   ephemeral_data=ephemeral_data)
-
-                if new_execute_data and isinstance(new_execute_data, dict):
-                    EXECUTE_DATA.update(new_execute_data)
+                try:
+                    import concurrent.futures
+                    import time
+
+                    with concurrent.futures.ThreadPoolExecutor() as executor:
+                        future = executor.submit(
+                            plugin.add_data,
+                            current_watch_uuid=current_watch_uuid,
+                            application_datastruct=application_datastruct,
+                            ephemeral_data=ephemeral_data
+                        )
+
+                        # Set a timeout of 10 seconds
+                        try:
+                            new_execute_data = future.result(timeout=10)
+                            if new_execute_data and isinstance(new_execute_data, dict):
+                                EXECUTE_DATA.update(new_execute_data)
+                        except concurrent.futures.TimeoutError:
+                            # The plugin took too long, abort processing for this watch
+                            raise Exception(f"Plugin {plugin.__class__.__name__} took more than 10 seconds to run.")
+                except Exception as e:
+                    # Log the error but continue with the next plugin
+                    import logging
+                    logging.error(f"Error executing plugin {plugin.__class__.__name__}: {str(e)}")
+                    continue
 
             # Create the ruleset
             ruleset = convert_to_jsonlogic(logic_operator=logic_operator, rule_dict=complete_rules)
@@ -132,3 +151,18 @@ def execute_ruleset_against_all_plugins(current_watch_uuid: str, application_dat
     if isinstance(new_field_choices, list):
         field_choices.extend(new_field_choices)
 
+def collect_ui_edit_stats_extras(watch):
+    """Collect and combine HTML content from all plugins that implement ui_edit_stats_extras"""
+    extras_content = []
+
+    for plugin in plugin_manager.get_plugins():
+        try:
+            content = plugin.ui_edit_stats_extras(watch=watch)
+            if content:
+                extras_content.append(content)
+        except Exception as e:
+            # Skip plugins that don't implement the hook or have errors
+            pass
+
+    return "\n".join(extras_content) if extras_content else ""
+

changedetectionio/conditions/pluggy_interface.py

@@ -1,5 +1,8 @@
 import pluggy
-from . import default_plugin  # Import the default plugin
+import os
+import importlib
+import sys
+from . import default_plugin
 
 # ✅ Ensure that the namespace in HookspecMarker matches PluginManager
 PLUGIN_NAMESPACE = "changedetectionio_conditions"
@@ -30,6 +33,11 @@ def register_field_choices():
     def add_data(current_watch_uuid, application_datastruct, ephemeral_data):
         """Add to the datadict"""
         pass
+
+    @hookspec
+    def ui_edit_stats_extras(watch):
+        """Return HTML content to add to the stats tab in the edit view"""
+        pass
 
 # ✅ Set up Pluggy Plugin Manager
 plugin_manager = pluggy.PluginManager(PLUGIN_NAMESPACE)
@@ -40,5 +48,27 @@ def add_data(current_watch_uuid, application_datastruct, ephemeral_data):
 # ✅ Register built-in plugins manually
 plugin_manager.register(default_plugin, "default_plugin")
 
+# ✅ Load plugins from the plugins directory
+def load_plugins_from_directory():
+    plugins_dir = os.path.join(os.path.dirname(__file__), 'plugins')
+    if not os.path.exists(plugins_dir):
+        return
+
+    # Get all Python files (excluding __init__.py)
+    for filename in os.listdir(plugins_dir):
+        if filename.endswith(".py") and filename != "__init__.py":
+            module_name = filename[:-3]  # Remove .py extension
+            module_path = f"changedetectionio.conditions.plugins.{module_name}"
+
+            try:
+                module = importlib.import_module(module_path)
+                # Register the plugin with pluggy
+                plugin_manager.register(module, module_name)
+            except (ImportError, AttributeError) as e:
+                print(f"Error loading plugin {module_name}: {e}")
+
+# Load plugins from the plugins directory
+load_plugins_from_directory()
+
 # ✅ Discover installed plugins from external packages (if any)
 plugin_manager.load_setuptools_entrypoints(PLUGIN_NAMESPACE)

changedetectionio/conditions/plugins/__init__.py

@@ -0,0 +1 @@
+# Import plugins package to make them discoverable

changedetectionio/conditions/plugins/levenshtein_plugin.py

@@ -0,0 +1,102 @@
+import pluggy
+from loguru import logger
+
+# Support both plugin systems
+conditions_hookimpl = pluggy.HookimplMarker("changedetectionio_conditions")
+global_hookimpl = pluggy.HookimplMarker("changedetectionio")
+
+def levenshtein_ratio_recent_history(watch, incoming_text=None):
+    try:
+        from Levenshtein import ratio, distance
+        k = list(watch.history.keys())
+        if len(k) >= 2:
+            # When called from ui_edit_stats_extras, we don't have incoming_text
+            if incoming_text is None:
+                a = watch.get_history_snapshot(timestamp=k[-1])  # Latest snapshot
+                b = watch.get_history_snapshot(timestamp=k[-2])  # Previous snapshot
+            else:
+                a = watch.get_history_snapshot(timestamp=k[-2]) # Second newest, incoming_text will be "newest"
+                b = incoming_text
+
+            distance_value = distance(a, b)
+            ratio_value = ratio(a, b)
+            return {
+                'distance': distance_value,
+                'ratio': ratio_value,
+                'percent_similar': round(ratio_value * 100, 2)
+            }
+    except Exception as e:
+        logger.warning(f"Unable to calc similarity: {str(e)}")
+
+    return ''
+
+@conditions_hookimpl
+def register_operators():
+    pass
+
+@conditions_hookimpl
+def register_operator_choices():
+    pass
+
+
+@conditions_hookimpl
+def register_field_choices():
+    return [
+        ("levenshtein_ratio", "Levenshtein - Text similarity ratio"),
+        ("levenshtein_distance", "Levenshtein - Text change distance"),
+    ]
+
+@conditions_hookimpl
+def add_data(current_watch_uuid, application_datastruct, ephemeral_data):
+    res = {}
+    watch = application_datastruct['watching'].get(current_watch_uuid)
+    # ephemeral_data['text'] will be the current text after filters, they may have edited filters but not saved them yet etc
+
+    if watch and 'text' in ephemeral_data:
+        lev_data = levenshtein_ratio_recent_history(watch, ephemeral_data['text'])
+        if isinstance(lev_data, dict):
+            res['levenshtein_ratio'] = lev_data.get('ratio', 0)
+            res['levenshtein_similarity'] = lev_data.get('percent_similar', 0)
+            res['levenshtein_distance'] = lev_data.get('distance', 0)
+
+    return res
+
+@global_hookimpl
+def ui_edit_stats_extras(watch):
+    """Add Levenshtein stats to the UI using the global plugin system"""
+    """Generate the HTML for Levenshtein stats - shared by both plugin systems"""
+    if len(watch.history.keys()) < 2:
+        return "<p>Not enough history to calculate Levenshtein metrics</p>"
+
+    try:
+        lev_data = levenshtein_ratio_recent_history(watch)
+        if not lev_data or not isinstance(lev_data, dict):
+            return "<p>Unable to calculate Levenshtein metrics</p>"
+
+        html = f"""
+        <div class="levenshtein-stats">
+            <h4>Levenshtein Text Similarity Details</h4>
+            <table class="pure-table">
+                <tbody>
+                    <tr>
+                        <td>Raw distance (edits needed)</td>
+                        <td>{lev_data['distance']}</td>
+                    </tr>
+                    <tr>
+                        <td>Similarity ratio</td>
+                        <td>{lev_data['ratio']:.4f}</td>
+                    </tr>
+                    <tr>
+                        <td>Percent similar</td>
+                        <td>{lev_data['percent_similar']}%</td>
+                    </tr>
+                </tbody>
+            </table>
+            <p style="font-size: 80%;">Levenshtein metrics compare the last two snapshots, measuring how many character edits are needed to transform one into the other.</p>
+        </div>
+        """
+        return html
+    except Exception as e:
+        logger.error(f"Error generating Levenshtein UI extras: {str(e)}")
+        return "<p>Error calculating Levenshtein metrics</p>"
+