# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""debusine Web forms."""

import datetime as dt
from collections.abc import Callable
from typing import Any, ParamSpec, TYPE_CHECKING, TypeVar, cast, override

import yaml
from django.core.exceptions import ValidationError
from django.db.models import Q
from django.db.models.fields import BLANK_CHOICE_DASH
from django.forms import (
    BaseFormSet,
    BooleanField,
    CharField,
    CheckboxInput,
    ChoiceField,
    DateTimeField,
    Field,
    FileField,
    Form,
    IntegerField,
    ModelChoiceField,
    ModelForm,
    MultipleChoiceField,
    NumberInput,
    Textarea,
    formset_factory,
)
from django.forms.fields import InvalidJSONInput, JSONString

from debusine.artifacts.models import (
    BareDataCategory,
    CollectionCategory,
    SINGLETON_COLLECTION_CATEGORIES,
)
from debusine.db.context import context
from debusine.db.models import (
    Collection,
    Group,
    Token,
    User,
    WorkRequest,
    Workspace,
)
from debusine.db.models.collections import CollectionRelation
from debusine.db.permissioncontext import PermissionUser

if TYPE_CHECKING:
    from debusine.web.views.ui.workspaces import WorkspaceUI

    BaseFormSetBase = BaseFormSet
    ModelFormBase = ModelForm
    ModelChoiceFieldBase = ModelChoiceField
else:
    # Django's ModelForm and ModelChoiceField don't support generic types at
    # run-time yet.
    class _BaseFormSetBase:
        def __class_getitem__(
            cls,
            *args: TypeVar | ParamSpec | tuple[TypeVar | ParamSpec, ...],
        ) -> type[BaseFormSet]:
            return BaseFormSet

    class _ModelFormBase:
        def __class_getitem__(
            cls,
            *args: TypeVar | ParamSpec | tuple[TypeVar | ParamSpec, ...],
        ) -> type[ModelForm]:
            return ModelForm

    class _ModelChoiceFieldBase:
        def __class_getitem__(
            cls,
            *args: TypeVar | ParamSpec | tuple[TypeVar | ParamSpec, ...],
        ) -> type[ModelChoiceField]:
            return ModelChoiceField

    BaseFormSetBase = _BaseFormSetBase
    ModelFormBase = _ModelFormBase
    ModelChoiceFieldBase = _ModelChoiceFieldBase


class DictWithCallback[KT, VT](dict[KT, VT]):
    r"""
    Dictionary that when setting a value it calls a method to process it.

    Usage: d = DictWithCallback(callback, \*dict_args, \*\*dict_kwargs)

    when doing:
    d["key"] = value

    Before setting the value it calls "callback" which can change, in place,
    the value. Then sets the value to the dictionary.

    When accessing the value (d["key"]) it return the value as it was modified
    by the callback.
    """

    def __init__(
        self, callback: Callable[[VT], None], *args: Any, **kwargs: Any
    ) -> None:
        """Create the object."""
        self._callback = callback
        super().__init__(*args, **kwargs)

    @override
    def __setitem__(self, key: KT, value: VT) -> None:
        """Call self._callback(value) and set the value."""
        self._callback(value)
        super().__setitem__(key, value)


class BootstrapMixin:
    """
    Mixin that adjusts the CSS classes of form fields with Bootstrap's UI.

    This mixin is intended to be used in combination with Django's form classes.
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize the mixin."""
        super().__init__(*args, **kwargs)

        self._adjust_bootstrap_classes_all_fields()

        self.fields: dict[str, Field] = DictWithCallback(
            self._adjust_bootstrap_for_field, self.fields
        )

    @staticmethod
    def _adjust_bootstrap_for_field(field: Field) -> None:
        """Adjust the CSS class for a field."""
        existing_class = field.widget.attrs.get("class", "")
        bootstrap_class = None

        if field.required:
            suffix = " *"
            if not (field.label_suffix or "").endswith(suffix):
                field.label_suffix = (field.label_suffix or "") + suffix

        if isinstance(field, ChoiceField):
            bootstrap_class = "form-select"
        elif isinstance(
            field,
            (CharField, DateTimeField, DaysField, FileField, IntegerField),
        ):
            bootstrap_class = "form-control"
        elif isinstance(field, BooleanField):
            bootstrap_class = "form-check-input"

        if bootstrap_class and bootstrap_class not in existing_class.split():
            field.widget.attrs["class"] = (
                f"{existing_class} {bootstrap_class}".strip()
            )

    def _adjust_bootstrap_classes_all_fields(self) -> None:
        """Adjust the CSS classes of form fields to be Bootstrap-compatible."""
        for field in self.fields.values():
            self._adjust_bootstrap_for_field(field)


class OrderingWidget(NumberInput):
    """Local customization of the ordering widget for formsets."""

    def __init__(self, attrs: dict[str, Any] | None = None) -> None:
        """Add form-control to the widget class."""
        super().__init__(attrs)
        existing_class = self.attrs.get("class", "")
        self.attrs["class"] = f"{existing_class} form-control"


class DeletionWidget(CheckboxInput):
    """Local customization of the deletion widget for formsets."""

    def __init__(self, attrs: dict[str, Any] | None = None) -> None:
        """Add form-check-input to the widget class."""
        super().__init__(attrs)
        existing_class = self.attrs.get("class", "")
        self.attrs["class"] = f"{existing_class} form-check-input"


class TokenForm(BootstrapMixin, ModelFormBase[Token]):
    """Form for creating or editing a token."""

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize TokenForm."""
        self.user = kwargs.pop("user")
        super().__init__(*args, **kwargs)

        if not self.instance.pk:
            # New instance (not loaded from the DB). Set defaults
            self.fields["enabled"].initial = True

    @override
    def save(self, commit: bool = True) -> Token:
        instance = super().save(commit=False)

        instance.token_type = Token.Types.USER
        instance.user = self.user

        if commit:
            instance.save()

        return instance

    class Meta:
        model = Token
        fields = ["comment", "enabled"]

        labels = {
            "comment": "Comment",
        }


class WorkspaceChoiceField(ModelChoiceFieldBase[Workspace]):
    """ChoiceField for the workspaces: set the label and order by name."""

    def __init__(
        self, user: PermissionUser | None, *args: Any, **kwargs: Any
    ) -> None:
        """Set the queryset."""
        kwargs["queryset"] = Workspace.objects.order_by("name")

        if user is None:
            # Non-authenticated users can list only public workspaces
            kwargs["queryset"] = kwargs["queryset"].filter(public=True)

        super().__init__(*args, **kwargs)

    @override
    def label_from_instance(self, obj: Workspace) -> str:
        """Return name of the workspace."""
        return obj.name


class YAMLField(CharField):
    """Edit structured data, such as a JSONField, as YAML."""

    # Note: this is mostly copied from django.forms.fields.JSONField

    default_error_messages = {
        "invalid": "Enter valid YAML data.",
    }
    widget = Textarea
    dump_kwargs: dict[str, Any] = {
        "allow_unicode": True,
        "explicit_end": False,
        "explicit_start": False,
        "sort_keys": False,
    }

    @override
    def __init__(
        self,
        encoder: Any = None,  # noqa: U100
        decoder: Any = None,  # noqa: U100
        **kwargs: Any,
    ) -> None:
        if kwargs.get("empty_value") in (dict, list):
            kwargs["empty_value"] = kwargs["empty_value"]()
        super().__init__(**kwargs)
        encoder  # fake usage for Vulture
        decoder  # fake usage for Vulture

    @override
    def to_python(self, value: Any) -> Any:
        if self.disabled:
            return value
        if value in self.empty_values:
            return self.empty_value
        elif isinstance(value, (list, dict, int, float, JSONString)):
            return value
        try:
            converted = yaml.safe_load(value)
        except yaml.YAMLError:
            raise ValidationError(
                self.error_messages["invalid"],
                code="invalid",
                params={"value": value},
            )
        if isinstance(converted, str):
            return JSONString(converted)
        else:
            return converted

    @override
    def bound_data(self, data: Any, initial: Any) -> Any:
        if self.disabled:
            return initial
        if data is None:
            return None
        try:
            return yaml.safe_load(data)
        except yaml.YAMLError:
            return InvalidJSONInput(data)

    @override
    def prepare_value(self, value: Any) -> str:
        if isinstance(value, InvalidJSONInput):
            return value
        if value in ("", None):
            value = self.empty_value
        res = yaml.safe_dump(value, **self.dump_kwargs)
        assert isinstance(res, str)
        return res

    @override
    def has_changed(self, initial: Any, data: Any) -> bool:
        if super().has_changed(initial, data):
            return True
        # For purposes of seeing whether something has changed, True isn't the
        # same as 1 and the order of keys doesn't matter.
        return yaml.safe_dump(initial, sort_keys=True) != yaml.safe_dump(
            self.to_python(data), sort_keys=True
        )


class WorkRequestUnblockForm(BootstrapMixin, Form):
    """Form for reviewing a work request awaiting manual approval."""

    # Django defaults to 10 rows, which is a bit much.  Just make it clear
    # that we accept multi-line input.
    notes = CharField(
        required=False, empty_value=None, widget=Textarea(attrs={"rows": 3})
    )


class WorkRequestConfirmForm(BootstrapMixin, Form):
    """Form for confirming a work request."""

    # Django defaults to 10 rows, which is a bit much.  Just make it clear
    # that we accept multi-line input.
    comment = CharField(
        required=False, empty_value=None, widget=Textarea(attrs={"rows": 3})
    )


class CollectionSearchForm(BootstrapMixin, Form):
    """Form for collection search fields."""

    category = ChoiceField(required=False)
    name = CharField(required=False)
    historical = BooleanField(required=False)

    def __init__(self, *args: Any, instance: Collection, **kwargs: Any) -> None:
        """Initialize category choices from the database."""
        super().__init__(*args, **kwargs)
        self.instance = instance
        choices = [("", "All")]
        choices.extend(
            (name, name)
            for name in self.instance.child_items.values_list(
                "category", flat=True
            )
            .distinct()
            .order_by("category")
        )
        cast(ChoiceField, self.fields["category"]).choices = choices


class WorkflowFilterForm(Form):
    """Form for filtering workflows."""

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize form."""
        super().__init__(*args, **kwargs)

        statuses: list[tuple[str, str | list[tuple[str, str]]]] = []
        for status_value, status_label in WorkRequest.Statuses.choices:
            if status_value == WorkRequest.Statuses.RUNNING:
                running_options: list[tuple[str, str]] = []
                for (
                    runtime_value,
                    runtime_label,
                ) in WorkRequest.RuntimeStatuses.choices:
                    running_options.append(
                        (
                            f"running__{runtime_value}",
                            f"Running (children: {runtime_label})",
                        )
                    )

                # Add "Any" for any Running status.
                running_options.append(
                    ("running__any", "Running (children: Any)")
                )
                statuses.append(("Running", running_options))
            else:
                statuses.append((status_value, status_label))

        self.fields["statuses"] = MultipleChoiceField(
            choices=statuses,
            required=False,
        )

        runtime_statuses_choices = [
            (status.value, status.label)
            for status in WorkRequest.RuntimeStatuses
        ]

        self.fields["runtime_statuses"] = MultipleChoiceField(
            choices=runtime_statuses_choices,
            required=False,
        )

        self.fields["results"] = MultipleChoiceField(
            choices=sorted(
                choice for choice in WorkRequest.Results.choices if choice[0]
            ),
            required=False,
        )

        self.fields["with_failed_work_requests"] = BooleanField(
            required=False,
        )


class DaysField(IntegerField):
    """Edit a timedelta as a number of days."""

    @override
    def prepare_value(self, value: Any) -> Any:
        """Convert a timedelta to days."""
        if isinstance(value, dt.timedelta):
            return super().prepare_value(value.days)
        return super().prepare_value(value)

    @override
    def to_python(self, value: Any) -> Any:
        """Convert days to timedelta."""
        res = super().to_python(value)
        if res is not None:
            return dt.timedelta(days=res)
        return res


class WorkspaceForm(BootstrapMixin, ModelFormBase[Workspace]):
    """Form for configuring a workspace."""

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Set help texts for the fields."""
        super().__init__(*args, **kwargs)

        self.fields["expiration_delay"].help_text = (
            "If unset, workspace is permanent. Otherwise, the workspace gets "
            "removed after the given period of inactivity (expressed as a "
            "number of days after the completion of the last work request)"
        )
        self.fields["default_expiration_delay"].help_text = (
            "Number of days that new artifacts and work requests are kept "
            "in the workspace before being expired (0 means no expiration)"
        )

    class Meta:
        model = Workspace
        fields = ["public", "expiration_delay", "default_expiration_delay"]
        field_classes = {
            "expiration_delay": DaysField,
            "default_expiration_delay": DaysField,
        }


class WorkspaceInheritanceForm(BootstrapMixin, Form):
    """Form for one item in the inheritance chain of a workspace."""

    parent = ChoiceField()

    def __init__(
        self, workspace_ui: "WorkspaceUI", *args: Any, **kwargs: Any
    ) -> None:
        """Set help texts for the fields."""
        super().__init__(*args, **kwargs)
        self.workspace_ui = workspace_ui
        cast(ChoiceField, self.fields["parent"]).choices = [(None, "None")] + [
            (ws.pk, str(ws)) for ws in self.workspace_ui.candidate_parents
        ]


class WorkspaceInheritanceFormSet(BaseFormSetBase[WorkspaceInheritanceForm]):
    """FormSet to edit an inheritance chain."""

    ordering_widget = OrderingWidget
    deletion_widget = DeletionWidget

    def __init__(
        self, *args: Any, workspace_ui: "WorkspaceUI", **kwargs: Any
    ) -> None:
        """Store a WorkspaceUI to use to fill in form choices."""
        super().__init__(*args, **kwargs)
        self.workspace_ui = workspace_ui

    @override
    def get_form_kwargs(self, index: int | None) -> dict[str, Any]:
        """Pass the WorkspaceUI to use to fill in form choices."""
        kwargs = super().get_form_kwargs(index)
        kwargs["workspace_ui"] = self.workspace_ui
        return kwargs

    @classmethod
    def formset_factory(
        cls, workspace_ui: "WorkspaceUI"
    ) -> type[BaseFormSetBase[WorkspaceInheritanceForm]]:
        """Create a Form class for a WorkspaceUI."""

        class Partial(WorkspaceInheritanceFormSet):
            def __init__(
                self,
                *args: Any,
                **kwargs: Any,
            ) -> None:
                """Store a WorkspaceUI to use to fill in form choices."""
                super().__init__(*args, workspace_ui=workspace_ui, **kwargs)

        return formset_factory(
            WorkspaceInheritanceForm,
            formset=Partial,
            can_order=True,
            can_delete=True,
            extra=1,
        )


class GroupAddUserForm(BootstrapMixin, Form):
    """Form for adding a user to a group."""

    username = CharField(required=True)

    def __init__(self, group: Group, *args: Any, **kwargs: Any) -> None:
        """Set the group the user would be added to."""
        self.group = group
        super().__init__(*args, **kwargs)

    @override
    def clean(self) -> dict[str, Any]:
        """
        Validate username.

        The user needs to exist and not be already a member of the group.
        """
        cleaned_data = super().clean()
        assert cleaned_data is not None

        if "username" not in cleaned_data:
            return cleaned_data

        try:
            user = User.objects.get(username=cleaned_data["username"])
        except User.DoesNotExist:
            self.add_error(
                "username", f"User {cleaned_data['username']} does not exist"
            )
            return cleaned_data

        if self.group.users.filter(pk=user.pk).exists():
            self.add_error(
                "username", f"User is already a member of {self.group}"
            )

        return cleaned_data


class TaskConfigurationInspectorForm(BootstrapMixin, Form):
    """Form for looking up task configurations."""

    task = ChoiceField()
    subject = ChoiceField(required=False)
    context = ChoiceField(required=False)
    tags_provided = CharField(
        required=False,
        help_text="space-separated list of task tags a task provides",
    )
    tags_required = CharField(
        required=False,
        help_text="space-separated list of worker tags a task requires",
    )

    def __init__(
        self, *args: Any, collection: Collection, **kwargs: Any
    ) -> None:
        """Initialize choices from the database."""
        super().__init__(*args, **kwargs)
        children = collection.child_items.filter(
            Q(data__template=None) | ~Q(data__has_key="template"),
            category=BareDataCategory.TASK_CONFIGURATION,
        )

        task_choices = []
        for ttype, tname in (
            children.order_by("data__task_type", "data__task_name")
            .values_list("data__task_type", "data__task_name")
            .distinct()
        ):
            task_choices.append((f"{ttype}:{tname}", f"{ttype}:{tname}"))
        cast(ChoiceField, self.fields["task"]).choices = task_choices

        subject_choices = [("", "(unspecified)")]
        for subject in (
            children.order_by("data__subject")
            .values_list("data__subject", flat=True)
            .distinct()
        ):
            if subject is not None:
                subject_choices.append((subject, subject))
        cast(ChoiceField, self.fields["subject"]).choices = subject_choices

        context_choices = [("", "(unspecified)")]
        for child_context in (
            children.order_by("data__context")
            .values_list("data__context", flat=True)
            .distinct()
        ):
            if child_context is not None:
                context_choices.append((child_context, child_context))
        cast(ChoiceField, self.fields["context"]).choices = context_choices


class CollectionFormBase(BootstrapMixin, ModelFormBase[Collection]):
    """Common functions for collection forms."""

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize object."""
        super().__init__(*args, **kwargs)
        # Add measurement unit to interval fields
        for field_name in (
            "full_history_retention_period",
            "metadata_only_retention_period",
        ):
            if field := self.fields.get(field_name):
                field.help_text = field.help_text + " (in days)"

    def check_name(
        self,
        workspace: Workspace,
        name: str,
        category: CollectionCategory,
        instance: Collection | None = None,
    ) -> None:
        """Check that the name field matches collection constraints."""
        conflicts = Collection.objects.filter(
            name=name, workspace=workspace, category=category
        )
        if instance is not None:
            conflicts = conflicts.exclude(pk=instance.pk)
        if conflicts.exists():
            self.add_error(
                "name",
                f"A collection called {name}@{category}"
                f" already exists in workspace {workspace}",
            )

        if category in SINGLETON_COLLECTION_CATEGORIES and name != "_":
            self.add_error(
                "name", 'A singleton collection needs to be named "_"'
            )
        if category not in SINGLETON_COLLECTION_CATEGORIES and name.startswith(
            "_"
        ):
            self.add_error(
                "name",
                'The name of a non-singleton collection cannot start with "_"',
            )


class CollectionUpdateForm(CollectionFormBase):
    """Form for configuring a collection."""

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize object."""
        super().__init__(*args, **kwargs)
        # Avoid conflicting with Form.data
        self.fields["data"] = YAMLField(required=False, empty_value=dict)

    @override
    def clean(self) -> dict[str, Any]:
        """Enforce collection name constraints."""
        cleaned_data = super().clean()
        assert cleaned_data is not None
        self.check_name(
            self.instance.workspace,
            cleaned_data["name"],
            CollectionCategory(self.instance.category),
            instance=self.instance,
        )
        return cleaned_data

    class Meta:
        model = Collection
        fields = [
            "name",
            "full_history_retention_period",
            "metadata_only_retention_period",
            "data",
        ]
        field_classes = {
            "full_history_retention_period": DaysField,
            "metadata_only_retention_period": DaysField,
        }


class CollectionCreateForm(CollectionFormBase):
    """Form for creating a collection."""

    category = ChoiceField()

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        """Initialize object."""
        super().__init__(*args, **kwargs)
        # Avoid conflicting with Form.data
        self.fields["data"] = YAMLField(required=False, empty_value=dict)

        assert isinstance(self.fields["category"], ChoiceField)
        self.fields["category"].choices = [
            (c, c)
            for c in sorted(CollectionCategory)
            if c != CollectionCategory.WORKFLOW_INTERNAL
        ]

    @override
    def clean(self) -> dict[str, Any]:
        """Enforce collection name constraints."""
        cleaned_data = super().clean()
        assert cleaned_data is not None
        if "name" in cleaned_data and "category" in cleaned_data:
            self.check_name(
                context.require_workspace(),
                str(cleaned_data["name"]),
                CollectionCategory(cleaned_data["category"]),
            )
        return cleaned_data

    @override
    def save(self, commit: bool = True) -> Collection:
        instance = super().save(commit=False)
        instance.workspace = context.require_workspace()
        instance.retains_artifacts = Collection.RetainsArtifacts.ALWAYS
        if commit:
            instance.save()
        return instance

    class Meta:
        model = Collection
        fields = [
            "name",
            "category",
            "full_history_retention_period",
            "metadata_only_retention_period",
            "data",
        ]
        field_classes = {
            "full_history_retention_period": DaysField,
            "metadata_only_retention_period": DaysField,
        }


class CollectionRelationCreateForm(
    BootstrapMixin, ModelFormBase[CollectionRelation]
):
    """Form for creating a collection relation."""

    source_collection: Collection

    def __init__(
        self,
        *args: Any,
        source_collection: Collection,
        allowed_types: set[CollectionRelation.Types],
        **kwargs: Any,
    ) -> None:
        """Initialize form."""
        super().__init__(*args, **kwargs)

        self.source_collection = source_collection

        # Relation targets sorted by name
        target_field = self.fields["target"]
        assert hasattr(target_field, "queryset")
        target_field.queryset = target_field.queryset.order_by("name")

        # Relation types: only the possible ones are displayed (and sorted)
        type_field = self.fields["type"]
        assert hasattr(type_field, "choices")

        self.fields[
            "target"
        ].help_text = "Only targets allowed for the selected type are shown"

        # Remove the type_fields that are not allowed and also the blank choice
        type_field.choices = sorted(
            [
                (value, label)
                for value, label in type_field.choices
                if (value, label) not in BLANK_CHOICE_DASH
                and value in allowed_types
            ]
        )

        type_field.help_text = (
            "Some types may be hidden if they have no available targets"
        )

    class Meta:
        model = CollectionRelation
        fields = ["type", "target"]
