Basic remove password bulk edit action

This commit is contained in:
shamoon
2025-10-31 21:09:17 -07:00
parent 919c54c6ba
commit 0d4e0baf56
7 changed files with 128 additions and 1 deletions

View File

@@ -644,6 +644,77 @@ def edit_pdf(
return "OK"
def remove_password(
doc_ids: list[int],
password: str,
*,
delete_original: bool = False,
update_document: bool = False,
include_metadata: bool = True,
user: User | None = None,
) -> Literal["OK"]:
"""
Remove password protection from PDF documents.
"""
import pikepdf
for doc_id in doc_ids:
doc = Document.objects.get(id=doc_id)
try:
logger.info(
f"Attempting password removal from document {doc_ids[0]}",
)
with pikepdf.open(doc.source_path, password=password) as pdf:
temp_path = doc.source_path.with_suffix(".tmp.pdf")
pdf.remove_unreferenced_resources()
pdf.save(temp_path)
if update_document:
# replace the original document with the unprotected one
temp_path.replace(doc.source_path)
doc.checksum = hashlib.md5(doc.source_path.read_bytes()).hexdigest()
doc.page_count = len(pdf.pages)
doc.save()
update_document_content_maybe_archive_file.delay(document_id=doc.id)
else:
consume_tasks = []
overrides = (
DocumentMetadataOverrides().from_document(doc)
if include_metadata
else DocumentMetadataOverrides()
)
if user is not None:
overrides.owner_id = user.id
filepath: Path = (
Path(tempfile.mkdtemp(dir=settings.SCRATCH_DIR))
/ f"{doc.id}_unprotected.pdf"
)
temp_path.replace(filepath)
consume_tasks.append(
consume_file.s(
ConsumableDocument(
source=DocumentSource.ConsumeFolder,
original_file=filepath,
),
overrides,
),
)
if delete_original:
chord(header=consume_tasks, body=delete.si([doc.id])).delay()
else:
group(consume_tasks).delay()
except Exception as e:
logger.exception(f"Error removing password from document {doc.id}: {e}")
raise ValueError(
f"An error occurred while removing the password: {e}",
) from e
return "OK"
def reflect_doclinks(
document: Document,
field: CustomField,

View File

@@ -1400,6 +1400,7 @@ class BulkEditSerializer(
"split",
"delete_pages",
"edit_pdf",
"remove_password",
],
label="Method",
write_only=True,
@@ -1475,6 +1476,8 @@ class BulkEditSerializer(
return bulk_edit.delete_pages
elif method == "edit_pdf":
return bulk_edit.edit_pdf
elif method == "remove_password":
return bulk_edit.remove_password
else: # pragma: no cover
# This will never happen as it is handled by the ChoiceField
raise serializers.ValidationError("Unsupported method.")
@@ -1671,6 +1674,12 @@ class BulkEditSerializer(
f"Page {op['page']} is out of bounds for document with {doc.page_count} pages.",
)
def validate_parameters_remove_password(self, parameters):
if "password" not in parameters:
raise serializers.ValidationError("password not specified")
if not isinstance(parameters["password"], str):
raise serializers.ValidationError("password must be a string")
def validate(self, attrs):
method = attrs["method"]
parameters = attrs["parameters"]
@@ -1711,6 +1720,8 @@ class BulkEditSerializer(
"Edit PDF method only supports one document",
)
self._validate_parameters_edit_pdf(parameters, attrs["documents"][0])
elif method == bulk_edit.remove_password:
self.validate_parameters_remove_password(parameters)
return attrs

View File

@@ -1503,6 +1503,7 @@ class BulkEditView(PassUserMixin):
"merge": None,
"edit_pdf": "checksum",
"reprocess": "checksum",
"remove_password": "checksum",
}
permission_classes = (IsAuthenticated,)
@@ -1521,6 +1522,7 @@ class BulkEditView(PassUserMixin):
bulk_edit.split,
bulk_edit.merge,
bulk_edit.edit_pdf,
bulk_edit.remove_password,
]:
parameters["user"] = user
@@ -1549,6 +1551,7 @@ class BulkEditView(PassUserMixin):
bulk_edit.rotate,
bulk_edit.delete_pages,
bulk_edit.edit_pdf,
bulk_edit.remove_password,
]
)
or (
@@ -1565,7 +1568,7 @@ class BulkEditView(PassUserMixin):
and (
method in [bulk_edit.split, bulk_edit.merge]
or (
method == bulk_edit.edit_pdf
method in [bulk_edit.edit_pdf, bulk_edit.remove_password]
and not parameters["update_document"]
)
)