266 lines
10 KiB
Python
266 lines
10 KiB
Python
# interventions/utils.py
|
|
from django.urls import reverse
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
from interventions.models import InterventionLocation
|
|
from assets.models import TrafficLightIntersection
|
|
from assets.models import PublicLightingStreet
|
|
from assets.models import NatureLocation
|
|
from assets.models import StructureLocation
|
|
from assets.models import ControlCenterLocation
|
|
from assets.models import SignStreet
|
|
from assets.models import ITSLocation
|
|
from assets.models import RoadStreet
|
|
from assets.models import ArtworkStreet
|
|
from assets.models import CleanLocation
|
|
from assets.models import NatureRWIZ
|
|
|
|
# Mapping modèle → (route_name, kwarg)
|
|
ROUTES = {
|
|
TrafficLightIntersection: ("assets:trafficlights_intersections_detail", "intersection_id"),
|
|
PublicLightingStreet: ("assets:publiclighting_streets_detail", "street_id"),
|
|
NatureLocation: ("assets:nature_locations_detail", "location_id"),
|
|
StructureLocation: ("assets:structures_locations_detail", "location_id"),
|
|
ControlCenterLocation: ("assets:controlcenters_locations_detail", "location_id"),
|
|
SignStreet: ("assets:sign_streets_detail", "street_id"),
|
|
ITSLocation: ("assets:its_locations_detail", "location_id"),
|
|
RoadStreet: ("assets:roads_streets_detail", "street_id"),
|
|
ArtworkStreet: ("assets:artworks_streets_detail", "street_id"),
|
|
CleanLocation: ("assets:clean_locations_detail", "location_id"),
|
|
NatureRWIZ: ("assets:zirw_detail", "location_id"),
|
|
}
|
|
|
|
|
|
def resolve_intervention_localisations(intervention):
|
|
"""
|
|
Retourne une liste [{"id": <pk InterventionLocation>, "ct_id": ..., "obj_id": ..., "name": ..., "url": ...}, ...]
|
|
à partir des liens génériques InterventionLocation.
|
|
"""
|
|
links = (InterventionLocation.objects
|
|
.filter(intervention=intervention)
|
|
.select_related("content_type"))
|
|
|
|
# Regrouper par ContentType pour charger en lots
|
|
by_ct: dict[int, set[int]] = {}
|
|
link_id_map: dict[tuple[int, int], int] = {}
|
|
for l in links:
|
|
by_ct.setdefault(l.content_type_id, set()).add(l.object_id)
|
|
link_id_map[(l.content_type_id, l.object_id)] = l.id
|
|
if not by_ct:
|
|
return []
|
|
|
|
# content_type_id -> (Model, route_name, kwarg)
|
|
ct_map = {}
|
|
for Model, (route, kwarg) in ROUTES.items():
|
|
ct_id = ContentType.objects.get_for_model(Model).id
|
|
ct_map[ct_id] = (Model, route, kwarg)
|
|
|
|
out = []
|
|
for ct_id, ids in by_ct.items():
|
|
spec = ct_map.get(ct_id)
|
|
if not spec:
|
|
continue
|
|
Model, route_name, kwarg = spec
|
|
for obj in Model.objects.filter(id__in=ids):
|
|
# Tous héritent d'AbstractLocation → get_name() respecte la langue active
|
|
# (fallback FR si besoin).
|
|
code = obj.code
|
|
name = obj.get_name() if hasattr(obj, "get_name") else str(obj)
|
|
url = obj.get_absolute_url() if hasattr(obj, "get_absolute_url") else reverse(route_name, kwargs={kwarg: obj.id})
|
|
out.append({
|
|
"id": link_id_map.get((ct_id, obj.id)),
|
|
"ct_id": ct_id,
|
|
"obj_id": obj.id,
|
|
"name": f"{code} - {name}",
|
|
"url": url,
|
|
})
|
|
|
|
# Tri par nom pour un rendu stable
|
|
out.sort(key=lambda d: (d["name"] or "").lower())
|
|
return out
|
|
|
|
|
|
def create_preventive_intervention_report(intervention, creator_user=None):
|
|
"""
|
|
Ensures standard structure subfolders are created for any Structure related to the
|
|
given preventive intervention, generates a PDF report, and attaches it.
|
|
|
|
Only applies to interventions with:
|
|
- maintain_type = "preventive"
|
|
- status = "validated"
|
|
- effective_date >= 2026-01-01
|
|
"""
|
|
from django.db import transaction
|
|
from django.db.models import Q
|
|
from django.utils import timezone
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.files.base import ContentFile
|
|
from django.utils.text import slugify
|
|
from django.utils.translation import activate
|
|
from datetime import datetime
|
|
import io
|
|
|
|
from assets.models.structures import Structure, StructureLocation, StructureGeoAsset
|
|
from documents.models import DocumentFolder, DocumentFolderAttachment, ManagedDocument, DocumentVersion, DocumentAttachment
|
|
from interventions.views.views_pdf_reporting import generate_intervention_report_pdf
|
|
|
|
# 1. Verification of preconditions
|
|
if intervention.maintain_type != "preventive":
|
|
return
|
|
|
|
if intervention.status != "validated":
|
|
return
|
|
|
|
# Check date: threshold 2026-01-01
|
|
tz = timezone.get_current_timezone()
|
|
start_date = timezone.make_aware(datetime(2026, 1, 1), tz)
|
|
|
|
effective_date = (
|
|
intervention.begin_time or
|
|
intervention.planned_begin_time or
|
|
intervention.expected_begin_time or
|
|
intervention.creation_time or
|
|
timezone.now()
|
|
)
|
|
if effective_date < start_date:
|
|
return
|
|
|
|
# Find structures related to this intervention
|
|
structure_ct = ContentType.objects.get_for_model(Structure)
|
|
structure_loc_ct = ContentType.objects.get_for_model(StructureLocation)
|
|
structure_asset_ct = ContentType.objects.get_for_model(StructureGeoAsset)
|
|
intervention_ct = ContentType.objects.get_for_model(intervention.__class__)
|
|
|
|
loc_ids = intervention.locations.filter(content_type=structure_loc_ct).values_list("object_id", flat=True)
|
|
asset_ids_1 = intervention.interventionasset_set.filter(content_type=structure_asset_ct).values_list("object_id", flat=True)
|
|
asset_ids_2 = intervention.operations.filter(asset_content_type=structure_asset_ct).values_list("asset_object_id", flat=True)
|
|
|
|
structures = Structure.objects.filter(
|
|
Q(all_locations__id__in=loc_ids) |
|
|
Q(assets__id__in=list(asset_ids_1) + list(asset_ids_2))
|
|
).distinct()
|
|
|
|
if not structures.exists():
|
|
return
|
|
|
|
# Select creator user
|
|
if not creator_user:
|
|
from django.contrib.auth import get_user_model
|
|
User = get_user_model()
|
|
creator_user = User.objects.filter(username="kdeterme").first() or User.objects.filter(is_superuser=True).first()
|
|
|
|
# Activate French language for generating standard French PDF reports
|
|
activate("fr")
|
|
|
|
title = f"Intervention report {intervention.code}"
|
|
filename = f"intervention_report_{intervention.code}.pdf"
|
|
|
|
for structure in structures:
|
|
# 1. Retrieve the structure's default root folder
|
|
root_attachment = DocumentFolderAttachment.objects.filter(
|
|
content_type=structure_ct,
|
|
object_id=structure.pk,
|
|
is_default=True
|
|
).select_related("folder").first()
|
|
|
|
if not root_attachment:
|
|
continue
|
|
|
|
root_folder = root_attachment.folder
|
|
|
|
# 2. Helper to retrieve/create subfolders safely
|
|
def get_or_create_subfolder(parent, name):
|
|
base_slug = f"{parent.slug}-{slugify(name)}"
|
|
slug = base_slug[:150]
|
|
|
|
# Try to get existing folder under parent
|
|
folder = parent.child_folders.filter(name=name).first()
|
|
if not folder:
|
|
# Try by slug
|
|
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
|
|
|
|
folder, created = DocumentFolder.objects.get_or_create(
|
|
slug=slug,
|
|
defaults={
|
|
"name": name,
|
|
"application_label": parent.application_label,
|
|
"created_by": creator_user
|
|
}
|
|
)
|
|
if created:
|
|
folder.thematics.set(parent.thematics.all())
|
|
folder.parent_folders.add(parent)
|
|
return folder
|
|
|
|
civil_eng_folder = get_or_create_subfolder(root_folder, "Civil engineering")
|
|
get_or_create_subfolder(civil_eng_folder, "Inspection reports")
|
|
get_or_create_subfolder(civil_eng_folder, "Exceptional transport")
|
|
|
|
electromec_folder = get_or_create_subfolder(root_folder, "Electromechanics")
|
|
prev_maint_folder = get_or_create_subfolder(electromec_folder, "Preventive maintenance")
|
|
|
|
# 3. Check if report already exists for this structure and intervention
|
|
report_exists = ManagedDocument.objects.filter(
|
|
attachments__content_type=intervention_ct,
|
|
attachments__object_id=intervention.pk,
|
|
folders=prev_maint_folder,
|
|
title=title
|
|
).exists()
|
|
|
|
if report_exists:
|
|
continue
|
|
|
|
# 4. Generate and save the PDF report
|
|
buffer = io.BytesIO()
|
|
generate_intervention_report_pdf(intervention, buffer)
|
|
buffer.seek(0)
|
|
pdf_data = buffer.read()
|
|
|
|
# Save using ManagedDocument & DocumentVersion
|
|
with transaction.atomic():
|
|
doc = ManagedDocument.objects.create(
|
|
title=title,
|
|
description=f"Preventive intervention report {intervention.code} of {effective_date.strftime('%d/%m/%Y')}",
|
|
created_by=creator_user,
|
|
)
|
|
|
|
version = DocumentVersion(
|
|
document=doc,
|
|
uploaded_by=creator_user,
|
|
comment="Automatically generated by maintenance system",
|
|
status=DocumentVersion.STATUS_APPROVED,
|
|
validated_at=timezone.now(),
|
|
)
|
|
version.file.save(filename, ContentFile(pdf_data), save=True)
|
|
|
|
doc.folders.add(prev_maint_folder)
|
|
|
|
# Create attachment to the structure
|
|
DocumentAttachment.objects.create(
|
|
document=doc,
|
|
content_type=structure_ct,
|
|
object_id=structure.pk,
|
|
attached_by=creator_user
|
|
)
|
|
|
|
# Ensure it's not placed in the root folder by default_folder_link trigger
|
|
if doc.folders.filter(pk=root_folder.pk).exists():
|
|
doc.folders.remove(root_folder)
|
|
|
|
# Create attachment to the intervention itself
|
|
DocumentAttachment.objects.create(
|
|
document=doc,
|
|
content_type=intervention_ct,
|
|
object_id=intervention.pk,
|
|
attached_by=creator_user
|
|
)
|
|
|
|
doc.title = title
|
|
doc.save(update_fields=["title"])
|
|
|