changedetection.io/changedetectionio/processors/__init__.py

from functools import lru_cache
from loguru import logger
from flask_babel import gettext
import importlib
import inspect
import os
import pkgutil

def find_sub_packages(package_name):
    """
    Find all sub-packages within the given package.

    :param package_name: The name of the base package to scan for sub-packages.
    :return: A list of sub-package names.
    """
    package = importlib.import_module(package_name)
    return [name for _, name, is_pkg in pkgutil.iter_modules(package.__path__) if is_pkg]


def find_processors():
    """
    Find all subclasses of DifferenceDetectionProcessor in the specified package.

    :param package_name: The name of the package to scan for processor modules.
    :return: A list of (module, class) tuples.
    """
    package_name = "changedetectionio.processors"  # Name of the current package/module

    processors = []
    sub_packages = find_sub_packages(package_name)
    from changedetectionio.processors.base import difference_detection_processor

    for sub_package in sub_packages:
        module_name = f"{package_name}.{sub_package}.processor"
        try:
            module = importlib.import_module(module_name)

            # Iterate through all classes in the module
            for name, obj in inspect.getmembers(module, inspect.isclass):
                # Only register classes that are actually defined in this module (not imported)
                if (issubclass(obj, difference_detection_processor) and
                    obj is not difference_detection_processor and
                    obj.__module__ == module.__name__):
                    processors.append((module, sub_package))
                    break  # Only need one processor per module
        except (ModuleNotFoundError, ImportError) as e:
            logger.warning(f"Failed to import module {module_name}: {e} (find_processors())")

    return processors


def get_parent_module(module):
    module_name = module.__name__
    if '.' not in module_name:
        return None  # Top-level module has no parent
    parent_module_name = module_name.rsplit('.', 1)[0]
    try:
        return importlib.import_module(parent_module_name)
    except Exception as e:
        pass

    return False



def get_custom_watch_obj_for_processor(processor_name):
    from changedetectionio.model import Watch
    watch_class = Watch.model
    processor_classes = find_processors()
    custom_watch_obj = next((tpl for tpl in processor_classes if tpl[1] == processor_name), None)
    if custom_watch_obj:
        # Parent of .processor.py COULD have its own Watch implementation
        parent_module = get_parent_module(custom_watch_obj[0])
        if hasattr(parent_module, 'Watch'):
            watch_class = parent_module.Watch

    return watch_class


def find_processor_module(processor_name):
    """
    Find the processor module by name.

    Args:
        processor_name: Processor machine name (e.g., 'image_ssim_diff')

    Returns:
        module: The processor's parent module, or None if not found
    """
    processor_classes = find_processors()
    processor_tuple = next((tpl for tpl in processor_classes if tpl[1] == processor_name), None)

    if processor_tuple:
        # Return the parent module (the package containing processor.py)
        return get_parent_module(processor_tuple[0])

    return None


def available_processors():
    """
    Get a list of processors by name and description for the UI elements.
    Can be filtered via ALLOWED_PROCESSORS environment variable (comma-separated list).
    :return: A list :)
    """

    processor_classes = find_processors()

    # Check if ALLOWED_PROCESSORS env var is set
    # For now we disable it, need to make a deploy with lots of new code and this will be an overload
    allowed_processors_env = os.getenv('ALLOWED_PROCESSORS', 'text_json_diff, restock_diff').strip()
    allowed_processors = None
    if allowed_processors_env:
        # Parse comma-separated list and strip whitespace
        allowed_processors = [p.strip() for p in allowed_processors_env.split(',') if p.strip()]
        logger.info(f"ALLOWED_PROCESSORS set, filtering to: {allowed_processors}")

    available = []
    for module, sub_package_name in processor_classes:
        # Filter by allowed processors if set
        if allowed_processors and sub_package_name not in allowed_processors:
            logger.debug(f"Skipping processor '{sub_package_name}' (not in ALLOWED_PROCESSORS)")
            continue

        # Try to get the 'name' attribute from the processor module first
        if hasattr(module, 'name'):
            description = gettext(module.name)
        else:
            # Fall back to processor_description from parent module's __init__.py
            parent_module = get_parent_module(module)
            if parent_module and hasattr(parent_module, 'processor_description'):
                description = gettext(parent_module.processor_description)
            else:
                # Final fallback to a readable name
                description = sub_package_name.replace('_', ' ').title()

        # Get weight for sorting (lower weight = higher in list)
        weight = 0  # Default weight for processors without explicit weight

        # Check processor module itself first
        if hasattr(module, 'processor_weight'):
            weight = module.processor_weight
        else:
            # Fall back to parent module (package __init__.py)
            parent_module = get_parent_module(module)
            if parent_module and hasattr(parent_module, 'processor_weight'):
                weight = parent_module.processor_weight

        available.append((sub_package_name, description, weight))

    # Sort by weight (lower weight = appears first)
    available.sort(key=lambda x: x[2])

    # Return as tuples without weight (for backwards compatibility)
    return [(name, desc) for name, desc, weight in available]


def get_processor_badge_texts():
    """
    Get a dictionary mapping processor names to their list_badge_text values.
    Translations are applied based on the current request locale.

    :return: A dict mapping processor name to badge text (e.g., {'text_json_diff': 'Text', 'restock_diff': 'Restock'})
    """
    processor_classes = find_processors()
    badge_texts = {}

    for module, sub_package_name in processor_classes:
        # Try to get the 'list_badge_text' attribute from the processor module
        if hasattr(module, 'list_badge_text'):
            badge_texts[sub_package_name] = gettext(module.list_badge_text)
        else:
            # Fall back to parent module's __init__.py
            parent_module = get_parent_module(module)
            if parent_module and hasattr(parent_module, 'list_badge_text'):
                badge_texts[sub_package_name] = gettext(parent_module.list_badge_text)

    return badge_texts


def get_processor_descriptions():
    """
    Get a dictionary mapping processor names to their description/name values.
    Translations are applied based on the current request locale.

    :return: A dict mapping processor name to description (e.g., {'text_json_diff': 'Webpage Text/HTML, JSON and PDF changes'})
    """
    processor_classes = find_processors()
    descriptions = {}

    for module, sub_package_name in processor_classes:
        # Try to get the 'name' or 'description' attribute from the processor module first
        if hasattr(module, 'name'):
            descriptions[sub_package_name] = gettext(module.name)
        elif hasattr(module, 'description'):
            descriptions[sub_package_name] = gettext(module.description)
        else:
            # Fall back to parent module's __init__.py
            parent_module = get_parent_module(module)
            if parent_module and hasattr(parent_module, 'processor_description'):
                descriptions[sub_package_name] = gettext(parent_module.processor_description)
            elif parent_module and hasattr(parent_module, 'name'):
                descriptions[sub_package_name] = gettext(parent_module.name)
            else:
                # Final fallback to a readable name
                descriptions[sub_package_name] = sub_package_name.replace('_', ' ').title()

    return descriptions


def generate_processor_badge_colors(processor_name):
    """
    Generate consistent colors for a processor badge based on its name.
    Uses a hash of the processor name to generate pleasing, accessible colors
    for both light and dark modes.

    :param processor_name: The processor name (e.g., 'text_json_diff')
    :return: A dict with 'light' and 'dark' color schemes, each containing 'bg' and 'color'
    """
    import hashlib

    # Generate a consistent hash from the processor name
    hash_obj = hashlib.md5(processor_name.encode('utf-8'))
    hash_int = int(hash_obj.hexdigest()[:8], 16)

    # Generate hue from hash (0-360)
    hue = hash_int % 360

    # Light mode: pastel background with darker text
    light_saturation = 60 + (hash_int % 25)  # 60-85%
    light_lightness = 85 + (hash_int % 10)   # 85-95% - very light
    text_lightness = 25 + (hash_int % 15)    # 25-40% - dark

    # Dark mode: solid, vibrant colors with white text
    dark_saturation = 55 + (hash_int % 20)   # 55-75%
    dark_lightness = 45 + (hash_int % 15)    # 45-60%

    return {
        'light': {
            'bg': f'hsl({hue}, {light_saturation}%, {light_lightness}%)',
            'color': f'hsl({hue}, 50%, {text_lightness}%)'
        },
        'dark': {
            'bg': f'hsl({hue}, {dark_saturation}%, {dark_lightness}%)',
            'color': '#fff'
        }
    }


@lru_cache(maxsize=1)
def get_processor_badge_css():
    """
    Generate CSS for all processor badges with auto-generated colors.
    This creates CSS rules for both light and dark modes for each processor.

    :return: A string containing CSS rules for all processor badges
    """
    processor_classes = find_processors()
    css_rules = []

    for module, sub_package_name in processor_classes:
        colors = generate_processor_badge_colors(sub_package_name)

        # Light mode rule
        css_rules.append(
            f".processor-badge-{sub_package_name} {{\n"
            f"  background-color: {colors['light']['bg']};\n"
            f"  color: {colors['light']['color']};\n"
            f"}}"
        )

        # Dark mode rule
        css_rules.append(
            f"html[data-darkmode=\"true\"] .processor-badge-{sub_package_name} {{\n"
            f"  background-color: {colors['dark']['bg']};\n"
            f"  color: {colors['dark']['color']};\n"
            f"}}"
        )

    return '\n\n'.join(css_rules)