366 lines
16 KiB
Python
366 lines
16 KiB
Python
"""
|
|
Service pour l'export des interventions au format Excel.
|
|
Ce module centralise la logique d'export pour éviter la duplication de code
|
|
entre la commande management et l'interface admin.
|
|
"""
|
|
|
|
from django.db.models import Prefetch, Q
|
|
from django.utils.translation import override
|
|
from django.utils import timezone
|
|
from datetime import datetime, date
|
|
from dateutil.relativedelta import relativedelta
|
|
import openpyxl
|
|
from openpyxl.styles import Font, PatternFill, Alignment
|
|
from openpyxl.utils.exceptions import IllegalCharacterError
|
|
import re
|
|
import io
|
|
|
|
# Regex pour les caractères illégaux dans les cellules Excel (contrôle ASCII hors \t, \n, \r)
|
|
_ILLEGAL_CHARS_RE = re.compile(r'[\x00-\x08\x0b\x0c\x0e-\x1f]')
|
|
|
|
|
|
def _sanitize_cell_value(value):
|
|
"""Supprime les caractères illégaux d'une valeur avant de l'écrire dans une cellule Excel."""
|
|
if isinstance(value, str):
|
|
return _ILLEGAL_CHARS_RE.sub('', value)
|
|
return value
|
|
|
|
|
|
def generate_interventions_excel(start_date=None, end_date=None, output_format='file'):
|
|
"""
|
|
Génère un export Excel des interventions selon les critères spécifiés.
|
|
|
|
Args:
|
|
start_date (date, optional): Date de début du filtrage. Si None, utilise le premier jour du mois précédent.
|
|
end_date (date, optional): Date de fin du filtrage. Si None, utilise le dernier jour du mois précédent.
|
|
output_format (str): Format de sortie - 'file' pour BytesIO (utilisé par admin),
|
|
'workbook' pour retourner le workbook directement (utilisé par la commande)
|
|
|
|
Returns:
|
|
io.BytesIO ou openpyxl.Workbook: Selon le paramètre output_format
|
|
"""
|
|
from interventions.models import Intervention, InterventionAsset, InterventionLocation, InterventionDocument, InterventionNote
|
|
from contracts.models import CompanyMember
|
|
from common.models import Municipality
|
|
from django.db.models import ForeignKey
|
|
|
|
# Si aucune date n'est fournie, utiliser le mois précédent par défaut
|
|
if not start_date and not end_date:
|
|
today = date.today()
|
|
start_date = (today.replace(day=1) - relativedelta(months=1))
|
|
end_date = today.replace(day=1) - relativedelta(days=1)
|
|
|
|
# Construction de la requête
|
|
queryset = Intervention.objects.select_related(
|
|
'symptom',
|
|
'contract',
|
|
'contract__company',
|
|
'source_category',
|
|
'created_by',
|
|
'thematic',
|
|
'asset_category'
|
|
).prefetch_related(
|
|
Prefetch(
|
|
'notes',
|
|
queryset=InterventionNote.objects.select_related('note_author').order_by('id')
|
|
),
|
|
Prefetch(
|
|
'interventionasset_set',
|
|
queryset=InterventionAsset.objects.select_related('content_type').order_by('id')
|
|
),
|
|
Prefetch(
|
|
'locations',
|
|
queryset=InterventionLocation.objects.select_related('content_type').order_by('id')
|
|
),
|
|
)
|
|
|
|
# Filtrage par type de maintenance (uniquement correctives)
|
|
queryset = queryset.filter(maintain_type='corrective', created_by__isnull=False)
|
|
|
|
# Filtrage par dates si spécifiées
|
|
if start_date:
|
|
queryset = queryset.filter(creation_time__date__gte=start_date)
|
|
# begin_time doit être nulle ou >= start_date
|
|
queryset = queryset.filter(
|
|
Q(begin_time__isnull=True) | Q(begin_time__date__gte=start_date)
|
|
)
|
|
|
|
if end_date:
|
|
queryset = queryset.filter(creation_time__date__lte=end_date)
|
|
|
|
queryset = queryset.order_by('-creation_time')
|
|
|
|
# Pré-calculer le mapping intervention_id -> noms des documents
|
|
intervention_ids = list(queryset.values_list('id', flat=True))
|
|
docs_map = {}
|
|
for doc in InterventionDocument.objects.filter(intervention_id__in=intervention_ids).only('intervention_id', 'file').order_by('id'):
|
|
docs_map.setdefault(doc.intervention_id, []).append(doc.filename)
|
|
|
|
# Pré-calculer le mapping user_id -> noms d'équipes pour éviter N+1 queries
|
|
user_ids = queryset.values_list('created_by_id', flat=True).distinct()
|
|
user_team_map = {}
|
|
for member in CompanyMember.objects.filter(user_id__in=user_ids).prefetch_related('teams'):
|
|
for team in member.teams.all():
|
|
user_team_map.setdefault(member.user_id, []).append(team.name)
|
|
|
|
# Créer le workbook Excel
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
ws.title = "Interventions"
|
|
|
|
# Définir les en-têtes et les largeurs de colonne de manière dynamique
|
|
column_definitions = [
|
|
('Code', 15),
|
|
('Titre', 30),
|
|
('Date de création', 20),
|
|
('Date de transmission', 20),
|
|
('Date début prévue', 20),
|
|
('Date fin prévue', 20),
|
|
('Date début planifiée', 20),
|
|
('Date fin planifiée', 20),
|
|
('Date de début', 20),
|
|
('Date de fin', 20),
|
|
('Date appel initial confirmé', 20),
|
|
('Type d\'origine', 15),
|
|
('Priorité', 12),
|
|
('Statut', 15),
|
|
('Symptôme', 30),
|
|
('Contrat', 30),
|
|
('Source catégorie', 20),
|
|
]
|
|
|
|
from common.hooks import hooks
|
|
for header in hooks.get_export_headers():
|
|
column_definitions.append((header, 15))
|
|
|
|
column_definitions.extend([
|
|
('Documents joints', 50),
|
|
('Notes', 50),
|
|
('Créé par', 20),
|
|
('Équipe (créateur)', 25),
|
|
('Thématique (code)', 15),
|
|
('Thématique (nom)', 25),
|
|
('Commune', 20),
|
|
('Localisation code 1', 15),
|
|
('Localisation label 1', 25),
|
|
('Localisation code 2', 15),
|
|
('Localisation label 2', 25),
|
|
('Localisation code 3', 15),
|
|
('Localisation label 3', 25),
|
|
('Localisation code 4', 15),
|
|
('Localisation label 4', 25),
|
|
('Equipement code', 15),
|
|
('Equipement nom', 25),
|
|
('Modèle code', 15),
|
|
('Modèle nom', 30),
|
|
])
|
|
|
|
headers = [col[0] for col in column_definitions]
|
|
|
|
# Écrire les en-têtes avec style
|
|
for col_num, header in enumerate(headers, 1):
|
|
cell = ws.cell(row=1, column=col_num)
|
|
cell.value = header
|
|
cell.font = Font(bold=True, color="FFFFFF")
|
|
cell.fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")
|
|
cell.alignment = Alignment(horizontal="center", vertical="center")
|
|
|
|
# Ajuster la largeur des colonnes
|
|
from openpyxl.utils import get_column_letter
|
|
for i, col_def in enumerate(column_definitions, start=1):
|
|
col_letter = get_column_letter(i)
|
|
ws.column_dimensions[col_letter].width = col_def[1]
|
|
|
|
# Forcer la langue française pour l'export
|
|
with override('fr'):
|
|
row_num = 2
|
|
for intervention in queryset:
|
|
# Obtenir le nom du symptôme (en français)
|
|
symptom_name = intervention.symptom.name_fr if intervention.symptom else ''
|
|
|
|
# Obtenir les informations du contrat
|
|
contrat_info = ''
|
|
if intervention.contract:
|
|
provider_name = intervention.contract.company.name if intervention.contract.company else ''
|
|
contract_code = intervention.contract.contract_number or ''
|
|
contrat_info = f"{provider_name} - {contract_code}" if provider_name or contract_code else ''
|
|
|
|
# Obtenir la source catégorie (en français)
|
|
source_category_name = intervention.source_category.name_fr if intervention.source_category else ''
|
|
|
|
# Formater les notes (date, auteur, contenu)
|
|
notes_formatted = []
|
|
for note in intervention.notes.all():
|
|
note_date = note.note_time.strftime('%Y-%m-%d %H:%M') if note.note_time else ''
|
|
note_author = note.note_author.get_full_name() if note.note_author else note.note_author.username if note.note_author else 'Anonyme'
|
|
note_content = note.content or ''
|
|
notes_formatted.append(f"{note_date}, {note_author}, {note_content}")
|
|
notes_str = '\n'.join(notes_formatted)
|
|
|
|
# Obtenir le créateur
|
|
created_by = ''
|
|
if intervention.created_by:
|
|
created_by = intervention.created_by.get_full_name() or intervention.created_by.username
|
|
|
|
# Obtenir l'équipe du créateur via la table pré-calculée
|
|
creator_team = ', '.join(user_team_map.get(intervention.created_by_id, []))
|
|
|
|
# Obtenir les noms des documents joints
|
|
documents_str = '\n'.join(docs_map.get(intervention.id, []))
|
|
|
|
# Obtenir la thématique
|
|
thematic_code = intervention.thematic.code if intervention.thematic else ''
|
|
thematic_name = intervention.thematic.name if intervention.thematic else ''
|
|
|
|
# Obtenir la catégorie d'équipement (en français)
|
|
category_code = intervention.asset_category.code if intervention.asset_category else ''
|
|
category_name = intervention.asset_category.name_fr if intervention.asset_category else ''
|
|
|
|
# Obtenir le modèle du premier équipement (en français)
|
|
model_code = ''
|
|
model_name = ''
|
|
first_asset = intervention.interventionasset_set.first()
|
|
if first_asset and first_asset.content_type.model_class() is not None:
|
|
try:
|
|
asset_obj = first_asset.asset
|
|
except Exception:
|
|
asset_obj = None
|
|
if asset_obj:
|
|
if hasattr(asset_obj, 'model') and asset_obj.model:
|
|
model_code = asset_obj.model.code or ''
|
|
model_name = asset_obj.model.name_fr
|
|
|
|
# Obtenir les localisations et remonter la hiérarchie
|
|
location_codes = ['', '', '', '']
|
|
location_labels = ['', '', '', '']
|
|
municipality_name = ''
|
|
|
|
first_location = intervention.locations.first()
|
|
if first_location and first_location.content_type.model_class() is not None:
|
|
try:
|
|
location_obj = first_location.location
|
|
except Exception:
|
|
location_obj = None
|
|
if location_obj:
|
|
# Construire la hiérarchie des localisations
|
|
location_hierarchy = []
|
|
current_loc = location_obj
|
|
max_depth = 10 # Limite pour éviter les boucles infinies
|
|
depth = 0
|
|
|
|
while current_loc and depth < max_depth:
|
|
location_hierarchy.append(current_loc)
|
|
|
|
# Vérifier de manière robuste si cette localisation a une ForeignKey vers Municipality
|
|
if not municipality_name:
|
|
for field in current_loc._meta.get_fields():
|
|
if isinstance(field, ForeignKey):
|
|
# Vérifier si c'est une FK vers Municipality
|
|
if field.related_model == Municipality:
|
|
municipality_obj = getattr(current_loc, field.name, None)
|
|
if municipality_obj:
|
|
municipality_name = str(municipality_obj)
|
|
break
|
|
|
|
# Vérifier si la localisation a un parent
|
|
if hasattr(current_loc, 'parent_location') and current_loc.parent_location:
|
|
current_loc = current_loc.parent_location
|
|
else:
|
|
break
|
|
depth += 1
|
|
|
|
# Inverser pour avoir l'ancêtre en premier
|
|
location_hierarchy.reverse()
|
|
|
|
# Remplir les codes et labels (max 4 niveaux, en français)
|
|
for i, loc in enumerate(location_hierarchy[:4]):
|
|
location_codes[i] = loc.code if hasattr(loc, 'code') else ''
|
|
# Obtenir le nom en français
|
|
if hasattr(loc, 'name_fr'):
|
|
location_labels[i] = loc.name_fr
|
|
elif hasattr(loc, 'get_name'):
|
|
location_labels[i] = loc.get_name()
|
|
elif hasattr(loc, 'name'):
|
|
location_labels[i] = loc.name
|
|
else:
|
|
location_labels[i] = str(loc)
|
|
|
|
# Obtenir les statuts et priorités traduits en français
|
|
status_display = intervention.get_status_display()
|
|
priority_display = intervention.get_priority_display()
|
|
origin_type_display = intervention.get_origin_type_display() if intervention.origin_type else ''
|
|
|
|
# Convertir les dates timezone-aware en dates naïves pour Excel
|
|
def make_naive(dt):
|
|
"""Convertit un datetime timezone-aware en datetime naïf (sans timezone)"""
|
|
if dt and hasattr(dt, 'tzinfo') and dt.tzinfo is not None:
|
|
return timezone.localtime(dt).replace(tzinfo=None)
|
|
return dt
|
|
|
|
# Écrire la ligne
|
|
row_data = [
|
|
intervention.code,
|
|
intervention.title,
|
|
make_naive(intervention.creation_time),
|
|
make_naive(intervention.transmission_time),
|
|
make_naive(intervention.expected_begin_time),
|
|
make_naive(intervention.expected_end_time),
|
|
make_naive(intervention.planned_begin_time),
|
|
make_naive(intervention.planned_end_time),
|
|
make_naive(intervention.begin_time),
|
|
make_naive(intervention.end_time),
|
|
make_naive(intervention.init_call_confirmed_time),
|
|
origin_type_display,
|
|
priority_display,
|
|
status_display,
|
|
symptom_name,
|
|
contrat_info,
|
|
source_category_name,
|
|
*hooks.extract_export_row(intervention),
|
|
documents_str,
|
|
notes_str,
|
|
created_by,
|
|
creator_team,
|
|
thematic_code,
|
|
thematic_name,
|
|
municipality_name,
|
|
location_codes[0],
|
|
location_labels[0],
|
|
location_codes[1],
|
|
location_labels[1],
|
|
location_codes[2],
|
|
location_labels[2],
|
|
location_codes[3],
|
|
location_labels[3],
|
|
category_code,
|
|
category_name,
|
|
model_code,
|
|
model_name,
|
|
]
|
|
|
|
for col_num, value in enumerate(row_data, 1):
|
|
cell = ws.cell(row=row_num, column=col_num)
|
|
try:
|
|
cell.value = _sanitize_cell_value(value)
|
|
except IllegalCharacterError:
|
|
cell.value = _ILLEGAL_CHARS_RE.sub('', str(value)) if value is not None else ''
|
|
cell.alignment = Alignment(vertical="top", wrap_text=True)
|
|
|
|
row_num += 1
|
|
|
|
# Activer les filtres automatiques sur toute la plage de données
|
|
ws.auto_filter.ref = ws.dimensions
|
|
|
|
# Figer la première ligne
|
|
ws.freeze_panes = 'A2'
|
|
|
|
# Retourner selon le format demandé
|
|
if output_format == 'file':
|
|
# Pour l'admin : retourner un BytesIO
|
|
output = io.BytesIO()
|
|
wb.save(output)
|
|
output.seek(0)
|
|
return output
|
|
else:
|
|
# Pour la commande : retourner le workbook directement
|
|
return wb
|