Plugin System

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(timestamp=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

from changedetectionio.processors.base import ProcessorBase

class MyCustomProcessor(ProcessorBase):
    """Custom processor for specialized monitoring"""
    
    def run(self, uuid, preferred_proxy=None, check_jitter=False):
        """
        Main processing method called for each check.
        
        Returns:
            - changed_detected (bool)
            - update_obj (dict): Data to update in watch
            - contents (str): Processed content for storage
        """
        # Fetch content
        fetcher = self.get_fetcher()
        contents = fetcher.run(uuid, preferred_proxy)
        
        # Your custom processing logic
        processed = self.process_content(contents)
        
        # Detect changes
        changed = self.detect_changes(processed)
        
        return changed, {}, processed
    
    def process_content(self, contents):
        """Implement your custom processing logic"""
        # Example: extract specific data
        return contents
    
    def detect_changes(self, contents):
        """Implement your change detection logic"""
        # Compare with previous check
        return True

components:
  schemas:
    processor_config_my_custom:
      type: object
      properties:
        custom_threshold:
          type: number
          description: Custom threshold value
          default: 10
        enable_feature:
          type: boolean
          description: Enable custom feature
          default: false

docker run -d \
  -e EXTRA_PACKAGES="changedetection.io-osint-processor my-custom-plugin" \
  -p 5000:5000 \
  dgtlmoon/changedetection.io

services:
  changedetection:
    image: dgtlmoon/changedetection.io
    environment:
      - EXTRA_PACKAGES=changedetection.io-osint-processor my-custom-plugin

EXTRA_PACKAGES="my-custom-plugin" changedetection.io

from setuptools import setup, find_packages

setup(
    name="changedetection-my-plugin",
    version="1.0.0",
    packages=find_packages(),
    install_requires=[
        "changedetection.io",
    ],
    entry_points={
        "changedetectionio": [
            "my_plugin = my_plugin.module:MyPlugin",
        ],
    },
)