From dda68cabecb7fae96c1297182a92f600f5b3cf33 Mon Sep 17 00:00:00 2001 From: kdeterme Date: Fri, 21 Aug 2026 14:51:13 +0200 Subject: [PATCH] feat: add As-Built subfolder to structures and implement import_as_built_documents management command --- .../commands/import_as_built_documents.py | 219 ++++++++++++++++++ ...create_structure_subfolders_and_reports.py | 2 + loko/interventions/utils.py | 1 + 3 files changed, 222 insertions(+) create mode 100644 loko/assets/management/commands/import_as_built_documents.py diff --git a/loko/assets/management/commands/import_as_built_documents.py b/loko/assets/management/commands/import_as_built_documents.py new file mode 100644 index 0000000..5a5e608 --- /dev/null +++ b/loko/assets/management/commands/import_as_built_documents.py @@ -0,0 +1,219 @@ +import os +from django.core.management.base import BaseCommand, CommandError +from django.contrib.contenttypes.models import ContentType +from django.contrib.auth import get_user_model +from django.core.files.base import ContentFile +from django.db import transaction +from django.utils import timezone +from django.utils.text import slugify + +from assets.models.structures import Structure +from documents.models import DocumentFolder, DocumentFolderAttachment, ManagedDocument, DocumentVersion, DocumentAttachment, DocumentTag + +User = get_user_model() + + +class Command(BaseCommand): + help = ( + "Imports As-Built documents from a local directory into the " + "'Electromechanics / As-Built' subfolder of specified structures." + ) + + def add_arguments(self, parser): + parser.add_argument( + "--source-dir", + required=True, + help="Path to the directory containing files/folders to import.", + ) + parser.add_argument( + "--structures", + nargs="+", + help="One or more structure codes (e.g. OA006 OA007) to target specifically.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + default=False, + help="Simulates document import without saving to database.", + ) + parser.add_argument( + "--user", + default=None, + help="Username of creator/uploader.", + ) + parser.add_argument( + "--tag", + default=None, + help="Optional tag name or slug to attach to imported documents.", + ) + + def handle(self, *args, **options): + source_dir = options["source_dir"] + target_codes = options["structures"] + dry_run = options["dry_run"] + username = options["user"] + tag_name = options["tag"] + + if not os.path.isdir(source_dir): + raise CommandError(f"Source directory '{source_dir}' does not exist or is not a directory.") + + if dry_run: + self.stdout.write(self.style.WARNING("DRY RUN — no changes will be committed.")) + + # Creator user lookup + creator_user = None + if username: + try: + creator_user = User.objects.get(username=username) + except User.DoesNotExist: + raise CommandError(f"User '{username}' not found.") + else: + creator_user = User.objects.filter(username="kdeterme").first() or User.objects.filter(is_superuser=True).first() + + structure_ct = ContentType.objects.get_for_model(Structure) + + # Optional tag lookup + doc_tag = None + if tag_name: + doc_tag = DocumentTag.objects.filter(slug=slugify(tag_name)).first() or DocumentTag.objects.filter(name_fr__iexact=tag_name).first() + + def get_or_create_as_built_folder(structure): + root_att = DocumentFolderAttachment.objects.filter( + content_type=structure_ct, + object_id=structure.pk, + is_default=True + ).select_related("folder").first() + + if not root_att: + return None + + root_folder = root_att.folder + + # Electromechanics subfolder + em_folder = root_folder.child_folders.filter(name="Electromechanics").first() + if not em_folder: + em_slug = f"{root_folder.slug}-electromechanics"[:150] + if dry_run: + em_folder = DocumentFolder(name="Electromechanics", slug=em_slug) + else: + em_folder, _ = DocumentFolder.objects.get_or_create( + slug=em_slug, + defaults={ + "name": "Electromechanics", + "application_label": root_folder.application_label, + "created_by": creator_user + } + ) + em_folder.parent_folders.add(root_folder) + + # As-Built subfolder + ab_folder = em_folder.child_folders.filter(name="As-Built").first() if em_folder.pk else None + if not ab_folder: + ab_slug = f"{em_folder.slug}-as-built"[:150] + if dry_run: + ab_folder = DocumentFolder(name="As-Built", slug=ab_slug) + else: + ab_folder, _ = DocumentFolder.objects.get_or_create( + slug=ab_slug, + defaults={ + "name": "As-Built", + "application_label": em_folder.application_label, + "created_by": creator_user + } + ) + ab_folder.parent_folders.add(em_folder) + + return ab_folder + + # Determine structures to process + if target_codes: + structures = Structure.objects.filter(code__in=target_codes) + if not structures.exists(): + raise CommandError(f"No structures found matching codes: {target_codes}") + else: + structures = Structure.objects.all() + + total_imported = 0 + + for structure in structures: + # Check if there is a dedicated subfolder for this structure code or name in source_dir + struct_dir = None + possible_dirs = [ + os.path.join(source_dir, structure.code), + os.path.join(source_dir, f"{structure.code}_{structure.name_fr}"), + os.path.join(source_dir, structure.name_fr or ""), + ] + for p in possible_dirs: + if os.path.isdir(p): + struct_dir = p + break + + # If no dedicated directory and specific structures were targeted, check direct files in source_dir + files_to_import = [] + if struct_dir: + for root, _, files in os.walk(struct_dir): + for f in files: + if not f.startswith("."): + files_to_import.append(os.path.join(root, f)) + 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) + + if not files_to_import: + continue + + self.stdout.write(f"\nProcessing Structure [{structure.code}] {structure.get_name()}") + as_built_folder = get_or_create_as_built_folder(structure) + + if not as_built_folder: + 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 import.") + + for file_path in files_to_import: + rel_name = os.path.basename(file_path) + doc_title = os.path.splitext(rel_name)[0] + + if dry_run: + self.stdout.write(f" [Dry-run] Would import '{rel_name}' into structure {structure.code}") + total_imported += 1 + else: + with open(file_path, "rb") as f: + file_data = f.read() + + with transaction.atomic(): + doc = ManagedDocument.objects.create( + title=doc_title, + description=f"As-Built document imported for structure {structure.code}", + created_by=creator_user, + ) + + version = DocumentVersion( + document=doc, + uploaded_by=creator_user, + comment="Imported As-Built document", + status=DocumentVersion.STATUS_APPROVED, + validated_at=timezone.now(), + ) + version.file.save(rel_name, ContentFile(file_data), save=True) + + doc.folders.add(as_built_folder) + if doc_tag: + doc.tags.add(doc_tag) + + DocumentAttachment.objects.create( + document=doc, + content_type=structure_ct, + object_id=structure.pk, + attached_by=creator_user + ) + + self.stdout.write(self.style.SUCCESS(f" Imported '{rel_name}' (Doc ID: {doc.pk})")) + total_imported += 1 + + self.stdout.write(self.style.SUCCESS(f"\nDone! Total documents imported: {total_imported}")) diff --git a/loko/interventions/management/commands/create_structure_subfolders_and_reports.py b/loko/interventions/management/commands/create_structure_subfolders_and_reports.py index c2ed8bc..68fa665 100644 --- a/loko/interventions/management/commands/create_structure_subfolders_and_reports.py +++ b/loko/interventions/management/commands/create_structure_subfolders_and_reports.py @@ -106,6 +106,7 @@ class Command(BaseCommand): # - Exceptional transport # - Electromechanics # - Preventive maintenance + # - As-Built # Helper to retrieve/create subfolders safely def get_or_create_subfolder(parent, name): @@ -150,6 +151,7 @@ class Command(BaseCommand): electromec_folder = get_or_create_subfolder(root_folder, "Electromechanics") prev_maint_folder = get_or_create_subfolder(electromec_folder, "Preventive maintenance") + get_or_create_subfolder(electromec_folder, "As-Built") # 3. Retrieve all preventive interventions since 01/01/2026 on this structure loc_ids = list(structure.all_locations.values_list("id", flat=True)) diff --git a/loko/interventions/utils.py b/loko/interventions/utils.py index 815d229..0c6df07 100644 --- a/loko/interventions/utils.py +++ b/loko/interventions/utils.py @@ -204,6 +204,7 @@ def create_preventive_intervention_report(intervention, creator_user=None): electromec_folder = get_or_create_subfolder(root_folder, "Electromechanics") prev_maint_folder = get_or_create_subfolder(electromec_folder, "Preventive maintenance") + get_or_create_subfolder(electromec_folder, "As-Built") # 3. Check if report already exists for this structure and intervention report_exists = ManagedDocument.objects.filter(