Valogeo/loko
Valogeo/loko
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: update import_as_built_documents command to support recursive directory structures and add a --clean-existing option.

Browse source
This commit is contained in:
kdeterme 2026-08-21 15:53:40 +02:00
parent 04eb087cf1
commit 5fd6ded07f
2 changed files with 86 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

99
loko/assets/management/commands/import_as_built_documents.py
View file

@ -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}"))

7
loko/documents/models.py
View file

@ -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):