281 lines
12 KiB
Python
281 lines
12 KiB
Python
import os
|
|
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
|
|
|
|
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."
|
|
)
|
|
|
|
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(
|
|
"--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"]
|
|
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}'..."))
|
|
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
|
|
|
|
# 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 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 process.")
|
|
|
|
for file_path in files_to_import:
|
|
rel_name = os.path.basename(file_path)
|
|
doc_title = os.path.splitext(rel_name)[0]
|
|
|
|
# Deduplication check
|
|
if as_built_folder.pk:
|
|
existing_doc = ManagedDocument.objects.filter(
|
|
folders=as_built_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.")
|
|
total_skipped += 1
|
|
continue
|
|
|
|
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! Imported: {total_imported}, Skipped (already imported): {total_skipped}"))
|