penracourses/atlas/admin.py

from django.contrib import admin
from django.contrib import messages
from .models import (
    Case,
    CaseDetail,
    CaseCollection,
    NormalCase,
    Differential,
    Series,
    SeriesImage,
    Condition,
    ConditionRelationship,
    Subspecialty,
    Finding,
    SeriesFinding,
    Structure,
    PathologicalProcess,
    Presentation,
    Procedure,
    UncategorisedDicom,
    SeriesDetail,
    Resource,
    CaseDisplaySet,
    UserReportAnswer,
    CidReportAnswer,
    APIToken,
)

from django.forms import ModelForm

from reversion.admin import VersionAdmin
from django.forms.widgets import RadioSelect
from django.utils.html import format_html
from django.urls import reverse

from tinymce.widgets import TinyMCE

# from generic.models import Examination, Sign, Site, Condition
from generic.models import Examination

try:
    from django_tasks_db.models import DBTaskResult
except ImportError:
    DBTaskResult = None

# Register your models here.
# admin.site.register(Examination)
# admin.site.register(Site)
admin.site.register(SeriesImage)
admin.site.register(Condition)
admin.site.register(ConditionRelationship)
admin.site.register(Subspecialty)
admin.site.register(Finding)
admin.site.register(Structure)
admin.site.register(SeriesFinding)
admin.site.register(Differential)
admin.site.register(PathologicalProcess)
admin.site.register(Presentation)
admin.site.register(CaseDetail)
admin.site.register(UncategorisedDicom)
admin.site.register(SeriesDetail)
admin.site.register(Resource)
admin.site.register(CaseDisplaySet)
admin.site.register(Procedure)
admin.site.register(UserReportAnswer)
admin.site.register(CidReportAnswer)


class DifferentialInline(admin.TabularInline):
    model = Differential

class CasesInline(admin.TabularInline):
    model = CaseDetail

class CaseCollectionAdminForm(ModelForm):
    class Meta:
        model = CaseCollection

        exclude = []


class CaseCollectionAdmin(VersionAdmin):
    form = CaseCollectionAdminForm

    save_on_top = True
    save_as = True
    view_on_site = True

    inlines = (CasesInline,)


admin.site.register(CaseCollection, CaseCollectionAdmin)


class NormalCaseAdmin(VersionAdmin):
    list_display = ("case", "display_age", "examination", "modality", "added_date")
    list_filter = ("examination", "modality")
    search_fields = ("case__title", "case__pk")

    def display_age(self, obj):
        try:
            return obj.display_age()
        except Exception:
            return ""
    display_age.short_description = "Age"


admin.site.register(NormalCase, NormalCaseAdmin)

class AtlasAdminForm(ModelForm):
    class Meta:
        model = Case

        fields = [
            "title",
            "description",
            "history",
            "discussion",
            "condition",
            "differential",
            "subspecialty",
            "author",
            "series",
            "cimar_uuid"
        ]

        widgets = {
            # "normal": RadioSelect(
            # choices=[(True, 'Yes'),
            #     (False, 'No')])
            "history": TinyMCE(attrs={"cols": 80, "rows": 30}),
            "discussion": TinyMCE(attrs={"cols": 80, "rows": 30}),
        }


class AtlasAdmin(VersionAdmin):
    form = AtlasAdminForm

    save_on_top = True
    save_as = True
    view_on_site = True

    inlines = (DifferentialInline,)


admin.site.register(Case, AtlasAdmin)


class SeriesAdmin(VersionAdmin):
    pass


admin.site.register(Series, SeriesAdmin)


class APITokenAdmin(admin.ModelAdmin):
    list_display = ("user", "name", "revoked", "created", "expires", "last_used")
    search_fields = ("user__username", "name")
    readonly_fields = ("token_hash", "created", "last_used")


admin.site.register(APIToken, APITokenAdmin)


if DBTaskResult is not None:
    try:
        # Ensure package admin registration has run before we customize it.
        import django_tasks_db.admin  # noqa: F401
    except Exception:
        pass

    def traceback_link(self, obj):
        if not obj.traceback:
            return "-"
        url = reverse(
            f"admin:{obj._meta.app_label}_{obj._meta.model_name}_change",
            args=[obj.pk],
        )
        return format_html('<a href="{}">View full traceback</a>', url)

    traceback_link.short_description = "Traceback"

    def traceback_pretty(self, obj):
        if not obj.traceback:
            return ""
        return format_html(
            '<pre style="white-space: pre-wrap; margin: 0;">{}</pre>',
            obj.traceback,
        )

    traceback_pretty.short_description = "Traceback"

    @admin.action(description="Reset FAILED tasks to READY")
    def reset_failed_to_ready(self, request, queryset):
        failed_qs = queryset.filter(status="FAILED")
        updated = failed_qs.update(
            status="READY",
            started_at=None,
            finished_at=None,
            worker_ids="",
            exception_class_path="",
            traceback="",
            return_value=None,
        )
        self.message_user(
            request,
            f"Reset {updated} failed task(s) to READY.",
            level=messages.SUCCESS,
        )

    @admin.action(description="Reset selected tasks to READY (except RUNNING)")
    def reset_selected_to_ready(self, request, queryset):
        blocked = queryset.filter(status="RUNNING").count()
        reset_qs = queryset.exclude(status="RUNNING")
        updated = reset_qs.update(
            status="READY",
            started_at=None,
            finished_at=None,
            worker_ids="",
            exception_class_path="",
            traceback="",
            return_value=None,
        )
        msg = f"Reset {updated} task(s) to READY."
        if blocked:
            msg += f" Skipped {blocked} running task(s)."
        self.message_user(request, msg, level=messages.SUCCESS)

    @admin.action(description="Delete selected COMPLETED tasks")
    def delete_selected_completed(self, request, queryset):
        completed_qs = queryset.filter(status="SUCCESSFUL")
        deleted, _ = completed_qs.delete()
        self.message_user(
            request,
            f"Deleted {deleted} completed task record(s).",
            level=messages.SUCCESS,
        )

    @admin.action(description="Delete selected FAILED tasks")
    def delete_selected_failed(self, request, queryset):
        failed_qs = queryset.filter(status="FAILED")
        deleted, _ = failed_qs.delete()
        self.message_user(
            request,
            f"Deleted {deleted} failed task record(s).",
            level=messages.SUCCESS,
        )

    def has_add_permission(self, request):
        return False

    task_admin = admin.site._registry.get(DBTaskResult)
    if task_admin is not None:
        admin_cls = task_admin.__class__
        admin_cls.list_display = (
            "id",
            "task_path",
            "status",
            "queue_name",
            "priority",
            "enqueued_at",
            "started_at",
            "finished_at",
            "worker_ids",
            "exception_class_path",
            "traceback_link",
        )
        admin_cls.list_filter = ("status", "queue_name", "task_path", "backend_name")
        admin_cls.search_fields = (
            "id",
            "task_path",
            "queue_name",
            "worker_ids",
            "exception_class_path",
            "traceback",
        )
        admin_cls.ordering = ("-enqueued_at",)
        admin_cls.actions = (
            "reset_failed_to_ready",
            "reset_selected_to_ready",
            "delete_selected_completed",
            "delete_selected_failed",
        )
        admin_cls.readonly_fields = (
            "id",
            "status",
            "enqueued_at",
            "started_at",
            "finished_at",
            "args_kwargs",
            "priority",
            "task_path",
            "worker_ids",
            "queue_name",
            "backend_name",
            "run_after",
            "return_value",
            "exception_class_path",
            "traceback_pretty",
        )
        admin_cls.fieldsets = (
            (
                "Task",
                {
                    "fields": (
                        "id",
                        "task_path",
                        "status",
                        "queue_name",
                        "priority",
                        "backend_name",
                    )
                },
            ),
            (
                "Timing",
                {
                    "fields": (
                        "enqueued_at",
                        "run_after",
                        "started_at",
                        "finished_at",
                        "worker_ids",
                    )
                },
            ),
            (
                "Payload",
                {"fields": ("args_kwargs", "return_value")},
            ),
            (
                "Error",
                {
                    "fields": ("exception_class_path", "traceback_pretty"),
                },
            ),
        )
        admin_cls.traceback_link = traceback_link
        admin_cls.traceback_pretty = traceback_pretty
        admin_cls.reset_failed_to_ready = reset_failed_to_ready
        admin_cls.reset_selected_to_ready = reset_selected_to_ready
        admin_cls.delete_selected_completed = delete_selected_completed
        admin_cls.delete_selected_failed = delete_selected_failed
        admin_cls.has_add_permission = has_add_permission