From b11078fd92cbe697372f77eca579119b3155f099 Mon Sep 17 00:00:00 2001 From: shamoon <4887959+shamoon@users.noreply.github.com> Date: Fri, 13 Nov 2026 13:11:42 -0800 Subject: [PATCH] Move the other bulk edit actions through the new taks tracking route --- src/documents/bulk_edit.py | 4 +- src/documents/bulk_operations.py | 67 ++++++++++++++++++++- src/documents/signals/handlers.py | 2 + src/documents/views.py | 96 ++++++++----------------------- 4 files changed, 93 insertions(+), 76 deletions(-) diff --git a/src/documents/bulk_edit.py b/src/documents/bulk_edit.py index c313d52e23..ea4bd04d86 100644 --- a/src/documents/bulk_edit.py +++ b/src/documents/bulk_edit.py @@ -475,9 +475,9 @@ def run_workflow( """ Manually run a single workflow against the given documents. - Done asynchronously via a Celery task. + Called from the tracked bulk operation task. """ - run_workflows_for_documents.delay( + run_workflows_for_documents( document_ids=doc_ids, workflow_id=workflow, filter_documents=filter_documents, diff --git a/src/documents/bulk_operations.py b/src/documents/bulk_operations.py index eee7f8b077..57cc55f862 100644 --- a/src/documents/bulk_operations.py +++ b/src/documents/bulk_operations.py @@ -1,4 +1,4 @@ -"""Queued document metadata edits.""" +"""Queued document bulk operations.""" from typing import Any @@ -12,6 +12,7 @@ from rest_framework.exceptions import PermissionDenied from documents import bulk_edit from documents.models import CustomFieldInstance from documents.models import Document +from documents.models import PaperlessTask from documents.permissions import has_document_operation_permissions from documents.tasks import bulk_update_documents @@ -25,7 +26,35 @@ MODIFIED_FIELDS = { "modify_custom_fields": "custom_fields", "set_permissions": None, } -QUEUED_OPERATIONS = frozenset(MODIFIED_FIELDS) +LEGACY_MODIFIED_FIELDS = {"delete": "deleted_at", "reprocess": "checksum"} +QUEUED_OPERATIONS = frozenset( + ( + *MODIFIED_FIELDS, + "delete", + "reprocess", + "run_workflow", + "rotate", + "merge", + "merge_as_versions", + "split", + "delete_pages", + "edit_pdf", + "remove_password", + ), +) +USER_OPERATIONS = frozenset( + { + "split", + "merge", + "rotate", + "delete_pages", + "edit_pdf", + "remove_password", + "merge_as_versions", + }, +) +# merge_as_versions doesn't queue any consume tasks +TRIGGER_SOURCE_OPERATIONS = USER_OPERATIONS - {"merge_as_versions"} def snapshot_field(doc_ids: list[int], field: str) -> dict[int, Any]: @@ -66,6 +95,7 @@ def execute_bulk_operation( document_ids: list[int], parameters: dict, user_id: int, + trigger_source: str = PaperlessTask.TriggerSource.API_UPLOAD, ) -> dict: if operation not in QUEUED_OPERATIONS: raise ValueError(f"Unsupported bulk operation: {operation}") @@ -79,6 +109,39 @@ def execute_bulk_operation( ): raise PermissionDenied("Insufficient permissions to execute bulk operation") + if operation == "run_workflow" and not user.has_perm("documents.view_workflow"): + raise PermissionDenied("Insufficient permissions to run this workflow") + + if operation not in MODIFIED_FIELDS: + execution_parameters = parameters.copy() + if operation in USER_OPERATIONS: + execution_parameters["user"] = user + if operation in TRIGGER_SOURCE_OPERATIONS: + execution_parameters["trigger_source"] = trigger_source + modified_field = LEGACY_MODIFIED_FIELDS.get(operation) + with set_actor(user): + old_values = ( + snapshot_field(document_ids, modified_field) + if settings.AUDIT_LOG_ENABLED and modified_field + else {} + ) + method(document_ids, **execution_parameters) + if settings.AUDIT_LOG_ENABLED and modified_field: + from auditlog.models import LogEntry + + new_values = snapshot_field(document_ids, modified_field) + for doc in Document.objects.filter(pk__in=document_ids): + LogEntry.objects.log_create( + instance=doc, + changes={ + modified_field: [old_values[doc.pk], new_values[doc.pk]], + }, + action=LogEntry.Action.UPDATE, + actor=user, + additional_data={"reason": f"Bulk edit: {operation}"}, + ) + return {"operation": operation, "document_count": len(document_ids)} + affected_ids: list[int] = [] modified_field = MODIFIED_FIELDS[operation] # Keep metadata writes and their audit entries together. Follow-up processing diff --git a/src/documents/signals/handlers.py b/src/documents/signals/handlers.py index 8297762e97..cb5c376b14 100644 --- a/src/documents/signals/handlers.py +++ b/src/documents/signals/handlers.py @@ -1206,6 +1206,8 @@ def before_task_publish_handler( "document_count": len(task_kwargs["document_ids"]), } owner_id = task_kwargs["user_id"] + if task_kwargs["operation"] == "delete": + task_type = PaperlessTask.TaskType.BULK_DELETE else: input_data = _extract_input_data(task_type, task_kwargs) owner_id = _extract_owner_id(task_type, task_kwargs) diff --git a/src/documents/views.py b/src/documents/views.py index 31aaa4b345..14faf32d74 100644 --- a/src/documents/views.py +++ b/src/documents/views.py @@ -107,9 +107,9 @@ from documents import bulk_edit from documents.bulk_download import ArchiveOnlyStrategy from documents.bulk_download import OriginalAndArchiveStrategy from documents.bulk_download import OriginalsOnlyStrategy -from documents.bulk_operations import QUEUED_OPERATIONS +from documents.bulk_operations import TRIGGER_SOURCE_OPERATIONS +from documents.bulk_operations import USER_OPERATIONS from documents.bulk_operations import execute_bulk_operation -from documents.bulk_operations import snapshot_field from documents.caching import get_llm_suggestion_cache from documents.caching import get_metadata_cache from documents.caching import get_suggestion_cache @@ -2974,38 +2974,30 @@ class DocumentSelectionMixin: class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin): permission_classes = (IsAuthenticated,) parser_classes = (parsers.JSONParser,) - METHOD_NAMES_REQUIRING_USER = { - "split", - "merge", - "rotate", - "delete_pages", - "edit_pdf", - "remove_password", - "merge_as_versions", - } - # merge_as_versions doesn't queue any consume tasks - METHOD_NAMES_REQUIRING_TRIGGER_SOURCE = METHOD_NAMES_REQUIRING_USER - { - "merge_as_versions", - } def _queue_bulk_operation(self, method, documents, parameters, from_webui): if "update_callback" in parameters: raise ValueError("Unsupported bulk operation parameter: update_callback") - signature(method).bind(documents, **parameters) + trigger_source = ( + PaperlessTask.TriggerSource.WEB_UI + if from_webui + else PaperlessTask.TriggerSource.API_UPLOAD + ) + final_parameters = parameters.copy() + if method.__name__ in USER_OPERATIONS: + final_parameters["user"] = self.request.user + if method.__name__ in TRIGGER_SOURCE_OPERATIONS: + final_parameters["trigger_source"] = trigger_source + signature(method).bind(documents, **final_parameters) task = execute_bulk_operation.apply_async( kwargs={ "operation": method.__name__, "document_ids": documents, "parameters": parameters, "user_id": self.request.user.pk, + "trigger_source": trigger_source, }, - headers={ - "trigger_source": ( - PaperlessTask.TriggerSource.WEB_UI - if from_webui - else PaperlessTask.TriggerSource.API_UPLOAD - ), - }, + headers={"trigger_source": trigger_source}, ) return Response({"result": "OK", "task_id": task.id}) @@ -3129,9 +3121,9 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin): user = self.request.user from_webui = validated_data.get("from_webui", False) - if method.__name__ in self.METHOD_NAMES_REQUIRING_USER: + if method.__name__ in USER_OPERATIONS: parameters["user"] = user - if method.__name__ in self.METHOD_NAMES_REQUIRING_TRIGGER_SOURCE: + if method.__name__ in TRIGGER_SOURCE_OPERATIONS: parameters["trigger_source"] = ( PaperlessTask.TriggerSource.WEB_UI if from_webui @@ -3169,19 +3161,13 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin): name="BulkEditDocumentsResult", fields={ "result": serializers.CharField(), - "task_id": serializers.CharField(required=False), + "task_id": serializers.CharField(), }, ), }, ), ) class BulkEditView(DocumentOperationPermissionMixin): - LEGACY_MODIFIED_FIELDS = { - "delete": "deleted_at", - "reprocess": "checksum", - "remove_password": None, - } - serializer_class = BulkEditSerializer def post(self, request, *args, **kwargs): @@ -3211,14 +3197,6 @@ class BulkEditView(DocumentOperationPermissionMixin): user=user, validated_data=serializer.validated_data, ) - if method.__name__ in self.METHOD_NAMES_REQUIRING_USER: - parameters["user"] = user - if method.__name__ in self.METHOD_NAMES_REQUIRING_TRIGGER_SOURCE: - parameters["trigger_source"] = ( - PaperlessTask.TriggerSource.WEB_UI - if from_webui - else PaperlessTask.TriggerSource.API_UPLOAD - ) if not has_document_operation_permissions( user=user, documents=documents, @@ -3228,38 +3206,12 @@ class BulkEditView(DocumentOperationPermissionMixin): return HttpResponseForbidden("Insufficient permissions") try: - if method.__name__ in QUEUED_OPERATIONS: - return self._queue_bulk_operation( - method, - documents, - parameters, - from_webui, - ) - modified_field = self.LEGACY_MODIFIED_FIELDS.get(method.__name__) - if settings.AUDIT_LOG_ENABLED and modified_field: - old_values = snapshot_field(documents, modified_field) - - result = method(documents, **parameters) - - if settings.AUDIT_LOG_ENABLED and modified_field: - new_values = snapshot_field(documents, modified_field) - for doc in Document.objects.filter(pk__in=documents): - LogEntry.objects.log_create( - instance=doc, - changes={ - modified_field: [ - old_values[doc.pk], - new_values[doc.pk], - ], - }, - action=LogEntry.Action.UPDATE, - actor=user, - additional_data={ - "reason": f"Bulk edit: {method.__name__}", - }, - ) - - return Response({"result": result}) + return self._queue_bulk_operation( + method, + documents, + parameters, + from_webui, + ) except Exception as e: logger.warning(f"An error occurred performing bulk edit: {e!s}") return HttpResponseBadRequest(