admin.py

import json
import logging
from datetime import timedelta

import reversion
import swapper
from django import forms
from django.conf import settings
from django.contrib import admin, messages
from django.contrib.admin.helpers import ACTION_CHECKBOX_NAME
from django.core.paginator import InvalidPage, Paginator
from django.core.serializers.json import DjangoJSONEncoder
from django.shortcuts import redirect
from django.template.response import TemplateResponse
from django.templatetags.static import static
from django.urls import resolve, reverse
from django.utils.html import format_html
from django.utils.safestring import mark_safe
from django.utils.timezone import localtime
from django.utils.translation import get_language
from django.utils.translation import gettext_lazy as _
from reversion.admin import VersionAdmin

from openwisp_controller.config.admin import DeactivatedDeviceReadOnlyMixin, DeviceAdmin
from openwisp_users.multitenancy import MultitenantAdminMixin, MultitenantOrgFilter
from openwisp_utils.admin import ReadOnlyAdmin, TimeReadonlyAdminMixin

from .filters import (
    BuildCategoryFilter,
    BuildCategoryOrganizationFilter,
    CategoryFilter,
    CategoryOrganizationFilter,
)
from .swapper import load_model
from .utils import get_upgrader_schema_for_device
from .widgets import FirmwareSchemaWidget

logger = logging.getLogger(__name__)
BatchUpgradeOperation = load_model("BatchUpgradeOperation")
UpgradeOperation = load_model("UpgradeOperation")
DeviceFirmware = load_model("DeviceFirmware")
FirmwareImage = load_model("FirmwareImage")
Category = load_model("Category")
Build = load_model("Build")
Device = swapper.load_model("config", "Device")
DeviceConnection = swapper.load_model("connection", "DeviceConnection")
Organization = swapper.load_model("openwisp_users", "Organization")


class BaseAdmin(MultitenantAdminMixin, TimeReadonlyAdminMixin, admin.ModelAdmin):
    save_on_top = True


class BaseVersionAdmin(MultitenantAdminMixin, TimeReadonlyAdminMixin, VersionAdmin):
    history_latest_first = True
    save_on_top = True


@admin.register(load_model("Category"))
class CategoryAdmin(BaseVersionAdmin):
    list_display = ["name", "organization", "created", "modified"]
    list_filter = [MultitenantOrgFilter]
    list_select_related = ["organization"]
    search_fields = ["name"]
    ordering = ["-name", "-created"]


class FirmwareImageInline(TimeReadonlyAdminMixin, admin.StackedInline):
    model = FirmwareImage
    extra = 0

    class Media:
        extra = "" if getattr(settings, "DEBUG", False) else ".min"
        i18n_name = admin.widgets.SELECT2_TRANSLATIONS.get(get_language())
        i18n_file = (
            ("admin/js/vendor/select2/i18n/%s.js" % i18n_name,) if i18n_name else ()
        )
        js = (
            (
                "admin/js/vendor/jquery/jquery%s.js" % extra,
                "admin/js/vendor/select2/select2.full%s.js" % extra,
            )
            + i18n_file
            + ("admin/js/jquery.init.js", "firmware-upgrader/js/build.js")
        )

        css = {
            "screen": ("admin/css/vendor/select2/select2%s.css" % extra,),
        }

    def has_change_permission(self, request, obj=None):
        if obj:
            return False
        return True


class BatchUpgradeConfirmationForm(forms.ModelForm):
    upgrade_options = forms.JSONField(widget=FirmwareSchemaWidget(), required=False)
    build = forms.ModelChoiceField(
        widget=forms.HiddenInput(), required=False, queryset=Build.objects.all()
    )

    class Meta:
        model = BatchUpgradeOperation
        fields = ("build", "upgrade_options")

    @property
    def media(self):
        js = [
            "firmware-upgrader/js/upgrade-selected-confirmation.js",
        ]
        css = {"all": ["firmware-upgrader/css/upgrade-selected-confirmation.css"]}
        return super().media + forms.Media(js=js, css=css)


@admin.register(load_model("Build"))
class BuildAdmin(BaseAdmin):
    list_display = ["__str__", "organization", "category", "created", "modified"]
    list_filter = [CategoryOrganizationFilter, CategoryFilter]
    list_select_related = ["category", "category__organization"]
    search_fields = ["category__name", "version", "os"]
    ordering = ["-created", "-version"]
    inlines = [FirmwareImageInline]
    actions = ["upgrade_selected"]
    multitenant_parent = "category"
    autocomplete_fields = ["category"]

    # Allows apps that extend this modules to use this template with less hacks
    change_form_template = "admin/firmware_upgrader/change_form.html"

    def organization(self, obj):
        return obj.category.organization

    organization.short_description = _("organization")

    @admin.action(
        description=_("Mass-upgrade devices related to the selected build"),
        permissions=["change"],
    )
    def upgrade_selected(self, request, queryset):
        opts = self.model._meta
        app_label = opts.app_label
        # multiple concurrent batch upgrades are not supported
        # (it's not yet possible to select more builds and upgrade
        #  all of them at the same time)
        if queryset.count() > 1:
            self.message_user(
                request,
                _(
                    "Multiple mass upgrades requested but at the moment only "
                    "a single mass upgrade operation at time is supported."
                ),
                messages.ERROR,
            )
            # returning None will display the change list page again
            return None
        upgrade_all = request.POST.get("upgrade_all")
        upgrade_related = request.POST.get("upgrade_related")
        upgrade_options = request.POST.get("upgrade_options")
        form = BatchUpgradeConfirmationForm()
        build = queryset.first()
        # upgrade has been confirmed
        if upgrade_all or upgrade_related:
            form = BatchUpgradeConfirmationForm(
                data={"upgrade_options": upgrade_options, "build": build}
            )
            form.full_clean()
            if not form.errors:
                upgrade_options = form.cleaned_data["upgrade_options"]
                batch = build.batch_upgrade(
                    firmwareless=upgrade_all, upgrade_options=upgrade_options
                )
                text = _(
                    "You can track the progress of this mass upgrade operation "
                    "in this page. Refresh the page from time to time to check "
                    "its progress."
                )
                self.message_user(request, mark_safe(text), messages.SUCCESS)
                url = reverse(
                    f"admin:{app_label}_batchupgradeoperation_change", args=[batch.pk]
                )
                return redirect(url)
        # upgrade needs to be confirmed
        result = BatchUpgradeOperation.dry_run(build=build)
        related_device_fw = result["device_firmwares"]
        firmwareless_devices = result["devices"]
        title = _("Confirm mass upgrade operation")
        context = self.admin_site.each_context(request)
        upgrader_schema = BatchUpgradeOperation(build=build)._get_upgrader_schema(
            related_device_fw=related_device_fw,
            firmwareless_devices=firmwareless_devices,
        )

        context.update(
            {
                "title": title,
                "related_device_fw": related_device_fw,
                "related_count": len(related_device_fw),
                "firmwareless_devices": firmwareless_devices,
                "firmwareless_count": len(firmwareless_devices),
                "form": form,
                "firmware_upgrader_schema": json.dumps(
                    upgrader_schema, cls=DjangoJSONEncoder
                ),
                "build": build,
                "opts": opts,
                "action_checkbox_name": ACTION_CHECKBOX_NAME,
                "media": self.media,
            }
        )
        request.current_app = self.admin_site.name
        return TemplateResponse(
            request,
            [
                "admin/%s/%s/upgrade_selected_confirmation.html"
                % (app_label, opts.model_name),
                "admin/%s/upgrade_selected_confirmation.html" % app_label,
                "admin/upgrade_selected_confirmation.html",
            ],
            context,
        )

    upgrade_selected.short_description = (
        "Mass-upgrade devices related " "to the selected build"
    )

    def change_view(self, request, object_id, form_url="", extra_context=None):
        app_label = self.model._meta.app_label
        extra_context = extra_context or {}
        upgrade_url = f"{app_label}_build_changelist"
        extra_context.update({"upgrade_url": upgrade_url})
        return super().change_view(request, object_id, form_url, extra_context)


class UpgradeOperationForm(forms.ModelForm):
    class Meta:
        fields = ["device", "image", "status", "log", "modified"]
        labels = {"modified": _("last updated")}


class UpgradeOperationInline(admin.StackedInline):
    model = UpgradeOperation
    form = UpgradeOperationForm
    readonly_fields = UpgradeOperationForm.Meta.fields
    extra = 0

    def has_delete_permission(self, request, obj):
        return False

    def has_add_permission(self, request, obj):
        return False

    class Media:
        css = {"all": ["firmware-upgrader/css/upgrade-options.css"]}


class ReadonlyUpgradeOptionsMixin:

    class Media:
        css = {"all": ["firmware-upgrader/css/upgrade-options.css"]}

    @admin.display(description=_("Upgrade options"))
    def readonly_upgrade_options(self, obj):
        upgrader_schema = obj.upgrader_schema
        if not upgrader_schema:
            return _("Upgrade options are not supported for this upgrader.")
        options = []
        for key, value in upgrader_schema["properties"].items():
            option_used = "yes" if obj.upgrade_options.get(key, False) else "no"
            option_title = value["title"]
            icon_url = static(f"admin/img/icon-{option_used}.svg")
            options.append(
                f'<li><img src="{icon_url}" alt="{option_used}">{option_title}</li>'
            )
        return format_html(
            mark_safe(f'<ul class="readonly-upgrade-options">{"".join(options)}</ul>')
        )


@admin.register(UpgradeOperation)
class UpgradeOperationAdmin(ReadOnlyAdmin, BaseAdmin):
    list_display = ["device", "status", "image", "modified"]
    list_filter = ["status"]
    search_fields = ["device__name"]
    readonly_fields = ["device", "image", "status", "log", "modified"]
    ordering = ["-modified"]
    fields = ["device", "image", "status", "log", "modified"]
    change_form_template = "admin/firmware_upgrader/upgrade_operation_change_form.html"

    def has_add_permission(self, request):
        return False

    def has_delete_permission(self, request, obj=None):
        return False


@admin.register(BatchUpgradeOperation)
class BatchUpgradeOperationAdmin(ReadonlyUpgradeOptionsMixin, ReadOnlyAdmin, BaseAdmin):
    list_display = ["build", "organization", "status", "created", "modified"]
    list_filter = [
        BuildCategoryOrganizationFilter,
        "status",
        BuildCategoryFilter,
    ]
    list_select_related = ["build__category__organization"]
    ordering = ["-created"]
    multitenant_parent = "build__category"
    fields = [
        "build",
        "status",
        "completed",
        "success_rate",
        "failed_rate",
        "aborted_rate",
        "cancelled_rate",
        "readonly_upgrade_options",
        "created",
        "modified",
    ]
    autocomplete_fields = ["build"]
    readonly_fields = [
        "completed",
        "success_rate",
        "failed_rate",
        "aborted_rate",
        "cancelled_rate",
        "readonly_upgrade_options",
    ]
    change_form_template = (
        "admin/firmware_upgrader/batch_upgrade_operation_change_form.html"
    )
    device_upgrades_per_page = 20

    def get_queryset(self, request):
        qs = super().get_queryset(request)
        return qs.prefetch_related(
            "upgradeoperation_set__device", "upgradeoperation_set__image"
        )

    def get_upgrade_operations(self, obj):
        return obj.upgradeoperation_set.all().select_related("device", "image")

    def changelist_view(self, request, extra_context=None):
        extra_context = extra_context or {}
        extra_context["title"] = _("Mass upgrade operations")
        return super().changelist_view(request, extra_context)

    def change_view(self, request, object_id, form_url="", extra_context=None):
        extra_context = extra_context or {}
        obj = self.get_object(request, object_id)
        if obj:
            # Get upgrade operations with search and filtering
            upgrades_qs = self.get_upgrade_operations(obj)

            # Handle search
            search_query = request.GET.get("q", "")
            if search_query:
                upgrades_qs = upgrades_qs.filter(device__name__icontains=search_query)

            # Get current filter values
            current_status = request.GET.get("status", "")
            current_org = request.GET.get("organization", "")

            # Apply filters to queryset
            if current_status:
                upgrades_qs = upgrades_qs.filter(status=current_status)
            if current_org:
                upgrades_qs = upgrades_qs.filter(device__organization_id=current_org)

            # Create filter specs for the template
            filter_specs = []

            # Status filter
            status_choices = [
                {
                    "display": _("All"),
                    "selected": not current_status,
                    "query_string": "?",
                },
                {
                    "display": _("idle"),
                    "selected": current_status == "idle",
                    "query_string": "?status=idle",
                },
                {
                    "display": _("in progress"),
                    "selected": current_status == "in-progress",
                    "query_string": "?status=in-progress",
                },
                {
                    "display": _("completed successfully"),
                    "selected": current_status == "success",
                    "query_string": "?status=success",
                },
                {
                    "display": _("completed with some failures"),
                    "selected": current_status == "failed",
                    "query_string": "?status=failed",
                },
            ]

            class StatusFilter:
                title = _("status")
                choices = status_choices

            filter_specs.append(StatusFilter())

            # Organization filter
            org_choices = [
                {"display": _("All"), "selected": not current_org, "query_string": "?"}
            ]
            for org in Organization.objects.all().order_by("name"):
                org_choices.append(
                    {
                        "display": org.name,
                        "selected": current_org == str(org.id),
                        "query_string": f"?organization={org.id}",
                    }
                )

            class OrganizationFilter:
                title = _("organization")
                choices = org_choices

            filter_specs.append(OrganizationFilter())

            # Pagination
            paginator = Paginator(
                upgrades_qs.order_by("id"), self.device_upgrades_per_page
            )
            page_number = request.GET.get("page", 1)
            try:
                page_obj = paginator.page(page_number)
            except InvalidPage:
                page_obj = paginator.page(1)

            # Get the app label for URL construction
            upgrade_operation_app_label = load_model("UpgradeOperation")._meta.app_label

            extra_context.update(
                {
                    "upgrade_operations": page_obj.object_list,
                    "page_obj": page_obj,
                    "paginator": paginator,
                    "filter_specs": filter_specs,
                    "has_active_filters": any(
                        request.GET.get(param) for param in ["status", "organization"]
                    ),
                    "upgrade_operation_app_label": upgrade_operation_app_label,
                }
            )

        return super().change_view(request, object_id, extra_context)

    def get_readonly_fields(self, request, obj=None):
        fields = super().get_readonly_fields(request, obj)
        return fields + self.__class__.readonly_fields

    def organization(self, obj):
        return obj.build.category.organization

    organization.short_description = _("organization")

    def completed(self, obj):
        return obj.progress_report

    def success_rate(self, obj):
        return self.__get_rate(obj.success_rate)

    def failed_rate(self, obj):
        return self.__get_rate(obj.failed_rate)

    def aborted_rate(self, obj):
        return self.__get_rate(obj.aborted_rate)

    def cancelled_rate(self, obj):
        return self.__get_rate(obj.cancelled_rate)

    def __get_rate(self, value):
        if value:
            return f"{value}%"
        return "N/A"

    completed.short_description = _("completed")
    success_rate.short_description = _("success rate")
    failed_rate.short_description = _("failure rate")
    aborted_rate.short_description = _("abortion rate")
    cancelled_rate.short_description = _("cancellation rate")


class DeviceFirmwareForm(forms.ModelForm):
    upgrade_options = forms.JSONField(widget=FirmwareSchemaWidget, required=False)

    class Meta:
        model = DeviceFirmware
        fields = "__all__"

    class Media:
        js = ["admin/js/jquery.init.js", "firmware-upgrader/js/device-firmware.js"]
        css = {"all": ["firmware-upgrader/css/device-firmware.css"]}

    def __init__(self, device, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.fields["image"].queryset = DeviceFirmware.get_image_queryset_for_device(
            device, device_firmware=self.instance
        )

    def full_clean(self):
        super().full_clean()
        if not self.errors and hasattr(self, "cleaned_data"):
            upgrade_op = UpgradeOperation(
                device=self.cleaned_data["device"],
                image=self.cleaned_data["image"],
                upgrade_options=self.cleaned_data["upgrade_options"],
            )
            try:
                upgrade_op.full_clean()
            except forms.ValidationError as error:
                self.add_error("__all__", error.messages[0])

    def save(self, commit=True):
        """
        Adapted from ModelForm.save()
        Passes "upgrade_options to DeviceFirmware.save()
        """
        if commit:
            # If committing, save the instance and the m2m data immediately.
            self.instance.save(upgrade_options=self.cleaned_data["upgrade_options"])
        return self.instance


class DeviceFormSet(forms.BaseInlineFormSet):
    def get_form_kwargs(self, index):
        kwargs = super().get_form_kwargs(index)
        kwargs["device"] = self.instance
        return kwargs


class DeviceFirmwareInline(
    MultitenantAdminMixin, DeactivatedDeviceReadOnlyMixin, admin.StackedInline
):
    model = DeviceFirmware
    formset = DeviceFormSet
    form = DeviceFirmwareForm
    exclude = ["created"]
    select_related = ["device", "image"]
    readonly_fields = ["installed", "modified"]
    verbose_name = _("Firmware")
    verbose_name_plural = verbose_name
    extra = 0
    multitenant_shared_relations = ["device"]
    template = "admin/firmware_upgrader/device_firmware_inline.html"
    # hack for openwisp-monitoring integartion
    # TODO: remove when this issue solved:
    # https://github.com/theatlantic/django-nested-admin/issues/128#issuecomment-665833142
    sortable_options = {"disabled": True}

    class Media:
        js = [
            "connection/js/lib/reconnecting-websocket.min.js",
            "firmware-upgrader/js/upgrade-progress.js",
        ]
        css = {
            "all": [
                "firmware-upgrader/css/upgrade-progress.css",
            ]
        }

    def _get_conditional_queryset(self, request, obj, select_related=False):
        return bool(obj)

    def get_formset(self, request, obj=None, **kwargs):
        formset = super().get_formset(request, obj=obj, **kwargs)
        if obj:
            try:
                schema = get_upgrader_schema_for_device(obj)
                formset.extra_context = json.dumps(schema, cls=DjangoJSONEncoder)
            except DeviceConnection.DoesNotExist:
                # We cannot retrieve the schema for upgrade options because this
                # device does not have any related DeviceConnection object.
                pass
        return formset


class DeviceUpgradeOperationForm(UpgradeOperationForm):
    class Meta(UpgradeOperationForm.Meta):
        pass

    def __init__(self, device, *args, **kwargs):
        self.device = device
        super().__init__(*args, **kwargs)


class DeviceUpgradeOperationInline(ReadonlyUpgradeOptionsMixin, UpgradeOperationInline):
    verbose_name = _("Recent Firmware Upgrades")
    verbose_name_plural = verbose_name
    formset = DeviceFormSet
    form = DeviceUpgradeOperationForm
    # hack for openwisp-monitoring integration
    # TODO: remove when this issue solved:
    # https://github.com/theatlantic/django-nested-admin/issues/128#issuecomment-665833142
    sortable_options = {"disabled": True}
    fields = [
        "device",
        "image",
        "status",
        "log",
        "readonly_upgrade_options",
        "modified",
    ]
    readonly_fields = fields

    class Media:
        js = [
            "connection/js/lib/reconnecting-websocket.min.js",
            "firmware-upgrader/js/upgrade-progress.js",
        ]
        css = {
            "all": [
                "firmware-upgrader/css/upgrade-progress.css",
                "firmware-upgrader/css/upgrade-options.css",
            ]
        }

    def get_queryset(self, request, select_related=True):
        """
        Return recent upgrade operations for this device
        (created within the last 7 days)
        """
        qs = super().get_queryset(request)
        resolved = resolve(request.path_info)
        if "object_id" in resolved.kwargs:
            seven_days = localtime() - timedelta(days=7)
            qs = qs.filter(
                device_id=resolved.kwargs["object_id"], created__gte=seven_days
            ).order_by("-created")
        if select_related:
            qs = qs.select_related()
        return qs

    def _get_conditional_queryset(self, request, obj, select_related=False):
        if obj:
            return self.get_queryset(request, select_related=False).exists()
        return False


# DeviceAdmin.get_inlines = device_admin_get_inlines
DeviceAdmin.conditional_inlines += [DeviceFirmwareInline, DeviceUpgradeOperationInline]

reversion.register(model=DeviceFirmware, follow=["device"])
reversion.register(model=UpgradeOperation)
DeviceAdmin.add_reversion_following(follow=["devicefirmware", "upgradeoperation_set"])