373 lines
17 KiB
Python
373 lines
17 KiB
Python
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import zipfile
|
|
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
|
|
from documents.services import DocumentManagementService
|
|
|
|
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 (preserving subfolder structure) of specified structures."
|
|
)
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
"--source-dir",
|
|
default="/tmp/as_built_sync",
|
|
help="Path to local directory containing files/folders to import (default: /tmp/as_built_sync).",
|
|
)
|
|
parser.add_argument(
|
|
"--source-zip",
|
|
default=None,
|
|
help="Path to a ZIP file downloaded from OneDrive to extract and import.",
|
|
)
|
|
parser.add_argument(
|
|
"--rclone-remote",
|
|
default=None,
|
|
help="Optional rclone remote path (e.g. 'onedrive:DPO_TUNNEL_EM_AS-BUILT') to sync before import.",
|
|
)
|
|
parser.add_argument(
|
|
"--structures",
|
|
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",
|
|
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"]
|
|
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"]
|
|
|
|
if dry_run:
|
|
self.stdout.write(self.style.WARNING("DRY RUN — no changes will be committed."))
|
|
|
|
# Traitement d'une archive ZIP si spécifiée
|
|
if source_zip:
|
|
if not os.path.isfile(source_zip):
|
|
raise CommandError(f"ZIP file '{source_zip}' does not exist.")
|
|
self.stdout.write(self.style.NOTICE(f"Extracting ZIP archive '{source_zip}' into '{source_dir}'..."))
|
|
if os.path.exists(source_dir):
|
|
shutil.rmtree(source_dir, ignore_errors=True)
|
|
os.makedirs(source_dir, exist_ok=True)
|
|
with zipfile.ZipFile(source_zip, 'r') as zip_ref:
|
|
zip_ref.extractall(source_dir)
|
|
self.stdout.write(self.style.SUCCESS("ZIP extraction completed."))
|
|
|
|
# Synchronisation via rclone si un remote est spécifié
|
|
elif rclone_remote:
|
|
self.stdout.write(self.style.NOTICE(f"Syncing from rclone remote '{rclone_remote}' to '{source_dir}'..."))
|
|
os.makedirs(source_dir, exist_ok=True)
|
|
rclone_cmd = ["rclone", "copy", rclone_remote, source_dir]
|
|
rclone_conf = os.path.join(os.getcwd(), "rclone.conf")
|
|
if os.path.exists(rclone_conf):
|
|
rclone_cmd.extend(["--config", rclone_conf])
|
|
|
|
res = subprocess.run(rclone_cmd, capture_output=True, text=True)
|
|
if res.returncode != 0:
|
|
raise CommandError(f"rclone sync failed:\n{res.stderr}")
|
|
self.stdout.write(self.style.SUCCESS("rclone sync completed successfully."))
|
|
|
|
if not os.path.isdir(source_dir):
|
|
raise CommandError(f"Source directory '{source_dir}' does not exist or is not a directory.")
|
|
|
|
# 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
|
|
|
|
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)
|
|
if not structures.exists():
|
|
raise CommandError(f"No structures found matching codes: {target_codes}")
|
|
else:
|
|
structures = Structure.objects.all()
|
|
|
|
total_imported = 0
|
|
total_skipped = 0
|
|
|
|
for structure in structures:
|
|
# Check if there is a dedicated subfolder for this structure code, location codes 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 ""),
|
|
]
|
|
|
|
# Also check all associated location codes (e.g. WTB03)
|
|
loc_codes = list(structure.all_locations.values_list("code", flat=True))
|
|
for loc_code in loc_codes:
|
|
possible_dirs.append(os.path.join(source_dir, loc_code))
|
|
if os.path.isdir(source_dir):
|
|
for folder_item in os.listdir(source_dir):
|
|
if folder_item.upper().startswith(loc_code.upper()):
|
|
possible_dirs.append(os.path.join(source_dir, folder_item))
|
|
|
|
for p in possible_dirs:
|
|
if os.path.isdir(p):
|
|
struct_dir = p
|
|
break
|
|
|
|
if not struct_dir and os.path.isdir(source_dir):
|
|
# Check if any subfolder name contains structure code or main keywords from name_fr
|
|
name_words = [w.lower() for w in (structure.name_fr or "").replace("-", " ").replace("_", " ").replace("(", "").replace(")", "").split() if len(w) > 3 and w.lower() not in ["tunnel", "vers", "pont", "viaduc"]]
|
|
for folder_item in os.listdir(source_dir):
|
|
full_p = os.path.join(source_dir, folder_item)
|
|
if os.path.isdir(full_p) and not folder_item.startswith("."):
|
|
folder_lower = folder_item.lower()
|
|
if structure.code.lower() in folder_lower or any(w in folder_lower for w in name_words):
|
|
struct_dir = full_p
|
|
break
|
|
|
|
# If still not found and single structure requested with single directory extracted
|
|
if not struct_dir and target_codes and len(target_codes) == 1:
|
|
subdirs = [os.path.join(source_dir, d) for d in os.listdir(source_dir) if os.path.isdir(os.path.join(source_dir, d)) and not d.startswith(".")]
|
|
if len(subdirs) == 1:
|
|
struct_dir = subdirs[0]
|
|
|
|
# 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("."):
|
|
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 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 file_entries:
|
|
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
|
|
|
|
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()
|
|
|
|
descendant_folder_ids = [f_id for f_id in all_ab_folders if f_id != as_built_folder.pk]
|
|
if descendant_folder_ids:
|
|
descendant_folders = DocumentFolder.objects.filter(pk__in=descendant_folder_ids)
|
|
f_count = descendant_folders.count()
|
|
if f_count > 0:
|
|
self.stdout.write(self.style.WARNING(f" Cleaning {f_count} existing subfolder(s) in As-Built folder..."))
|
|
if not dry_run:
|
|
descendant_folders.delete()
|
|
|
|
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 target_folder.pk and not clean_existing:
|
|
existing_doc = ManagedDocument.objects.filter(
|
|
folders=target_folder,
|
|
title=doc_title,
|
|
is_archived=False
|
|
).first()
|
|
if existing_doc:
|
|
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 folder 'As-Built / {folder_display}'")
|
|
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(target_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}' 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}"))
|