From 5fd6ded07f9b7ac1403707dcf383a99e7ba45f70 Mon Sep 17 00:00:00 2001 From: kdeterme Date: Fri, 21 Aug 2026 15:53:40 +0200 Subject: [PATCH] feat: update import_as_built_documents command to support recursive directory structures and add a --clean-existing option. --- .../commands/import_as_built_documents.py | 99 +++++++++++++++---- loko/documents/models.py | 7 +- 2 files changed, 86 insertions(+), 20 deletions(-) diff --git a/loko/assets/management/commands/import_as_built_documents.py b/loko/assets/management/commands/import_as_built_documents.py index 2830187..731a44b 100644 --- a/loko/assets/management/commands/import_as_built_documents.py +++ b/loko/assets/management/commands/import_as_built_documents.py @@ -12,6 +12,7 @@ from django.utils.text import slugify from assets.models.structures import Structure from documents.models import DocumentFolder, DocumentFolderAttachment, ManagedDocument, DocumentVersion, DocumentAttachment, DocumentTag +from documents.services import DocumentManagementService User = get_user_model() @@ -19,7 +20,7 @@ User = get_user_model() class Command(BaseCommand): help = ( "Imports As-Built documents from a local directory, a ZIP archive, or via rclone " - "into the 'Electromechanics / As-Built' subfolder of specified structures." + "into the 'Electromechanics / As-Built' subfolder (preserving subfolder structure) of specified structures." ) def add_arguments(self, parser): @@ -43,6 +44,12 @@ class Command(BaseCommand): nargs="+", help="One or more structure codes (e.g. OA006 OA007) to target specifically.", ) + parser.add_argument( + "--clean-existing", + action="store_true", + default=False, + help="Deletes existing documents in the As-Built subfolder hierarchy before importing.", + ) parser.add_argument( "--dry-run", action="store_true", @@ -65,6 +72,7 @@ class Command(BaseCommand): source_zip = options["source_zip"] rclone_remote = options["rclone_remote"] target_codes = options["structures"] + clean_existing = options["clean_existing"] dry_run = options["dry_run"] username = options["user"] tag_name = options["tag"] @@ -164,6 +172,44 @@ class Command(BaseCommand): return ab_folder + def get_or_create_nested_folder(parent, rel_dir_path): + if not rel_dir_path or rel_dir_path == ".": + return parent + + parts = [p.strip() for p in rel_dir_path.split("/") if p.strip()] + current = parent + + for part in parts: + base_slug = f"{current.slug}-{slugify(part)}" + slug = base_slug[:150] + + folder = current.child_folders.filter(name=part).first() if current.pk else None + if not folder: + folder = DocumentFolder.objects.filter(slug=slug).first() + + if not folder: + counter = 1 + while DocumentFolder.objects.filter(slug=slug).exists(): + slug = f"{base_slug[:140]}-{counter}" + counter += 1 + + if dry_run: + folder = DocumentFolder(name=part, slug=slug) + else: + folder, _ = DocumentFolder.objects.get_or_create( + slug=slug, + defaults={ + "name": part, + "application_label": current.application_label, + "created_by": creator_user + } + ) + folder.thematics.set(current.thematics.all()) + folder.parent_folders.add(current) + current = folder + + return current + # Determine structures to process if target_codes: structures = Structure.objects.filter(code__in=target_codes) @@ -198,21 +244,23 @@ class Command(BaseCommand): struct_dir = p break - # If no dedicated directory and specific structures were targeted, check direct files in source_dir - files_to_import = [] + # Collect files with their relative subfolder paths + file_entries = [] # tuples of (full_file_path, relative_dir_path) if struct_dir: for root, _, files in os.walk(struct_dir): + rel_dir = os.path.relpath(root, struct_dir) for f in files: if not f.startswith("."): - files_to_import.append(os.path.join(root, f)) + file_entries.append((os.path.join(root, f), rel_dir)) elif target_codes: # If specific structures requested but no subfolder, import all files in source_dir for these structures - for f in os.listdir(source_dir): - full_path = os.path.join(source_dir, f) - if os.path.isfile(full_path) and not f.startswith("."): - files_to_import.append(full_path) + for root, _, files in os.walk(source_dir): + rel_dir = os.path.relpath(root, source_dir) + for f in files: + if not f.startswith("."): + file_entries.append((os.path.join(root, f), rel_dir)) - if not files_to_import: + if not file_entries: continue self.stdout.write(f"\nProcessing Structure [{structure.code}] {structure.get_name()}") @@ -222,27 +270,42 @@ class Command(BaseCommand): self.stdout.write(self.style.ERROR(f" No default folder/As-Built folder for Structure {structure.code}. Skipped.")) continue - self.stdout.write(f" Target subfolder: Electromechanics / As-Built (ID: {as_built_folder.pk or 'dry-run'})") - self.stdout.write(f" Found {len(files_to_import)} file(s) to process.") + if clean_existing and as_built_folder.pk: + all_ab_folders = DocumentManagementService.get_folder_and_descendants(as_built_folder) + existing_docs = ManagedDocument.objects.filter( + folders__pk__in=all_ab_folders, + is_archived=False + ).distinct() + deleted_count = existing_docs.count() + if deleted_count > 0: + self.stdout.write(self.style.WARNING(f" Cleaning {deleted_count} existing document(s) in As-Built folder...")) + if not dry_run: + existing_docs.delete() - for file_path in files_to_import: + self.stdout.write(f" Target root subfolder: Electromechanics / As-Built (ID: {as_built_folder.pk or 'dry-run'})") + self.stdout.write(f" Found {len(file_entries)} file(s) to process.") + + for file_path, rel_dir_path in file_entries: rel_name = os.path.basename(file_path) doc_title = os.path.splitext(rel_name)[0] + target_folder = get_or_create_nested_folder(as_built_folder, rel_dir_path) + # Deduplication check - if as_built_folder.pk: + if target_folder.pk and not clean_existing: existing_doc = ManagedDocument.objects.filter( - folders=as_built_folder, + folders=target_folder, title=doc_title, is_archived=False ).first() if existing_doc: - self.stdout.write(f" Document '{rel_name}' already exists in As-Built folder (Doc ID: {existing_doc.pk}). Skipped.") + self.stdout.write(f" Document '{rel_name}' already exists in folder '{target_folder.name}' (Doc ID: {existing_doc.pk}). Skipped.") total_skipped += 1 continue + folder_display = rel_dir_path if rel_dir_path != "." else "As-Built" if dry_run: - self.stdout.write(f" [Dry-run] Would import '{rel_name}' into structure {structure.code}") + self.stdout.write(f" [Dry-run] Would import '{rel_name}' into folder 'As-Built / {folder_display}'") total_imported += 1 else: with open(file_path, "rb") as f: @@ -264,7 +327,7 @@ class Command(BaseCommand): ) version.file.save(rel_name, ContentFile(file_data), save=True) - doc.folders.add(as_built_folder) + doc.folders.add(target_folder) if doc_tag: doc.tags.add(doc_tag) @@ -275,7 +338,7 @@ class Command(BaseCommand): attached_by=creator_user ) - self.stdout.write(self.style.SUCCESS(f" Imported '{rel_name}' (Doc ID: {doc.pk})")) + self.stdout.write(self.style.SUCCESS(f" Imported '{rel_name}' into 'As-Built / {folder_display}' (Doc ID: {doc.pk})")) total_imported += 1 self.stdout.write(self.style.SUCCESS(f"\nDone! Imported: {total_imported}, Skipped (already imported): {total_skipped}")) diff --git a/loko/documents/models.py b/loko/documents/models.py index 1636365..a45bc3a 100644 --- a/loko/documents/models.py +++ b/loko/documents/models.py @@ -599,6 +599,8 @@ class DocumentAttachment(models.Model): self.apply_default_folder() def apply_default_folder(self): + if self.document.folders.exists(): + return DocumentFolderAttachment = apps.get_model("documents", "DocumentFolderAttachment") default_folder_link = ( DocumentFolderAttachment.objects.select_related("folder") @@ -646,13 +648,14 @@ class DocumentFolderAttachment(models.Model): def save(self, *args, **kwargs): super().save(*args, **kwargs) if self.is_default: - # ensure default folder is applied to all attached documents + # ensure default folder is applied to attached documents without folders DocumentAttachment = apps.get_model("documents", "DocumentAttachment") attachments = DocumentAttachment.objects.filter( content_type=self.content_type, object_id=self.object_id ).select_related("document") for attachment in attachments: - attachment.document.ensure_folder(self.folder) + if not attachment.document.folders.exists(): + attachment.document.ensure_folder(self.folder) def duplicate_file(field):