-
+
{% if page.live %}
Published: Yes
{% else %}
@@ -49,7 +49,7 @@
{% trans "Are you sure you want to publish this page and all of its children?" %}
-
+
{% trans "No, don't publish." %}
diff --git a/wagtail_to_ion/utils.py b/wagtail_to_ion/utils.py
index 54eb7bc..a995a57 100644
--- a/wagtail_to_ion/utils.py
+++ b/wagtail_to_ion/utils.py
@@ -1,6 +1,11 @@
-from django.db.models import Q
+import functools
+from typing import Any, Dict, Generator, List, NamedTuple, Tuple, Type, Union
-from wagtail.core.models import Collection, Page, PageViewRestriction
+from django.db.models import Q, Model
+
+from wagtail.core.blocks import Block, BoundBlock, StreamValue, StructBlock, StructValue
+from wagtail.core.fields import StreamField
+from wagtail.core.models import Collection, Page, PageViewRestriction, get_page_models
from wagtail.images import get_image_model
from wagtail.documents import get_document_model
@@ -61,7 +66,7 @@ def get_collection_for_page(page):
# TODO: might be obsolete once https://github.com/wagtail/wagtail/pull/6300 has been merged
def visible_tree_by_user(root, user):
collection = IonCollection.objects.get(live=True, slug=get_collection_for_page(root))
-
+
if collection.view_restrictions.exists():
restrictions = collection.view_restrictions.filter(
restriction_type=PageViewRestriction.GROUPS,
@@ -115,3 +120,110 @@ def isoDate(d):
return d.replace(microsecond=0, tzinfo=None).strftime("%Y-%m-%dT%H:%M:%SZ")
else:
return 'None'
+
+
+def get_object_block_usage(obj, block_types: Union[Type[Block], Tuple[Type[Block]]]):
+ """
+ Returns a queryset of pages that contain a block linked to a particular object.
+
+ Works like `wagtail.admin.models.get_object_usage` but inspects all pages
+ which might contain a block of the specified type(s).
+
+ To avoid inspecting all pages the following optimizations are applied:
+ 1. operate only on page models with a StreamField containing the specified block type(s)
+ 2. filter pages by expected JSON string in StreamField column(s)
+ """
+ block_usage = get_page_models_using_blocks(block_types=block_types)
+ page_ptr_ids = set()
+
+ for page_model in block_usage.keys():
+ stream_field_filter_q = Q()
+ stream_fields = set()
+
+ # create a (postgres specific) filter for every block; look for:
+ # - `"value":
` string if block is in a StreamValue
+ # - `"": ` if block is in a StructValue
+ for block in block_usage[page_model]:
+ filter_att = block.block_name if block.in_struct else 'value'
+ stream_field_filter_q |= Q(**{f'{block.stream_field_name}__regex': rf'"{filter_att}":\s*{obj.pk}\M'})
+ stream_fields.add(block.stream_field_name)
+
+ for page_with_obj_pk_in_blocks in page_model.objects.filter(stream_field_filter_q):
+ for field_name in stream_fields:
+ for bound_block, _ in get_stream_value_blocks(getattr(page_with_obj_pk_in_blocks, field_name)):
+ if isinstance(bound_block.block, block_types) and bound_block.value == obj:
+ page_ptr_ids.add(page_with_obj_pk_in_blocks.page_ptr_id)
+
+ return Page.objects.filter(pk__in=page_ptr_ids)
+
+
+class StreamFieldBlockInfo(NamedTuple):
+ block_name: str
+ block_type: Block
+ in_struct: bool
+
+
+def get_stream_field_blocks(stream_field) -> Generator[StreamFieldBlockInfo, None, None]:
+ """Generates an un-nested list of blocks of a `StreamField`."""
+ def unnest_blocks(blocks: dict, in_struct: bool = False):
+ for block in blocks.items():
+ yield StreamFieldBlockInfo(block[0], block[1], in_struct)
+ if hasattr(block[1], 'child_blocks'):
+ yield from unnest_blocks(block[1].child_blocks, in_struct=isinstance(block[1], StructBlock))
+
+ return unnest_blocks(stream_field.stream_block.child_blocks)
+
+
+class StreamValueBlockInfo(NamedTuple):
+ bound_block: BoundBlock
+ json_value_field_name: str
+
+
+def get_stream_value_blocks(stream_value: StreamValue) -> Generator[StreamValueBlockInfo, None, None]:
+ """Generates an un-nested list of blocks of a `StreamValue`."""
+ def unnest_blocks(value: Union[StreamValue, StructValue]):
+ if isinstance(value, StreamValue):
+ for stream_block in value:
+ assert isinstance(stream_block, BoundBlock)
+ if isinstance(stream_block.value, StructValue):
+ yield from unnest_blocks(stream_block.value)
+ else:
+ yield StreamValueBlockInfo(stream_block, 'value')
+ elif isinstance(value, StructValue):
+ for block_name, bound_block in value.bound_blocks.items():
+ assert isinstance(bound_block, BoundBlock)
+ if isinstance(bound_block.value, StreamValue):
+ yield from unnest_blocks(bound_block.value)
+ else:
+ yield StreamValueBlockInfo(bound_block, block_name)
+ else:
+ raise RuntimeError(f'Unexpected type: {type(value)}')
+
+ return unnest_blocks(stream_value)
+
+
+class ModelStreamFieldBlockInfo(NamedTuple):
+ stream_field_name: str
+ block_name: str
+ block_type: Block
+ in_struct: bool
+
+
+@functools.lru_cache()
+def get_page_models_using_blocks(
+ block_types: Union[Type[Block], Tuple[Type[Block]]],
+) -> Dict[Type[Model], List[ModelStreamFieldBlockInfo]]:
+ """Returns information about all page models which use the specified block type(s) in a StreamField."""
+ models_with_block: Dict[Type[Model], List[ModelStreamFieldBlockInfo]] = {}
+
+ for page_model in get_page_models():
+ for stream_field in [field for field in page_model._meta.get_fields() if isinstance(field, StreamField)]:
+ for stream_field_block in get_stream_field_blocks(stream_field):
+ if isinstance(stream_field_block.block_type, block_types):
+ if page_model not in models_with_block:
+ models_with_block[page_model] = []
+ models_with_block[page_model].append(
+ ModelStreamFieldBlockInfo(stream_field.attname, *stream_field_block),
+ )
+
+ return models_with_block
diff --git a/wagtail_to_ion/views/wagtail_override/__init__.py b/wagtail_to_ion/views/wagtail_override/__init__.py
index fdabaa0..f12ae1e 100644
--- a/wagtail_to_ion/views/wagtail_override/__init__.py
+++ b/wagtail_to_ion/views/wagtail_override/__init__.py
@@ -1 +1,2 @@
from .pages import ion_unpublish, ion_publish_with_children
+from .file_based_models import document_safe_delete, image_safe_delete, media_safe_delete
diff --git a/wagtail_to_ion/views/wagtail_override/file_based_models.py b/wagtail_to_ion/views/wagtail_override/file_based_models.py
new file mode 100644
index 0000000..8a6bd59
--- /dev/null
+++ b/wagtail_to_ion/views/wagtail_override/file_based_models.py
@@ -0,0 +1,48 @@
+from django.shortcuts import get_object_or_404, redirect
+from django.utils.translation import gettext_lazy
+
+from wagtail.admin import messages
+from wagtail.admin.auth import PermissionPolicyChecker
+from wagtail.documents import get_document_model
+from wagtail.documents.permissions import permission_policy as document_permission_policy
+from wagtail.documents.views.documents import delete as document_delete_view
+from wagtail.images import get_image_model
+from wagtail.images.permissions import permission_policy as image_permission_policy
+from wagtail.images.views.images import delete as image_delete_view
+from wagtailmedia.models import get_media_model
+from wagtailmedia.permissions import permission_policy as media_permission_policy
+from wagtailmedia.views.media import delete as media_delete_view
+
+
+document_permission_checker = PermissionPolicyChecker(document_permission_policy)
+image_permission_checker = PermissionPolicyChecker(image_permission_policy)
+media_permission_checker = PermissionPolicyChecker(media_permission_policy)
+
+MESSAGE = gettext_lazy('Cannot delete object if used by any page')
+
+
+@image_permission_checker.require('delete')
+def image_safe_delete(request, image_id):
+ image = get_object_or_404(get_image_model(), pk=image_id)
+ if image.get_usage().exists():
+ messages.error(request, MESSAGE)
+ return redirect('wagtailimages:edit', image_id=image_id)
+ return image_delete_view(request, image_id)
+
+
+@document_permission_checker.require('delete')
+def document_safe_delete(request, document_id):
+ document = get_object_or_404(get_document_model(), pk=document_id)
+ if document.get_usage().exists():
+ messages.error(request, MESSAGE)
+ return redirect('wagtaildocs:edit', document_id=document_id)
+ return document_delete_view(request, document_id)
+
+
+@media_permission_checker.require('delete')
+def media_safe_delete(request, media_id):
+ media = get_object_or_404(get_media_model(), pk=media_id)
+ if media.get_usage().exists():
+ messages.error(request, MESSAGE)
+ return redirect('wagtailmedia:edit', media_id)
+ return media_delete_view(request, media_id)
diff --git a/wagtail_to_ion/views/wagtail_override/pages.py b/wagtail_to_ion/views/wagtail_override/pages.py
index 34aaca6..3529d70 100644
--- a/wagtail_to_ion/views/wagtail_override/pages.py
+++ b/wagtail_to_ion/views/wagtail_override/pages.py
@@ -1,14 +1,16 @@
# Copyright © 2017 anfema GmbH. All rights reserved.
from django.shortcuts import get_object_or_404, redirect
from django.template.response import TemplateResponse
+from django.utils.text import slugify
from django.utils.translation import ugettext as _
from django.urls import reverse
from django.core.exceptions import PermissionDenied
from wagtail.core import hooks
from wagtail.admin import messages
+from wagtail.admin.forms.pages import CopyForm
from wagtail.admin.widgets import Button
-from wagtail.admin.auth import PermissionPolicyChecker
+from wagtail.admin.auth import PermissionPolicyChecker, user_has_any_page_permission, user_passes_test
from wagtail.images.permissions import permission_policy
from wagtail.core.models import Page
from wagtail.admin.views.pages.utils import get_valid_next_url_from_request
@@ -19,7 +21,7 @@
# add publish button to dropdown menu
@hooks.register('register_page_listing_more_buttons')
-def page_listing_more_buttons(page, page_perms, is_parent=False):
+def page_listing_more_buttons(page, page_perms, is_parent=False, next_url=None):
if page_perms.can_publish():
yield Button(
'Publish with children',
@@ -100,16 +102,23 @@ def ion_unpublish(request, page_id):
def ion_publish_with_children(request, page_id):
page = get_object_or_404(Page, id=page_id)
page_perms = page.permissions_for_user(request.user)
-
+
if not page_perms.can_publish():
raise PermissionDenied
revision = page.get_latest_revision()
- unpublished_descendant_pages = page.get_descendants().filter(live=False).specific()
+ unpublished_descendant_pages = page.get_descendants().not_live()
+ has_unpublished_alias_pages = unpublished_descendant_pages.filter(alias_of__isnull=False).exists()
+
+ if has_unpublished_alias_pages:
+ messages.error(request, "The page has unpublished alias subpages. Please publish the original page(s) first.")
next_url = get_valid_next_url_from_request(request)
if request.method == 'POST':
+ if has_unpublished_alias_pages:
+ return redirect(next_url) if next_url else redirect('wagtailadmin_explore', page.get_parent().id)
+
revision.publish()
for unpublished_descendant_page in unpublished_descendant_pages:
@@ -126,5 +135,134 @@ def ion_publish_with_children(request, page_id):
'page': page,
'next': next_url,
'unpublished_descendants': unpublished_descendant_pages,
- 'unpublished_descendant_count': page.get_descendants().not_live().count(),
+ 'unpublished_descendant_count': unpublished_descendant_pages.count(),
+ 'has_unpublished_alias_pages': has_unpublished_alias_pages,
})
+
+
+class AutoTitleCopyForm(CopyForm):
+ """
+ Copy form for pages with auto-generated title & slug fields.
+
+ The `new_title` & `new_slug` fields are marked as optional and can be omitted from the form.
+ """
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ for omitted_field in ('new_title', 'new_slug'):
+ self.fields[omitted_field].required = False
+ self.fields[omitted_field].initial = ''
+
+ def clean(self):
+ # generate `new_title` & `new_slug`:
+ # only required for copy as alias functionality; new title & slug for standard copy operations are handled
+ # in `AbstractIonPage.copy()`
+ new_title = self.page.specific_class.generate_page_title()
+ return {
+ **super().clean(),
+ 'new_title': new_title,
+ 'new_slug': slugify(new_title),
+ }
+
+
+# Copy of the wagtail `wagtail.admin.views.pages.copy.copy()` view
+#
+# Adjustments:
+# - `CopyForm` replaced by `AutoTitleCopyForm`
+# - disabled `before_copy_page` hook handling
+# - replaced template
+# - fix crash when alias field is not available in form:
+# replaced `form.cleaned_data['alias']` with `form.cleaned_data.get('alias')`
+#
+@user_passes_test(user_has_any_page_permission)
+def ion_copy_auto_title(request, page_id):
+ page = Page.objects.get(id=page_id)
+
+ # Parent page defaults to parent of source page
+ parent_page = page.get_parent()
+
+ # Check if the user has permission to publish subpages on the parent
+ can_publish = parent_page.permissions_for_user(request.user).can_publish_subpage()
+
+ # Create the form
+ form = AutoTitleCopyForm(request.POST or None, user=request.user, page=page, can_publish=can_publish)
+
+ next_url = get_valid_next_url_from_request(request)
+
+ # for fn in hooks.get_hooks('before_copy_page'):
+ # result = fn(request, page)
+ # if hasattr(result, 'status_code'):
+ # return result
+
+ # Check if user is submitting
+ if request.method == 'POST':
+ # Prefill parent_page in case the form is invalid (as prepopulated value for the form field,
+ # because ModelChoiceField seems to not fall back to the user given value)
+ parent_page = Page.objects.get(id=request.POST['new_parent_page'])
+
+ if form.is_valid():
+ # Receive the parent page (this should never be empty)
+ if form.cleaned_data['new_parent_page']:
+ parent_page = form.cleaned_data['new_parent_page']
+
+ if not page.permissions_for_user(request.user).can_copy_to(parent_page,
+ form.cleaned_data.get('copy_subpages')):
+ raise PermissionDenied
+
+ # Re-check if the user has permission to publish subpages on the new parent
+ can_publish = parent_page.permissions_for_user(request.user).can_publish_subpage()
+ keep_live = can_publish and form.cleaned_data.get('publish_copies')
+
+ # Copy the page
+ # Note that only users who can publish in the new parent page can create an alias.
+ # This is because alias pages must always match their original page's state.
+ if can_publish and form.cleaned_data.get('alias'):
+ new_page = page.specific.create_alias(
+ recursive=form.cleaned_data.get('copy_subpages'),
+ parent=parent_page,
+ update_slug=form.cleaned_data['new_slug'],
+ user=request.user,
+ )
+ else:
+ new_page = page.specific.copy(
+ recursive=form.cleaned_data.get('copy_subpages'),
+ to=parent_page,
+ update_attrs={
+ 'title': form.cleaned_data['new_title'],
+ 'slug': form.cleaned_data['new_slug'],
+ },
+ keep_live=keep_live,
+ user=request.user,
+ )
+
+ # Give a success message back to the user
+ if form.cleaned_data.get('copy_subpages'):
+ messages.success(
+ request,
+ _("Page '{0}' and {1} subpages copied.").format(page.get_admin_display_title(),
+ new_page.get_descendants().count())
+ )
+ else:
+ messages.success(request, _("Page '{0}' copied.").format(page.get_admin_display_title()))
+
+ for fn in hooks.get_hooks('after_copy_page'):
+ result = fn(request, page, new_page)
+ if hasattr(result, 'status_code'):
+ return result
+
+ # Redirect to explore of parent page
+ if next_url:
+ return redirect(next_url)
+ return redirect('wagtailadmin_explore', parent_page.id)
+
+ return TemplateResponse(request, 'wagtailadmin/pages/copy_auto_title.html', {
+ 'page': page,
+ 'form': form,
+ 'next': next_url,
+ })
+
+
+@hooks.register('before_copy_page')
+def use_auto_title_copy_form(request, page):
+ if getattr(page.specific_class, 'ion_generate_page_title', None):
+ return ion_copy_auto_title(request, page.pk)
diff --git a/wagtail_to_ion/wagtail_hooks/__init__.py b/wagtail_to_ion/wagtail_hooks/__init__.py
index 1a62fdd..3dcccac 100644
--- a/wagtail_to_ion/wagtail_hooks/__init__.py
+++ b/wagtail_to_ion/wagtail_hooks/__init__.py
@@ -1,3 +1,4 @@
from .show_only_user_images import show_only_user_images
from .show_only_user_documents import show_only_user_documents
from .dual_range_slider import global_admin_css, global_admin_js
+from .wagtail_admin_urls import register_admin_urls
diff --git a/wagtail_to_ion/wagtail_hooks/wagtail_admin_urls.py b/wagtail_to_ion/wagtail_hooks/wagtail_admin_urls.py
new file mode 100644
index 0000000..da16764
--- /dev/null
+++ b/wagtail_to_ion/wagtail_hooks/wagtail_admin_urls.py
@@ -0,0 +1,14 @@
+from django.urls import path
+
+from wagtail.core import hooks
+
+from wagtail_to_ion.views.wagtail_override import image_safe_delete, document_safe_delete, media_safe_delete
+
+
+@hooks.register('register_admin_urls')
+def register_admin_urls():
+ return [
+ path('documents/delete//', document_safe_delete, name='documents-safe-delete'),
+ path('images//delete/', image_safe_delete, name='images-safe-delete'),
+ path('media/delete//', media_safe_delete, name='media-safe-delete'),
+ ]