neptune/core/forms.py

from django import forms
from django.contrib.auth.forms import UserCreationForm
from django.core.exceptions import FieldDoesNotExist
from django.forms import ModelForm
from mixins.restrictions import RestrictedFormMixin

from core.db.storage import db
from core.lib.parsing import QueryError
from core.lib.rules import NotificationRuleData, RuleParseError

from .models import NotificationRule, NotificationSettings, User

# flake8: noqa: E501


class NewUserForm(UserCreationForm):
    email = forms.EmailField(required=True)

    class Meta:
        model = User
        fields = (
            "username",
            "email",
            "first_name",
            "last_name",
            "password1",
            "password2",
        )

    def save(self, commit=True):
        user = super(NewUserForm, self).save(commit=False)
        user.email = self.cleaned_data["email"]
        if commit:
            user.save()
        return user


class CustomUserCreationForm(UserCreationForm):
    class Meta:
        model = User
        fields = "__all__"


class NotificationSettingsForm(RestrictedFormMixin, ModelForm):
    def __init__(self, *args, **kwargs):
        super(NotificationSettingsForm, self).__init__(*args, **kwargs)
        self.fields["url"].label = "URL"

    class Meta:
        model = NotificationSettings
        fields = (
            "topic",
            "url",
            "service",
        )
        help_texts = {
            "topic": "The topic to send notifications to.",
            "url": "Custom NTFY server/webhook destination. Leave blank to use the default server for NTFY. For webhooks this field is required.",
            "service": "The service to use for notifications.",
        }

    def clean(self):
        cleaned_data = super(NotificationSettingsForm, self).clean()
        if "service" in cleaned_data:
            if cleaned_data["service"] == "webhook":
                if not cleaned_data.get("url"):
                    self.add_error(
                        "url",
                        "You must set a URL for webhooks.",
                    )


class NotificationRuleForm(RestrictedFormMixin, ModelForm):
    def __init__(self, *args, **kwargs):
        super(NotificationRuleForm, self).__init__(*args, **kwargs)
        self.fields["url"].label = "URL"

    class Meta:
        model = NotificationRule
        fields = (
            "name",
            "data",
            "interval",
            "window",
            "amount",
            "priority",
            "topic",
            "url",
            "service",
            "policy",
            "enabled",
        )
        help_texts = {
            "name": "The name of the rule.",
            "priority": "The notification priority of the rule.",
            "url": "Custom NTFY server/webhook destination. Leave blank to use the default server for NTFY. For webhooks this field is required.",
            "service": "The service to use for notifications",
            "topic": "The topic to send notifications to. Leave blank for default.",
            "enabled": "Whether the rule is enabled.",
            "data": "The notification rule definition.",
            "interval": "How often to run the search. On demand evaluates messages as they are received, without running a scheduled search. The remaining options schedule a search of the database with the window below.",
            "window": "Time window to search: 1d, 1h, 1m, 1s, etc.",
            "amount": "Amount of matches to be returned for scheduled queries. Cannot be used with on-demand queries.",
            "policy": "When to trigger this policy.",
        }

    def clean(self):
        cleaned_data = super(NotificationRuleForm, self).clean()

        # TODO: should this be in rules.py?
        if "service" in cleaned_data:
            if cleaned_data["service"] == "webhook":
                if not cleaned_data.get("url"):
                    self.add_error(
                        "url",
                        "You must set a URL for webhooks.",
                    )
        try:
            # Passing db to avoid circular import
            parsed_data = NotificationRuleData(self.request.user, cleaned_data, db=db)
            if cleaned_data["enabled"]:
                parsed_data.test_schedule()
        except RuleParseError as e:
            self.add_error(e.field, f"Parsing error: {e}")
            return
        except QueryError as e:
            self.add_error("data", f"Query error: {e}")
            return

        # Write back the validated data
        # We need this to populate the index and source variable if
        # they are not set
        to_store = str(parsed_data)
        cleaned_data["data"] = to_store

        return cleaned_data