angrybeanie_wagtail/env/lib/python3.12/site-packages/wagtail/admin/views/home.py

from collections.abc import Mapping
from typing import Any, Union

from django.conf import settings
from django.contrib.auth import get_user_model
from django.contrib.auth.decorators import permission_required
from django.db.models import Exists, IntegerField, Max, OuterRef, Q
from django.db.models.functions import Cast
from django.forms import Media
from django.http import Http404, HttpResponse
from django.utils.translation import gettext_lazy as _
from django.views.generic.base import TemplateView

from wagtail import hooks
from wagtail.admin.forms.search import SearchForm
from wagtail.admin.icons import get_icons
from wagtail.admin.navigation import get_site_for_user
from wagtail.admin.site_summary import SiteSummaryPanel
from wagtail.admin.ui.components import Component
from wagtail.admin.views.generic import WagtailAdminTemplateMixin
from wagtail.models import (
    Page,
    PageLogEntry,
    Revision,
    TaskState,
    WorkflowState,
    get_default_page_content_type,
)
from wagtail.permissions import page_permission_policy

User = get_user_model()


# Panels for the homepage


class UpgradeNotificationPanel(Component):
    template_name = "wagtailadmin/home/upgrade_notification.html"
    dismissible_id = "last_upgrade_check"

    def get_upgrade_check_setting(self) -> Union[bool, str]:
        return getattr(settings, "WAGTAIL_ENABLE_UPDATE_CHECK", True)

    def upgrade_check_lts_only(self) -> bool:
        upgrade_check = self.get_upgrade_check_setting()
        if isinstance(upgrade_check, str) and upgrade_check.lower() == "lts":
            return True
        return False

    def get_dismissible_value(self, user) -> str:
        if profile := getattr(user, "wagtail_userprofile", None):
            return profile.dismissibles.get(self.dismissible_id)
        return None

    def get_context_data(self, parent_context: Mapping[str, Any]) -> Mapping[str, Any]:
        return {
            "lts_only": self.upgrade_check_lts_only(),
            "dismissible_id": self.dismissible_id,
            "dismissible_value": self.get_dismissible_value(
                parent_context["request"].user
            ),
        }

    def render_html(self, parent_context: Mapping[str, Any] = None) -> str:
        if (
            parent_context["request"].user.is_superuser
            and self.get_upgrade_check_setting()
        ):
            return super().render_html(parent_context)
        else:
            return ""


class WhatsNewInWagtailVersionPanel(Component):
    name = "whats_new_in_wagtail_version"
    template_name = "wagtailadmin/home/whats_new_in_wagtail_version.html"
    order = 110
    _version = "4"

    def get_whats_new_banner_setting(self) -> Union[bool, str]:
        return getattr(settings, "WAGTAIL_ENABLE_WHATS_NEW_BANNER", True)

    def get_dismissible_id(self) -> str:
        return f"{self.name}_{self._version}"

    def get_context_data(self, parent_context: Mapping[str, Any]) -> Mapping[str, Any]:
        return {"dismissible_id": self.get_dismissible_id(), "version": self._version}

    def is_shown(self, parent_context: Mapping[str, Any] = None) -> bool:
        if not self.get_whats_new_banner_setting():
            return False

        profile = getattr(parent_context["request"].user, "wagtail_userprofile", None)
        if profile and profile.dismissibles.get(self.get_dismissible_id()):
            return False

        return True

    def render_html(self, parent_context: Mapping[str, Any] = None) -> str:
        if not self.is_shown(parent_context):
            return ""
        return super().render_html(parent_context)


class UserObjectsInWorkflowModerationPanel(Component):
    name = "user_objects_in_workflow_moderation"
    template_name = "wagtailadmin/home/user_objects_in_workflow_moderation.html"
    order = 210

    def get_context_data(self, parent_context):
        request = parent_context["request"]
        context = super().get_context_data(parent_context)
        if getattr(settings, "WAGTAIL_WORKFLOW_ENABLED", True):
            # Need to cast the page ids to string because Postgres doesn't support
            # implicit type casts when querying on GenericRelations. We also need
            # to cast the object_id to integer when querying the pages for the same reason.
            # https://code.djangoproject.com/ticket/16055
            # Once the issue is resolved, this query can be removed and the
            # filter can be changed to:
            # Q(page__owner=request.user) | Q(requested_by=request.user)
            pages_owned_by_user = Q(
                base_content_type_id=get_default_page_content_type().id
            ) & Exists(
                Page.objects.filter(
                    owner=request.user,
                    id=Cast(OuterRef("object_id"), output_field=IntegerField()),
                )
            )
            # Find in progress workflow states which are either requested by the user or on pages owned by the user
            context["workflow_states"] = (
                WorkflowState.objects.active()
                .filter(pages_owned_by_user | Q(requested_by=request.user))
                .prefetch_related(
                    "content_object",
                    "content_object__latest_revision",
                )
                .select_related(
                    "current_task_state",
                    "current_task_state__task",
                )
                .order_by("-current_task_state__started_at")
            )
            # Filter out workflow states where the GenericForeignKey points to
            # a nonexistent object. This can happen if the model does not define
            # a GenericRelation to WorkflowState and the instance is deleted.
            context["workflow_states"] = [
                state for state in context["workflow_states"] if state.content_object
            ]
        else:
            context["workflow_states"] = WorkflowState.objects.none()
        context["request"] = request
        return context


class WorkflowObjectsToModeratePanel(Component):
    name = "workflow_objects_to_moderate"
    template_name = "wagtailadmin/home/workflow_objects_to_moderate.html"
    order = 220

    def get_context_data(self, parent_context):
        request = parent_context["request"]
        context = super().get_context_data(parent_context)
        context["states"] = []
        context["request"] = request
        context["csrf_token"] = parent_context["csrf_token"]

        if not getattr(settings, "WAGTAIL_WORKFLOW_ENABLED", True):
            return context

        states = (
            TaskState.objects.reviewable_by(request.user)
            .select_related(
                "revision",
                "revision__user",
                "workflow_state",
                "workflow_state__workflow",
            )
            .prefetch_related(
                "revision__content_object",
                "revision__content_object__latest_revision",
            )
            .order_by("-started_at")
            .annotate(
                previous_revision_id=Revision.objects.previous_revision_id_subquery(),
            )
        )

        for state in states:
            obj = state.revision.content_object
            # Skip task states where the revision's GenericForeignKey points to
            # a nonexistent object. This can happen if the model does not define
            # a GenericRelation to WorkflowState and/or Revision and the instance
            # is deleted.
            if not obj:
                continue
            actions = state.task.specific.get_actions(obj, request.user)
            workflow_tasks = state.workflow_state.all_tasks_with_status()

            workflow_action_url_name = "wagtailadmin_pages:workflow_action"
            workflow_preview_url_name = "wagtailadmin_pages:workflow_preview"
            revisions_compare_url_name = "wagtailadmin_pages:revisions_compare"

            # Snippets can also have workflows
            if not isinstance(obj, Page):
                viewset = obj.snippet_viewset
                workflow_action_url_name = viewset.get_url_name("workflow_action")
                workflow_preview_url_name = viewset.get_url_name("workflow_preview")
                revisions_compare_url_name = viewset.get_url_name("revisions_compare")

            if not getattr(obj, "is_previewable", False):
                workflow_preview_url_name = None

            context["states"].append(
                {
                    "obj": obj,
                    "revision": state.revision,
                    "previous_revision_id": state.previous_revision_id,
                    "live_revision_id": obj.live_revision_id,
                    "task_state": state,
                    "actions": actions,
                    "workflow_tasks": workflow_tasks,
                    "workflow_action_url_name": workflow_action_url_name,
                    "workflow_preview_url_name": workflow_preview_url_name,
                    "revisions_compare_url_name": revisions_compare_url_name,
                }
            )

        return context


class LockedPagesPanel(Component):
    name = "locked_pages"
    template_name = "wagtailadmin/home/locked_pages.html"
    order = 300

    def get_context_data(self, parent_context):
        request = parent_context["request"]
        context = super().get_context_data(parent_context)
        context.update(
            {
                "locked_pages": Page.objects.filter(
                    locked=True,
                    locked_by=request.user,
                )
                .order_by("-locked_at", "-latest_revision_created_at", "-pk")
                .specific(defer=True),
                "can_remove_locks": page_permission_policy.user_has_permission(
                    request.user, "unlock"
                ),
                "request": request,
                "csrf_token": parent_context["csrf_token"],
            }
        )
        return context


class RecentEditsPanel(Component):
    name = "recent_edits"
    template_name = "wagtailadmin/home/recent_edits.html"
    order = 250

    def get_context_data(self, parent_context):
        request = parent_context["request"]
        context = super().get_context_data(parent_context)

        # Last n edited pages
        edit_count = getattr(settings, "WAGTAILADMIN_RECENT_EDITS_LIMIT", 5)

        # Query the audit log to get a resultset of (page ID, latest edit timestamp)
        last_edits_dates = (
            PageLogEntry.objects.filter(user=request.user, action="wagtail.edit")
            .values("page_id")
            .annotate(latest_date=Max("timestamp"))
            .order_by("-latest_date")[:edit_count]
        )
        # Retrieve the page objects for those IDs
        pages_mapping = (
            Page.objects.specific()
            .prefetch_workflow_states()
            .annotate_approved_schedule()
            .in_bulk([log["page_id"] for log in last_edits_dates])
        )
        # Compile a list of (latest edit timestamp, page object) tuples
        last_edits = []
        for log in last_edits_dates:
            page = pages_mapping.get(log["page_id"])
            if page:
                last_edits.append((log["latest_date"], page))

        context["last_edits"] = last_edits
        context["request"] = request
        return context


class HomeView(WagtailAdminTemplateMixin, TemplateView):
    template_name = "wagtailadmin/home.html"
    page_title = _("Dashboard")
    permission_policy = page_permission_policy

    def get_context_data(self, **kwargs):
        context = super().get_context_data(**kwargs)
        panels = self.get_panels()
        site_summary = SiteSummaryPanel(self.request)
        site_details = self.get_site_details()

        context["media"] = self.get_media([*panels, site_summary])
        context["panels"] = sorted(panels, key=lambda p: p.order)
        context["site_summary"] = site_summary
        context["upgrade_notification"] = UpgradeNotificationPanel()
        context["search_form"] = SearchForm(placeholder=_("Search all pages…"))
        context["user"] = self.request.user

        return {**context, **site_details}

    def get_media(self, panels=[]):
        media = Media()

        for panel in panels:
            media += panel.media

        return media

    def get_panels(self):
        request = self.request
        panels = [
            # Disabled until a release warrants the banner.
            # WhatsNewInWagtailVersionPanel(),
            WorkflowObjectsToModeratePanel(),
            UserObjectsInWorkflowModerationPanel(),
            RecentEditsPanel(),
            LockedPagesPanel(),
        ]

        for fn in hooks.get_hooks("construct_homepage_panels"):
            fn(request, panels)

        return panels

    def get_site_details(self):
        request = self.request
        site = get_site_for_user(request.user)

        return {
            "root_page": site["root_page"],
            "root_site": site["root_site"],
            "site_name": site["site_name"],
        }


def error_test(request):
    raise Exception("This is a test of the emergency broadcast system.")


@permission_required("wagtailadmin.access_admin", login_url="wagtailadmin_login")
def default(request):
    """
    Called whenever a request comes in with the correct prefix (eg /admin/) but
    doesn't actually correspond to a Wagtail view.

    For authenticated users, it'll raise a 404 error. Anonymous users will be
    redirected to the login page.
    """
    raise Http404


def sprite(request):
    return HttpResponse(get_icons(), content_type="image/svg+xml; charset=utf-8")