changedetection.io/changedetectionio/blueprint/settings/__init__.py

import os
from copy import deepcopy
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo, available_timezones
import secrets
import time
import flask_login
from flask import Blueprint, render_template, request, redirect, url_for, flash
from flask_babel import gettext

from changedetectionio.store import ChangeDetectionStore
from changedetectionio.auth_decorator import login_optionally_required
from changedetectionio.model.LLMSettings import LLMSettings


def construct_blueprint(datastore: ChangeDetectionStore):
    from changedetectionio.llm.evaluator import is_llm_features_disabled
    settings_blueprint = Blueprint('settings', __name__, template_folder="templates")
    if not is_llm_features_disabled():
        from changedetectionio.blueprint.settings.llm import construct_llm_blueprint
        settings_blueprint.register_blueprint(construct_llm_blueprint(datastore), url_prefix='/llm')

    # Notification settings live in their own child blueprint so future backends
    # (simple_email, webhooks, etc.) slot in as /settings/notifications/<backend>
    # without further URL-shaping churn.
    from changedetectionio.blueprint.settings.notifications import construct_notifications_blueprint
    settings_blueprint.register_blueprint(
        construct_notifications_blueprint(datastore), url_prefix='/notifications',
    )

    @settings_blueprint.route("", methods=['GET', "POST"])
    @login_optionally_required
    def settings_page():
        from changedetectionio import forms
        from changedetectionio.pluggy_interface import (
            get_plugin_settings_tabs,
            load_plugin_settings,
            save_plugin_settings
        )


        default = deepcopy(datastore.data['settings'])

        # api_key is intentionally blanked on GET — PasswordField never re-renders
        # its value, and a blank submission preserves the stored key.
        default['llm'] = LLMSettings.model_validate(
            datastore.data['settings']['application'].get('llm') or {}
        ).model_dump()
        default['llm']['api_key'] = ''

        if datastore.proxy_list is not None:
            available_proxies = list(datastore.proxy_list.keys())
            # When enabled
            system_proxy = datastore.data['settings']['requests']['proxy']
            # In the case it doesnt exist anymore
            if not system_proxy in available_proxies:
                system_proxy = None

            default['requests']['proxy'] = system_proxy if system_proxy is not None else available_proxies[0]
            # Used by the form handler to keep or remove the proxy settings
            default['proxy_list'] = available_proxies[0]

        # Don't use form.data on POST so that it doesnt overrid the checkbox status from the POST status
        form = forms.globalSettingsForm(formdata=request.form if request.method == 'POST' else None,
                                        data=default,
                                        extra_notification_tokens=datastore.get_unique_notification_tokens_available()
                                        )

        # Remove the last option 'System default'
        form.application.form.notification_format.choices.pop()

        if datastore.proxy_list is None:
            # @todo - Couldn't get setattr() etc dynamic addition working, so remove it instead
            del form.requests.form.proxy
        else:
            form.requests.form.proxy.choices = []
            for p in datastore.proxy_list:
                form.requests.form.proxy.choices.append(tuple((p, datastore.proxy_list[p]['label'])))

        if request.method == 'POST':
            # Password unset is a GET, but we can lock the session to a salted env password to always need the password
            if form.application.form.data.get('removepassword_button', False):
                # SALTED_PASS means the password is "locked" to what we set in the Env var
                if not os.getenv("SALTED_PASS", False):
                    datastore.remove_password()
                    flash(gettext("Password protection removed."), 'notice')
                    flask_login.logout_user()
                    return redirect(url_for('settings.settings_page'))

            if form.validate():
                # Don't set password to False when a password is set - should be only removed with the `removepassword` button
                app_update = dict(deepcopy(form.data['application']))

                # Never update password with '' or False (Added by wtforms when not in submission)
                if 'password' in app_update and not app_update['password']:
                    del (app_update['password'])

                # Notification fields live on the dedicated /settings/notifications page now.
                # The application sub-form still defines them (inherited from commonSettingsForm),
                # so an unrelated save here would clobber the stored values with empty WTForms
                # defaults. Drop them from the merge to leave the notifications config alone.
                for nf in ('notification_urls', 'notification_title', 'notification_body',
                           'notification_format', 'base_url'):
                    app_update.pop(nf, None)

                datastore.data['settings']['application'].update(app_update)

                # LLM config lives under settings.application.llm.* (post update_31).
                # Hydrate the stored dict into LLMSettings, then merge form input over it.
                # WTForms field names match LLMSettings field names exactly, so both sides
                # of the merge use the same key shape.
                existing_llm = LLMSettings.model_validate(
                    datastore.data['settings']['application'].get('llm') or {}
                )

                llm_form_input = dict(form.data.get('llm') or {})

                # Empty IntegerField submissions come back as None from WTForms;
                # the schema declares those fields as strict `int`, so passing
                # them through would fail validation. Treat None like the
                # absent-key case: keep the stored value, don't merge.
                llm_form_input = {k: v for k, v in llm_form_input.items() if v is not None}

                # PasswordField never re-renders, so a blank submitted value means
                # "keep stored key" — drop it from the merge.
                if not (llm_form_input.get('api_key') or '').strip():
                    llm_form_input.pop('api_key', None)

                # Env-var overrides make these fields read-only in the UI — ignore form input.
                if os.getenv('LLM_TOKEN_BUDGET_MONTH', '').strip():
                    llm_form_input.pop('token_budget_month', None)
                if os.getenv('LLM_MAX_INPUT_CHARS', '').strip():
                    llm_form_input.pop('max_input_chars', None)

                # System-managed counters must never come from the form.
                for protected in LLMSettings.PROTECTED_FIELDS:
                    llm_form_input.pop(protected, None)

                merged = LLMSettings.model_validate({**existing_llm.model_dump(), **llm_form_input})

                # Clearing the model field strips only the provider-connection fields.
                # User toggles, budgets, prompts and system counters survive (matches /llm/clear).
                exclude = set(LLMSettings.CONNECTION_FIELDS) if not merged.model.strip() else None
                datastore.data['settings']['application']['llm'] = merged.model_dump(exclude=exclude)

                # Handle dynamic worker count adjustment
                old_worker_count = datastore.data['settings']['requests'].get('workers', 1)
                new_worker_count = form.data['requests'].get('workers', 1)

                datastore.data['settings']['requests'].update(form.data['requests'])
                datastore.commit()

                # Clear all checksums to force reprocessing with new settings
                # Global settings can affect watch behavior (filters, rendering, etc.)
                datastore.clear_all_last_checksums()

                # Adjust worker count if it changed
                if new_worker_count != old_worker_count:
                    from changedetectionio import worker_pool
                    from changedetectionio.flask_app import update_q, notification_q, app, datastore as ds

                    # Check CPU core availability and warn if worker count is high
                    cpu_count = os.cpu_count()
                    if cpu_count and new_worker_count >= (cpu_count * 0.9):
                        flash(gettext("Warning: Worker count ({worker_count}) is close to or exceeds available CPU cores ({cpu_count})").format(
                            worker_count=new_worker_count, cpu_count=cpu_count), 'warning')

                    result = worker_pool.adjust_async_worker_count(
                        new_count=new_worker_count,
                        update_q=update_q,
                        notification_q=notification_q,
                        app=app,
                        datastore=ds
                    )

                    if result['status'] == 'success':
                        flash(gettext("Worker count adjusted: {}").format(result['message']), 'notice')
                    elif result['status'] == 'not_supported':
                        flash(gettext("Dynamic worker adjustment not supported for sync workers"), 'warning')
                    elif result['status'] == 'error':
                        flash(gettext("Error adjusting workers: {}").format(result['message']), 'error')

                if not os.getenv("SALTED_PASS", False) and len(form.application.form.password.encrypted_password):
                    datastore.data['settings']['application']['password'] = form.application.form.password.encrypted_password
                    datastore.commit()
                    flash(gettext("Password protection enabled."), 'notice')
                    flask_login.logout_user()
                    return redirect(url_for('watchlist.index'))

                # Also save plugin settings from the same form submission
                plugin_tabs_list = get_plugin_settings_tabs()
                for tab in plugin_tabs_list:
                    plugin_id = tab['plugin_id']
                    form_class = tab['form_class']

                    # Instantiate plugin form with POST data
                    plugin_form = form_class(formdata=request.form)

                    # Save plugin settings (validation is optional for plugins)
                    if plugin_form.data:
                        save_plugin_settings(datastore.datastore_path, plugin_id, plugin_form.data)

                flash(gettext("Settings updated."))

            else:
                flash(gettext("An error occurred, please see below."), "error")

        # Convert to ISO 8601 format, all date/time relative events stored as UTC time
        utc_time = datetime.now(ZoneInfo("UTC")).isoformat()

        # Get active plugins
        from changedetectionio.pluggy_interface import get_active_plugins
        import sys
        active_plugins = get_active_plugins()
        python_version = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"

        # Calculate uptime in seconds
        uptime_seconds = time.time() - datastore.start_time

        # Get plugin settings tabs and instantiate forms
        plugin_tabs = get_plugin_settings_tabs()
        plugin_forms = {}

        for tab in plugin_tabs:
            plugin_id = tab['plugin_id']
            form_class = tab['form_class']

            # Load existing settings
            settings = load_plugin_settings(datastore.datastore_path, plugin_id)

            # Instantiate the form with existing settings
            plugin_forms[plugin_id] = form_class(data=settings)

        from changedetectionio.llm.evaluator import (
            get_llm_config as _get_llm_cfg,
            llm_configured_via_env,
            get_global_token_budget_month,
        )
        llm_config = _get_llm_cfg(datastore) or {}
        llm_env_configured = llm_configured_via_env()
        llm_stored = datastore.data['settings']['application'].get('llm') or {}
        llm_token_budget_month = get_global_token_budget_month(datastore)
        llm_token_budget_month_env = get_global_token_budget_month()  # env var only, for readonly logic
        _max_input_chars_env_str = os.getenv('LLM_MAX_INPUT_CHARS', '').strip()
        llm_max_input_chars_env = int(_max_input_chars_env_str) if _max_input_chars_env_str.isdigit() else 0
        from changedetectionio.llm.evaluator import _get_max_input_chars, _DEFAULT_MAX_INPUT_CHARS
        llm_effective_max_input_chars = _get_max_input_chars(datastore)
        # Cost display: only when user configured their own key (not hosted/operator-managed)
        llm_show_costs = not llm_env_configured

        output = render_template("settings.html",
                                active_plugins=active_plugins,
                                api_key=datastore.data['settings']['application'].get('api_access_token'),
                                llm_config=llm_config,
                                llm_env_configured=llm_env_configured,
                                llm_stored=llm_stored,
                                llm_token_budget_month=llm_token_budget_month,
                                llm_token_budget_month_env=llm_token_budget_month_env,
                                llm_max_input_chars_env=llm_max_input_chars_env,
                                llm_effective_max_input_chars=llm_effective_max_input_chars,
                                llm_show_costs=llm_show_costs,
                                python_version=python_version,
                                uptime_seconds=uptime_seconds,
                                available_timezones=sorted(available_timezones()),
                                emailprefix=os.getenv('NOTIFICATION_MAIL_BUTTON_PREFIX', False),
                                extra_notification_token_placeholder_info=datastore.get_unique_notification_token_placeholders_available(),
                                form=form,
                                hide_remove_pass=os.getenv("SALTED_PASS", False),
                                min_system_recheck_seconds=int(os.getenv('MINIMUM_SECONDS_RECHECK_TIME', 3)),
                                settings_application=datastore.data['settings']['application'],
                                timezone_default_config=datastore.data['settings']['application'].get('scheduler_timezone_default'),
                                utc_time=utc_time,
                                plugin_tabs=plugin_tabs,
                                plugin_forms=plugin_forms,
                                )

        return output

    @settings_blueprint.route("/reset-api-key", methods=['GET'])
    @login_optionally_required
    def settings_reset_api_key():
        secret = secrets.token_hex(16)
        datastore.data['settings']['application']['api_access_token'] = secret
        datastore.commit()
        flash(gettext("API Key was regenerated."))
        return redirect(url_for('settings.settings_page')+'#api')
        
    @settings_blueprint.route("/notification-logs", methods=['GET'])
    @login_optionally_required
    def notification_logs():
        from changedetectionio.flask_app import notification_debug_log
        output = render_template("notification-log.html",
                               logs=notification_debug_log if len(notification_debug_log) else ["Notification logs are empty - no notifications sent yet."])
        return output

    @settings_blueprint.route("/toggle-all-paused", methods=['GET'])
    @login_optionally_required
    def toggle_all_paused():
        current_state = datastore.data['settings']['application'].get('all_paused', False)
        datastore.data['settings']['application']['all_paused'] = not current_state
        datastore.commit()

        if datastore.data['settings']['application']['all_paused']:
            flash(gettext("Automatic scheduling paused - checks will not be queued."), 'notice')
        else:
            flash(gettext("Automatic scheduling resumed - checks will be queued normally."), 'notice')

        return redirect(url_for('watchlist.index'))

    @settings_blueprint.route("/toggle-all-muted", methods=['GET'])
    @login_optionally_required
    def toggle_all_muted():
        current_state = datastore.data['settings']['application'].get('all_muted', False)
        datastore.data['settings']['application']['all_muted'] = not current_state
        datastore.commit()

        if datastore.data['settings']['application']['all_muted']:
            flash(gettext("All notifications muted."), 'notice')
        else:
            flash(gettext("All notifications unmuted."), 'notice')

        return redirect(url_for('watchlist.index'))

    return settings_blueprint