import os from itertools import islice from pathlib import Path from typing import TYPE_CHECKING from typing import Any from allauth.mfa.models import Authenticator from allauth.socialaccount.models import SocialAccount from allauth.socialaccount.models import SocialApp from allauth.socialaccount.models import SocialToken from django.conf import settings from django.contrib.auth.models import Group from django.contrib.auth.models import Permission from django.contrib.auth.models import User from django.contrib.contenttypes.models import ContentType from django.core import serializers from django.core.management.base import CommandError from django.db import transaction from django.utils import timezone from filelock import FileLock from guardian.models import GroupObjectPermission from guardian.models import UserObjectPermission if TYPE_CHECKING: from collections.abc import Generator from django.db.models import QuerySet if settings.AUDIT_LOG_ENABLED: from auditlog.models import LogEntry from documents.export.compression import COMPRESSION_CHOICES from documents.export.compression import COMPRESSION_METHODS from documents.export.compression import ZSTD from documents.export.compression import compression_available from documents.export.compression import level_error from documents.export.sinks import DirectoryExportSink from documents.export.sinks import ExportSink from documents.export.sinks import StreamingManifestWriter from documents.export.sinks import ZipExportSink from documents.file_handling import generate_filename from documents.management.commands.base import PaperlessCommand from documents.management.commands.mixins import CryptMixin from documents.models import Correspondent from documents.models import CustomField from documents.models import CustomFieldInstance from documents.models import Document from documents.models import DocumentType from documents.models import Note from documents.models import SavedView from documents.models import SavedViewFilterRule from documents.models import ShareLink from documents.models import ShareLinkBundle from documents.models import StoragePath from documents.models import Tag from documents.models import UiSettings from documents.models import Workflow from documents.models import WorkflowAction from documents.models import WorkflowActionEmail from documents.models import WorkflowActionWebhook from documents.models import WorkflowTrigger from documents.settings import EXPORTER_ARCHIVE_NAME from documents.settings import EXPORTER_FILE_NAME from documents.settings import EXPORTER_SHARE_LINK_BUNDLE_NAME from documents.settings import EXPORTER_THUMBNAIL_NAME from documents.utils import QuerySetStream from paperless import version from paperless.models import ApplicationConfiguration from paperless_mail.models import MailAccount from paperless_mail.models import MailRule def serialize_queryset_batched( queryset: "QuerySet[Any]", *, batch_size: int = 500, ) -> "Generator[list[dict], None, None]": """Yield batches of serialized records from a QuerySet. Each batch is a list of dicts in Django's Python serialization format. Uses QuerySet.iterator() to avoid loading the full queryset into memory, and islice to collect chunk-sized batches serialized in a single call. """ iterator = queryset.iterator(chunk_size=batch_size) while chunk := list(islice(iterator, batch_size)): yield serializers.serialize("python", chunk) class Command(CryptMixin, PaperlessCommand): help = ( "Decrypt and rename all files in our collection into a given target " "directory. And include a manifest file containing document data for " "easy import." ) supports_progress_bar = True supports_multiprocessing = False def add_arguments(self, parser) -> None: super().add_arguments(parser) parser.add_argument("target") parser.add_argument( "-c", "--compare-checksums", default=False, action="store_true", help=( "Compare file checksums when determining whether to export " "a file or not. If not specified, file size and time " "modified is used instead." ), ) parser.add_argument( "-cj", "--compare-json", default=False, action="store_true", help=( "Compare json file checksums when determining whether to " "export a json file or not (manifest or metadata). " "If not specified, the file is always exported." ), ) parser.add_argument( "-d", "--delete", default=False, action="store_true", help=( "After exporting, delete files in the export directory that " "do not belong to the current export, such as files from " "deleted documents." ), ) parser.add_argument( "-f", "--use-filename-format", default=False, action="store_true", help=( "Use PAPERLESS_FILENAME_FORMAT for storing files in the " "export directory, if configured." ), ) parser.add_argument( "-na", "--no-archive", default=False, action="store_true", help="Avoid exporting archive files", ) parser.add_argument( "-nt", "--no-thumbnail", default=False, action="store_true", help="Avoid exporting thumbnail files", ) parser.add_argument( "-p", "--use-folder-prefix", default=False, action="store_true", help=( "Export files in dedicated folders according to their nature: " "archive, originals or thumbnails" ), ) parser.add_argument( "-sm", "--split-manifest", default=False, action="store_true", help="Export document information in individual manifest json files.", ) parser.add_argument( "-z", "--zip", default=False, action="store_true", help="Export the documents to a zip file in the given directory", ) parser.add_argument( "-zn", "--zip-name", default=f"export-{timezone.localdate().isoformat()}", help="Sets the export zip file name", ) parser.add_argument( "--zip-compression", choices=COMPRESSION_CHOICES, default=None, help=( "Compression method for the export zip (requires --zip). " "Default: deflated. 'zstd' requires Python 3.14+ on both the " "exporting and importing machine." ), ) parser.add_argument( "--zip-compression-level", type=int, default=None, help=( "Compression level for the export zip (requires --zip). " "deflated: 0-9, bzip2: 1-9, zstd: -22..22; ignored for " "stored/lzma." ), ) parser.add_argument( "--data-only", default=False, action="store_true", help="If set, only the database will be imported, not files", ) parser.add_argument( "--passphrase", help="If provided, is used to encrypt sensitive data in the export", ) parser.add_argument( "--batch-size", type=int, default=500, help=( "Number of records to process per batch during serialization. " "Lower values reduce peak memory usage; higher values improve " "throughput. Default: 500." ), ) def handle(self, *args, **options) -> None: self.target = Path(options["target"]).resolve() self.split_manifest: bool = options["split_manifest"] self.compare_checksums: bool = options["compare_checksums"] self.compare_json: bool = options["compare_json"] self.use_filename_format: bool = options["use_filename_format"] self.use_folder_prefix: bool = options["use_folder_prefix"] self.delete: bool = options["delete"] self.no_archive: bool = options["no_archive"] self.no_thumbnail: bool = options["no_thumbnail"] self.zip_export: bool = options["zip"] self.data_only: bool = options["data_only"] self.passphrase: str | None = options.get("passphrase") self.batch_size: int = options["batch_size"] self.exported_files: set[str] = set() if self.zip_export and (self.compare_checksums or self.compare_json): raise CommandError( "--compare-checksums and --compare-json have no effect when " "used with --zip", ) if not self.target.exists(): raise CommandError("That path doesn't exist") if not self.target.is_dir(): raise CommandError("That path isn't a directory") if not os.access(self.target, os.W_OK): raise CommandError("That path doesn't appear to be writable") zip_compression: str | None = options["zip_compression"] zip_compression_level: int | None = options["zip_compression_level"] if not self.zip_export and ( zip_compression is not None or zip_compression_level is not None ): raise CommandError( "--zip-compression and --zip-compression-level require --zip", ) compression_method = zip_compression or "deflated" if self.zip_export: if not compression_available(compression_method): if compression_method == "zstd" and ZSTD is None: raise CommandError( "zstd compression requires Python 3.14 or newer", ) raise CommandError( f"Compression method '{compression_method}' is not " f"available on this Python runtime", ) level_msg = level_error(compression_method, zip_compression_level) if level_msg is not None: raise CommandError(level_msg) sink: ExportSink if self.zip_export: sink = ZipExportSink( self.target, options["zip_name"], delete=self.delete, compression=COMPRESSION_METHODS[compression_method], compresslevel=zip_compression_level, ) else: sink = DirectoryExportSink( self.target, compare_checksums=self.compare_checksums, compare_json=self.compare_json, delete=self.delete, ) # Prevent any ongoing changes in the documents while exporting with FileLock(settings.MEDIA_LOCK), sink: self.dump(sink) def dump(self, sink: ExportSink) -> None: # 1. Create manifest, containing all correspondents, types, tags, storage # paths, note, documents and ui_settings _excluded_usernames = ["consumer", "AnonymousUser"] manifest_key_to_object_query: dict[str, QuerySet[Any]] = { "correspondents": Correspondent.objects.all(), "tags": Tag.objects.all(), "document_types": DocumentType.objects.all(), "storage_paths": StoragePath.objects.all(), "mail_accounts": MailAccount.objects.all(), "mail_rules": MailRule.objects.all(), "saved_views": SavedView.objects.all(), "saved_view_filter_rules": SavedViewFilterRule.objects.all(), "groups": Group.objects.all(), "users": User.objects.exclude( username__in=_excluded_usernames, ).all(), "ui_settings": UiSettings.objects.exclude( user__username__in=_excluded_usernames, ), "content_types": ContentType.objects.all(), "permissions": Permission.objects.all(), "user_object_permissions": UserObjectPermission.objects.exclude( user__username__in=_excluded_usernames, ), "group_object_permissions": GroupObjectPermission.objects.all(), "workflow_triggers": WorkflowTrigger.objects.all(), "workflow_actions": WorkflowAction.objects.all(), "workflow_email_actions": WorkflowActionEmail.objects.all(), "workflow_webhook_actions": WorkflowActionWebhook.objects.all(), "workflows": Workflow.objects.all(), "custom_fields": CustomField.objects.all(), "custom_field_instances": CustomFieldInstance.global_objects.all(), "app_configs": ApplicationConfiguration.objects.all(), "notes": Note.global_objects.all(), "documents": Document.global_objects.order_by("id").all(), "share_links": ShareLink.global_objects.all(), "share_link_bundles": ShareLinkBundle.objects.order_by("id").all(), "social_accounts": SocialAccount.objects.exclude( user__username__in=_excluded_usernames, ), "social_apps": SocialApp.objects.all(), "social_tokens": SocialToken.objects.exclude( account__user__username__in=_excluded_usernames, ), "authenticators": Authenticator.objects.exclude( user__username__in=_excluded_usernames, ), } if settings.AUDIT_LOG_ENABLED: manifest_key_to_object_query["log_entries"] = LogEntry.objects.all() # Crypto setup before streaming begins if self.passphrase: self.setup_crypto(passphrase=self.passphrase) elif MailAccount.objects.count() > 0 or SocialToken.objects.count() > 0: self.stdout.write( self.style.NOTICE( "No passphrase was given, sensitive fields will be in plaintext", ), ) document_manifest: list[dict] = [] share_link_bundle_manifest: list[dict] = [] with sink.stream("manifest.json") as handle: writer = StreamingManifestWriter(handle) with transaction.atomic(): for key, qs in manifest_key_to_object_query.items(): if key == "documents": # Accumulate for file-copy loop; written to manifest after for batch in serialize_queryset_batched( qs, batch_size=self.batch_size, ): for record in batch: self._encrypt_record_inline(record) document_manifest.extend(batch) elif key == "share_link_bundles": # Accumulate for file-copy loop; written to manifest after for batch in serialize_queryset_batched( qs, batch_size=self.batch_size, ): for record in batch: self._encrypt_record_inline(record) share_link_bundle_manifest.extend(batch) elif self.split_manifest and key in ( "notes", "custom_field_instances", ): # Written per-document in _write_split_manifest pass else: for batch in serialize_queryset_batched( qs, batch_size=self.batch_size, ): for record in batch: self._encrypt_record_inline(record) writer.write_batch(batch) share_link_bundle_map: dict[int, ShareLinkBundle] = { b.pk: b for b in ShareLinkBundle.objects.order_by("id").prefetch_related( "documents", ) } # 2. Export files from each document # document_manifest and this stream are both ordered by id from the # same underlying rows, so zip them in lockstep instead of building # a dict of every Document instance up front (QuerySetStream keeps # only one batch of documents resident at a time). documents_stream = QuerySetStream( Document.global_objects.order_by("id"), chunk_size=self.batch_size, ) for document_dict, document in self.track( zip(document_manifest, documents_stream, strict=True), description="Exporting documents...", total=len(document_manifest), ): # Both document_manifest and documents_stream come from the same # Document.global_objects.order_by("id") query, taken while # MEDIA_LOCK is held, so this should be unreachable -- it guards # against silent data corruption if that invariant ever breaks. if document.pk != document_dict["pk"]: # pragma: no cover raise CommandError( "Document export ordering mismatch: expected " f"pk={document_dict['pk']}, got pk={document.pk}. " "Documents may have changed during export.", ) # generate a unique filename, then the arcnames for its files base_name = self.generate_base_name(document) original_arc, thumbnail_arc, archive_arc = ( self.generate_document_targets(document, base_name, document_dict) ) if not self.data_only: self.copy_document_files( document, sink, original_arc, thumbnail_arc, archive_arc, ) if self.split_manifest: self._write_split_manifest(sink, document_dict, document, base_name) else: writer.write_record(document_dict) for bundle_dict in share_link_bundle_manifest: bundle = share_link_bundle_map[bundle_dict["pk"]] bundle_arc = self.generate_share_link_bundle_target( bundle, bundle_dict, ) if not self.data_only and bundle_arc is not None: self.copy_share_link_bundle_file(bundle, sink, bundle_arc) writer.write_record(bundle_dict) writer.close() # 3. Write version (and crypto params) to metadata.json # Django stores most crypto values in the field itself; we store # them once here for the whole export metadata: dict[str, str | int | dict[str, str | int]] = { "version": version.__full_version_str__, } if self.passphrase: metadata.update(self.get_crypt_params()) sink.add_json(metadata, "metadata.json") def generate_base_name(self, document: Document) -> Path: """ Generates a unique name for the document, one which hasn't already been exported (or will be) """ filename_counter = 0 while True: if self.use_filename_format: base_name = generate_filename( document, counter=filename_counter, ) else: base_name = document.get_public_filename(counter=filename_counter) if base_name not in self.exported_files: self.exported_files.add(base_name) break else: filename_counter += 1 return Path(base_name) def generate_document_targets( self, document: Document, base_name: Path, document_dict: dict, ) -> tuple[str, str | None, str | None]: """ Generates the relative POSIX arcnames for a document's original, thumbnail and archive files (depending on settings), and records them in the manifest. """ original_name = base_name if self.use_folder_prefix: original_name = Path("originals") / original_name original_arc = original_name.as_posix() document_dict[EXPORTER_FILE_NAME] = original_arc if not self.no_thumbnail: thumbnail_name = base_name.parent / (base_name.stem + "-thumbnail.webp") if self.use_folder_prefix: thumbnail_name = Path("thumbnails") / thumbnail_name thumbnail_arc = thumbnail_name.as_posix() document_dict[EXPORTER_THUMBNAIL_NAME] = thumbnail_arc else: thumbnail_arc = None if not self.no_archive and document.has_archive_version: archive_name = base_name.parent / (base_name.stem + "-archive.pdf") if self.use_folder_prefix: archive_name = Path("archive") / archive_name archive_arc = archive_name.as_posix() document_dict[EXPORTER_ARCHIVE_NAME] = archive_arc else: archive_arc = None return original_arc, thumbnail_arc, archive_arc def copy_document_files( self, document: Document, sink: ExportSink, original_arc: str, thumbnail_arc: str | None, archive_arc: str | None, ) -> None: """ Hands the document's files to the sink (original, thumbnail, archive). """ sink.add_file(document.source_path, original_arc, checksum=document.checksum) if thumbnail_arc: sink.add_file(document.thumbnail_path, thumbnail_arc) if archive_arc: if TYPE_CHECKING: assert isinstance(document.archive_path, Path) sink.add_file( document.archive_path, archive_arc, checksum=document.archive_checksum, ) def generate_share_link_bundle_target( self, bundle: ShareLinkBundle, bundle_dict: dict, ) -> str | None: """ Generates the relative POSIX arcname for a share link bundle file, if any. """ if not bundle.file_path: return None stored_bundle_path = Path(bundle.file_path) portable_bundle_path = ( stored_bundle_path if not stored_bundle_path.is_absolute() else Path(stored_bundle_path.name) ) export_bundle_path = Path("share_link_bundles") / portable_bundle_path bundle_dict["fields"]["file_path"] = portable_bundle_path.as_posix() bundle_dict[EXPORTER_SHARE_LINK_BUNDLE_NAME] = export_bundle_path.as_posix() return export_bundle_path.as_posix() def copy_share_link_bundle_file( self, bundle: ShareLinkBundle, sink: ExportSink, bundle_arc: str, ) -> None: """ Hands a share link bundle ZIP to the sink. """ bundle_source_path = bundle.absolute_file_path if bundle_source_path is None: raise FileNotFoundError(f"Share link bundle {bundle.pk} has no file path") sink.add_file(bundle_source_path, bundle_arc) def _encrypt_record_inline(self, record: dict) -> None: """Encrypt sensitive fields in a single record, if passphrase is set.""" if not self.passphrase: return fields = self.CRYPT_FIELDS_BY_MODEL.get(record.get("model", "")) if fields: for field in fields: if record["fields"].get(field): record["fields"][field] = self.encrypt_string( value=record["fields"][field], ) def _write_split_manifest( self, sink: ExportSink, document_dict: dict, document: Document, base_name: Path, ) -> None: """Write per-document manifest file for --split-manifest mode.""" content = [document_dict] content.extend( serializers.serialize( "python", Note.global_objects.filter(document=document), ), ) content.extend( serializers.serialize( "python", CustomFieldInstance.global_objects.filter(document=document), ), ) manifest_name = base_name.with_name(f"{base_name.stem}-manifest.json") if self.use_folder_prefix: manifest_name = Path("json") / manifest_name sink.add_json(content, manifest_name.as_posix())