""" 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]') from django.utils.functional import Promise 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, Promise): value = str(value) 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 = str(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