feat: add As-Built subfolder to structures and implement import_as_built_documents management command

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

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

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(