From f72fe1bfd7dc5bed45c97b5482f34912c36d1861 Mon Sep 17 00:00:00 2001 From: kdeterme Date: Mon, 24 Aug 2026 15:15:17 +0200 Subject: [PATCH] feat: add management command and service for analyzing and exporting pending interventions and observations --- .../commands/analyze_pending_items.py | 147 ++ loko/interventions/services/__init__.py | 7 +- .../services/pending_analysis_service.py | 1254 +++++++++++++++++ .../tests/test_pending_analysis.py | 233 +++ 4 files changed, 1640 insertions(+), 1 deletion(-) create mode 100644 loko/interventions/management/commands/analyze_pending_items.py create mode 100644 loko/interventions/services/pending_analysis_service.py create mode 100644 loko/interventions/tests/test_pending_analysis.py diff --git a/loko/interventions/management/commands/analyze_pending_items.py b/loko/interventions/management/commands/analyze_pending_items.py new file mode 100644 index 0000000..f6af0b6 --- /dev/null +++ b/loko/interventions/management/commands/analyze_pending_items.py @@ -0,0 +1,147 @@ +""" +Commande Django pour analyser et diagnostiquer les observations et interventions non terminées. + +Cette commande extrait les données sur les 24 derniers mois (ou paramètre personnalisé), +analyse les causes de blocage (orphelines, désynchronisations, motifs de pause, retards SLA), +affiche une synthèse en console et génère un rapport Excel (.xlsx) complet multi-onglets. + +Usage: + # Analyse standard sur 24 mois + python manage.py analyze_pending_items + + # Analyse sur 12 mois avec chemin de sortie spécifique + python manage.py analyze_pending_items --months=12 --output=exports/analyse_2026.xlsx + + # Filtrer par thématique ou contrat + python manage.py analyze_pending_items --thematic=lighting --contract=42 +""" + +import os +from datetime import datetime, date +from django.core.management.base import BaseCommand +from django.conf import settings +from django.utils import timezone + +from interventions.services import generate_pending_analysis_excel, get_pending_analysis_data + + +class Command(BaseCommand): + help = "Analyse et diagnostic des observations et interventions non terminées avec export Excel (.xlsx)" + + def add_arguments(self, parser): + parser.add_argument( + '--months', + type=int, + default=24, + help="Nombre de mois d'antériorité à analyser (défaut: 24)", + ) + parser.add_argument( + '--output', + type=str, + default='exports/analyse_encours_loko_%(date)s_%(time)s.xlsx', + help="Nom ou chemin du fichier Excel de sortie (.xlsx)", + ) + parser.add_argument( + '--thematic', + type=str, + default=None, + help="Code de la thématique à filtrer (optionnel)", + ) + parser.add_argument( + '--contract', + type=str, + default=None, + help="Numéro de contrat à filtrer (optionnel)", + ) + parser.add_argument( + '--provider', + type=int, + default=None, + help="ID du prestataire à filtrer (optionnel)", + ) + + def handle(self, *args, **options): + months = options['months'] + output_arg = options['output'] + thematic_code = options['thematic'] + contract_number = options['contract'] + provider_id = options['provider'] + + now = timezone.now() + date_str = now.strftime('%Y%m%d') + time_str = now.strftime('%H%M') + + # Formattage du nom de fichier + output_file = output_arg.replace('%(date)s', date_str).replace('%(time)s', time_str) + if not output_file.endswith('.xlsx'): + output_file += '.xlsx' + + if not os.path.isabs(output_file): + output_file = os.path.join(settings.BASE_DIR, output_file) + + output_dir = os.path.dirname(output_file) + if output_dir and not os.path.exists(output_dir): + os.makedirs(output_dir, exist_ok=True) + + self.stdout.write(self.style.MIGRATE_HEADING("\n" + "=" * 70)) + self.stdout.write(self.style.MIGRATE_HEADING(" LOKO — AUDIT DES OBSERVATIONS & INTERVENTIONS NON TERMINÉES")) + self.stdout.write(self.style.MIGRATE_HEADING("=" * 70)) + self.stdout.write(f"Périmètre : Historique {months} mois | Exécution en lecture seule (Safe Prod)") + if thematic_code: + self.stdout.write(f"Filtre thématique : {thematic_code}") + if contract_number: + self.stdout.write(f"Filtre contrat : {contract_number}") + if provider_id: + self.stdout.write(f"Filtre prestataire ID : {provider_id}") + + self.stdout.write("\nExtraction et analyse des données en cours...") + + # Récupération des données et stats + data = get_pending_analysis_data( + months=months, + thematic_code=thematic_code, + contract_number=contract_number, + provider_id=provider_id, + ) + stats = data['stats'] + + # Affichage Synthèse Console + self.stdout.write(self.style.SUCCESS("\n✓ Analyse terminée. Synthèse des résultats :")) + self.stdout.write("-" * 70) + self.stdout.write(f" • Total Interventions en cours (< 'finished') : {stats['total_interventions']:,}".replace(',', ' ')) + self.stdout.write(f" - Âge moyen de l'encours : {stats['avg_itv_age']} jours (médiane : {stats['median_itv_age']} j)") + self.stdout.write(f" - Interventions en retard SLA / échéance : {stats['delayed_itvs_count']} ({stats['delayed_itvs_pct']}%)") + self.stdout.write(f" - Interventions en pause : {stats['paused_itvs_count']}") + self.stdout.write(f"\n • Total Observations ouvertes : {stats['total_observations']:,}".replace(',', ' ')) + self.stdout.write(f" - Âge moyen : {stats['avg_obs_age']} jours (médiane : {stats['median_obs_age']} j)") + self.stdout.write(f" - Observations orphelines (sans itv liée) : {stats['orphan_obs_count']}") + self.stdout.write(f" - Observations désynchronisées (itv close) : {stats['desynced_obs_count']}") + self.stdout.write("-" * 70) + + # Pyramide des âges synthétique + self.stdout.write("\nPyramide des âges (Interventions en cours) :") + itv_b = stats['itv_aging_buckets'] + self.stdout.write(f" - < 7 jours : {itv_b['less_7d']:>5}") + self.stdout.write(f" - 7 à 30 jours : {itv_b['7_to_30d']:>5}") + self.stdout.write(f" - 30 à 90 jours : {itv_b['30_to_90d']:>5}") + self.stdout.write(f" - 90 à 180 jours: {itv_b['90_to_180d']:>5}") + self.stdout.write(f" - > 180 jours : {itv_b['more_180d']:>5}") + + self.stdout.write("\nGénération du classeur Excel multi-onglets...") + wb = generate_pending_analysis_excel( + months=months, + thematic_code=thematic_code, + contract_number=contract_number, + provider_id=provider_id, + output_format='workbook', + ) + + try: + wb.save(output_file) + file_size_kb = round(os.path.getsize(output_file) / 1024.0, 1) + self.stdout.write(self.style.SUCCESS(f"\n✓ Fichier Excel généré avec succès !")) + self.stdout.write(self.style.SUCCESS(f" -> Chemin : {output_file} ({file_size_kb} Ko)")) + self.stdout.write(self.style.SUCCESS(f" -> Onglets : {', '.join(wb.sheetnames)}")) + except Exception as e: + self.stderr.write(self.style.ERROR(f"Erreur lors de la sauvegarde du fichier Excel : {e}")) + raise diff --git a/loko/interventions/services/__init__.py b/loko/interventions/services/__init__.py index 99c8509..e5a5fd9 100644 --- a/loko/interventions/services/__init__.py +++ b/loko/interventions/services/__init__.py @@ -3,5 +3,10 @@ Services pour le module interventions. """ from .export_service import generate_interventions_excel +from .pending_analysis_service import generate_pending_analysis_excel, get_pending_analysis_data -__all__ = ['generate_interventions_excel'] +__all__ = [ + 'generate_interventions_excel', + 'generate_pending_analysis_excel', + 'get_pending_analysis_data', +] diff --git a/loko/interventions/services/pending_analysis_service.py b/loko/interventions/services/pending_analysis_service.py new file mode 100644 index 0000000..fbe673a --- /dev/null +++ b/loko/interventions/services/pending_analysis_service.py @@ -0,0 +1,1254 @@ +""" +Service pour l'analyse et le diagnostic des observations et interventions non terminées. + +Ce module permet d'extraire, de diagnostiquer et de générer un rapport Excel (.xlsx) +multi-onglets complet pour identifier les causes de blocage et les goulots d'étranglement +sur les 24 derniers mois (ou une période personnalisée). +""" + +import io +import re +from datetime import datetime, date, timedelta +from dateutil.relativedelta import relativedelta + +from django.conf import settings +from django.db.models import Prefetch, Q, Count, Max +from django.utils import timezone +from django.utils.functional import Promise +from django.utils.translation import gettext_lazy as _ + +import openpyxl +from openpyxl.styles import Font, PatternFill, Alignment, Border, Side +from openpyxl.utils import get_column_letter + +# Regex pour éliminer les caractères illégaux pour openpyxl / XML +_ILLEGAL_CHARS_RE = re.compile(r'[\x00-\x08\x0b\x0c\x0e-\x1f]') + + +def _sanitize_cell_value(value): + """Supprime les caractères de contrôle non supportés par Excel.""" + if isinstance(value, Promise): + value = str(value) + if isinstance(value, str): + return _ILLEGAL_CHARS_RE.sub('', value) + return value + + +# Statuts d'intervention considérés comme "non terminés" (avant réalisation terrain 'finished') +PENDING_INTERVENTION_STATUSES = [ + 'in_preparation', + 'to_be_approved', + 'to_be_planned', + 'to_be_processed', + 'assigned', + 'in_progress', + 'on_pause', + 'to_be_corrected', +] + +# Statuts d'observation considérés comme "non terminés" +PENDING_OBSERVATION_STATUSES = [ + 'in_preparation', + 'to_process', + 'in_progress', +] + +# Libellés complets des statuts d'intervention +INTERVENTION_STATUS_LABELS = { + 'in_preparation': "En préparation", + 'to_be_approved': "À approuver", + 'to_be_planned': "À planifier", + 'to_be_processed': "À traiter", + 'assigned': "Pris en charge", + 'in_progress': "Démarré", + 'on_pause': "En pause", + 'to_be_corrected': "À corriger", + 'finished': "Terminé", + 'processed': "Traité", + 'corrected': "Corrigé", + 'validated': "Validé", + 'invoiced': "Comptabilisé", + 'closed': "Finalisé", + 'canceled': "Annulé", +} + +# Libellés des statuts d'observation +OBSERVATION_STATUS_LABELS = { + 'in_preparation': "En préparation", + 'to_process': "À traiter", + 'in_progress': "En cours", + 'processed': "Traité", + 'not_relevant': "Non relevant", + 'to_redirect': "À rediriger", + 'duplicate': "Doublon", + 'closed': "Clôturé", + 'archived': "Archivé", +} + +PRIORITY_LABELS = { + '1': "1 - Urgent", + '2': "2 - Élevé", + '3': "3 - Normal", + '4': "4 - Bas", +} + +PAUSE_REASON_LABELS = { + 'reschedule': "Intervention à replanifier", + 'order_material': "Matériel à commander", + 'treated_in_next_maintenance': "Traité dans le prochain entretien", + 'waiting_for_material_delivery': "En attente livraison de matériel", + 'waiting_for_supplier': "En attente sous-traitant", + 'offer_by_provider': "Offre à faire par le prestataire", + 'waiting_for_validation': "En attente de validation", + 'under_framework_contract': "Sous garantie marché transversal", + 'other': "Autre", +} + + +def get_pending_analysis_data(months=24, thematic_code=None, contract_number=None, provider_id=None): + """ + Extrait et structure toutes les données d'analyse pour les observations et interventions + non terminées sur la période spécifiée. + + Returns: + dict contenant: + - 'start_date': date de début + - 'now': datetime courant + - 'interventions': list de dicts enrichis + - 'observations': list de dicts enrichis + - 'stats': statistiques globales et agrégées + """ + from interventions.models import Intervention, InterventionTimeLine, InterventionDocument + from observations.models import Observation, ObservationPhoto + from common.models import UserConfig + + now = timezone.now() + start_date = now - relativedelta(months=months) + + # ------------------------------------------------------------------------- + # 1. Requête sur les Interventions non terminées + # ------------------------------------------------------------------------- + itv_qs = Intervention.objects.filter( + status__in=PENDING_INTERVENTION_STATUSES, + creation_time__gte=start_date, + ).select_related( + 'thematic', + 'asset_category', + 'symptom', + 'created_by', + 'contract', + 'contract__company', + 'assigned_provider', + 'assigned_team', + 'assigned_member', + 'intervention_manager', + 'source_category', + ).prefetch_related( + Prefetch( + 'events', + queryset=InterventionTimeLine.objects.select_related('event_user').order_by('event_time'), + to_attr='prefetched_events' + ), + ) + + if thematic_code: + itv_qs = itv_qs.filter(thematic__code=thematic_code) + if contract_number: + itv_qs = itv_qs.filter(contract__contract_number=contract_number) + if provider_id: + itv_qs = itv_qs.filter(assigned_provider_id=provider_id) + + itv_list = list(itv_qs.order_by('-creation_time')) + itv_ids = [itv.id for itv in itv_list] + + # Documents count map + doc_counts = {} + if itv_ids: + for row in InterventionDocument.objects.filter(intervention_id__in=itv_ids).values('intervention_id').annotate(cnt=Count('id')): + doc_counts[row['intervention_id']] = row['cnt'] + + # Traitement des interventions + enriched_interventions = [] + for itv in itv_list: + age_days = (now - itv.creation_time).total_seconds() / 86400.0 + + # Récupération des événements timeline + events = getattr(itv, 'prefetched_events', []) + + # Date du dernier événement de changement vers le statut courant + status_entry_date = None + for ev in reversed(events): + if ev.to_status == itv.status: + status_entry_date = ev.event_time + break + + if not status_entry_date: + status_entry_date = itv.creation_time + + days_in_status = (now - status_entry_date).total_seconds() / 86400.0 + + # Date de toute dernière activité timeline + last_activity_date = events[-1].event_time if events else itv.creation_time + days_since_activity = (now - last_activity_date).total_seconds() / 86400.0 + + # Évaluation du statut de retard / SLA + sla_target_date = itv.expected_end_time or itv.planned_end_time + is_delayed = False + delay_days = 0 + sla_status_label = "Dans les délais" + + if sla_target_date: + if now > sla_target_date: + is_delayed = True + delay_days = round((now - sla_target_date).total_seconds() / 86400.0, 1) + sla_status_label = f"En retard (+{delay_days}j)" + else: + # Règle heuristique selon priorité + if itv.priority == '1' and age_days > 2: + is_delayed = True + delay_days = round(age_days - 2, 1) + sla_status_label = f"En retard P1 (+{delay_days}j)" + elif itv.priority == '2' and age_days > 7: + is_delayed = True + delay_days = round(age_days - 7, 1) + sla_status_label = f"En retard P2 (+{delay_days}j)" + elif itv.priority == '3' and age_days > 30: + sla_status_label = "Ancien (>30j)" + elif itv.priority == '4' and age_days > 90: + sla_status_label = "Ancien (>90j)" + else: + sla_status_label = "Non planifié" + + # Diagnostic automatique de la cause de blocage + causes = [] + if itv.status == 'in_preparation': + if age_days > 7: + causes.append(f"Brouillon non finalisé depuis {round(age_days)}j") + else: + causes.append("En préparation récente") + elif itv.status == 'to_be_approved': + manager_name = itv.intervention_manager.get_full_name() if itv.intervention_manager else "Aucun gestionnaire" + causes.append(f"Attente approbation ({manager_name}) depuis {round(days_in_status)}j") + elif itv.status == 'to_be_planned': + causes.append(f"En attente de planification depuis {round(days_in_status)}j") + elif itv.status == 'to_be_processed': + if not itv.assigned_provider and not itv.contract: + causes.append("Blocage : Aucun prestataire ni contrat associé") + elif not itv.assigned_provider: + causes.append("Blocage : Aucun prestataire assigné") + elif days_in_status > 10: + causes.append(f"Non prise en charge par prestataire depuis {round(days_in_status)}j") + else: + causes.append("Transmise au prestataire, en attente de prise en charge") + elif itv.status == 'assigned': + if days_in_status > 7: + causes.append(f"Prise en charge mais non démarrée depuis {round(days_in_status)}j") + else: + causes.append("Prise en charge par le prestataire") + elif itv.status == 'on_pause': + p_label = PAUSE_REASON_LABELS.get(itv.pause_reason, itv.pause_reason or "Non spécifié") + if itv.pause_reason_other: + p_label += f" ({itv.pause_reason_other})" + causes.append(f"En pause [{p_label}] depuis {round(days_in_status)}j") + elif itv.status == 'to_be_corrected': + c_detail = itv.reason_for_correction or "Motif non précisé" + causes.append(f"Correction requise [{c_detail[:60]}] depuis {round(days_in_status)}j") + elif itv.status == 'in_progress': + if days_in_status > 15: + causes.append(f"Chantier en cours prolongé ({round(days_in_status)}j)") + else: + causes.append("Travaux en cours de réalisation") + + if days_since_activity > 30 and itv.status != 'on_pause': + causes.append(f"Inactivité timeline (> {round(days_since_activity)}j sans mise à jour)") + + diagnostic_summary = " | ".join(causes) + + enriched_interventions.append({ + 'id': itv.id, + 'code': itv.code, + 'title': itv.title, + 'thematic_code': itv.thematic.code if itv.thematic else "", + 'thematic_name': itv.thematic.name_fr if itv.thematic else "Sans thématique", + 'asset_category': itv.asset_category.name_fr if itv.asset_category else "", + 'symptom_code': itv.symptom.code if itv.symptom else "", + 'symptom_name': itv.symptom.name_fr if itv.symptom else "", + 'priority': itv.priority, + 'priority_label': PRIORITY_LABELS.get(itv.priority, itv.priority), + 'maintain_type': itv.get_maintain_type_display(), + 'status': itv.status, + 'status_label': INTERVENTION_STATUS_LABELS.get(itv.status, itv.status), + 'creation_time': itv.creation_time, + 'age_days': round(age_days, 1), + 'status_entry_date': status_entry_date, + 'days_in_status': round(days_in_status, 1), + 'planned_begin_time': itv.planned_begin_time, + 'expected_end_time': itv.expected_end_time or itv.planned_end_time, + 'is_delayed': is_delayed, + 'delay_days': delay_days, + 'sla_status_label': sla_status_label, + 'provider_name': itv.assigned_provider.name if itv.assigned_provider else (itv.contract.company.name if itv.contract and itv.contract.company else "Non assigné"), + 'contract_number': itv.contract.contract_number if itv.contract else "Sans contrat", + 'team_name': itv.assigned_team.name if itv.assigned_team else "", + 'manager_name': itv.intervention_manager.get_full_name() if itv.intervention_manager else "", + 'pause_reason': itv.pause_reason or "", + 'pause_reason_label': PAUSE_REASON_LABELS.get(itv.pause_reason, itv.pause_reason or ""), + 'pause_reason_other': itv.pause_reason_other or "", + 'reason_for_correction': itv.reason_for_correction or "", + 'last_activity_date': last_activity_date, + 'days_since_activity': round(days_since_activity, 1), + 'documents_count': doc_counts.get(itv.id, 0), + 'diagnostic': diagnostic_summary, + 'address': itv.address or "", + 'description': itv.description or "", + }) + + # ------------------------------------------------------------------------- + # 2. Requête sur les Observations non terminées + # ------------------------------------------------------------------------- + obs_qs = Observation.objects.filter( + status__in=PENDING_OBSERVATION_STATUSES, + created_at__gte=start_date, + ).select_related( + 'thematic', + 'category', + 'symptom', + 'created_by', + 'intervention', + 'project', + ).prefetch_related( + Prefetch('photos', queryset=ObservationPhoto.objects.only('id', 'observation_id')) + ) + + if thematic_code: + obs_qs = obs_qs.filter(thematic__code=thematic_code) + + obs_list = list(obs_qs.order_by('-created_at')) + + enriched_observations = [] + for obs in obs_list: + age_days = (now - obs.created_at).total_seconds() / 86400.0 + photos_count = len(obs.photos.all()) + + # Déterminer si le créateur est interne ou externe + creator_type = "Non défini" + if obs.created_by: + u_cfg = UserConfig.objects.filter(user=obs.created_by).first() + if u_cfg: + creator_type = "Interne" if u_cfg.is_intern else "Prestataire/Externe" + + # Diagnostic de cause pour l'observation + causes = [] + is_desynced = False + is_orphan = False + + if obs.status == 'in_preparation': + if age_days > 15: + causes.append(f"Brouillon abandonné (Créé il y a {round(age_days)}j)") + else: + causes.append("Brouillon récent") + elif obs.status == 'to_process': + if not obs.intervention_id: + is_orphan = True + if obs.symptom: + if not obs.symptom.auto_create_intervention: + causes.append("En attente de traitement manuel (Symptôme sans auto-création)") + elif not obs.symptom.provider and not obs.symptom.contract: + causes.append("Bloquée : Aucun contrat/prestataire configuré sur le symptôme") + else: + causes.append(f"Orpheline : Non transmise en intervention ({round(age_days)}j)") + else: + causes.append(f"Orpheline : Aucun symptôme défini ({round(age_days)}j)") + else: + causes.append("Désynchronisée : Intervention liée existante mais statut toujours 'À traiter'") + is_desynced = True + elif obs.status == 'in_progress': + if not obs.intervention_id: + causes.append("Anomalie : Statut 'En cours' sans aucune intervention liée") + is_desynced = True + else: + itv_status = obs.intervention.status + if itv_status in ['closed', 'canceled', 'validated', 'finished', 'processed']: + is_desynced = True + causes.append(f"Désynchronisée : Intervention liée {obs.intervention.code} est déjà {INTERVENTION_STATUS_LABELS.get(itv_status, itv_status)}") + else: + causes.append(f"En cours via {obs.intervention.code} ({INTERVENTION_STATUS_LABELS.get(itv_status, itv_status)})") + + if not obs.latitude or not obs.longitude: + causes.append("Coordonnées GPS manquantes") + if photos_count == 0: + causes.append("Aucune photo associée") + + enriched_observations.append({ + 'id': obs.id, + 'code': obs.code or f"OBS{obs.id:05d}", + 'thematic_code': obs.thematic.code if obs.thematic else "", + 'thematic_name': obs.thematic.name_fr if obs.thematic else "Sans thématique", + 'category_name': obs.category.name_fr if obs.category else "", + 'symptom_name': obs.symptom.name_fr if obs.symptom else "", + 'status': obs.status, + 'status_label': OBSERVATION_STATUS_LABELS.get(obs.status, obs.status), + 'created_at': obs.created_at, + 'age_days': round(age_days, 1), + 'created_by': obs.created_by.get_full_name() or obs.created_by.username if obs.created_by else "Système / Inconnu", + 'creator_type': creator_type, + 'address': obs.address or "", + 'latitude': obs.latitude, + 'longitude': obs.longitude, + 'photos_count': photos_count, + 'intervention_code': obs.intervention.code if obs.intervention else "", + 'intervention_status': obs.intervention.status if obs.intervention else "", + 'intervention_status_label': INTERVENTION_STATUS_LABELS.get(obs.intervention.status, obs.intervention.status) if obs.intervention else "", + 'is_desynced': is_desynced, + 'is_orphan': is_orphan, + 'diagnostic': " | ".join(causes), + 'description': obs.description or "", + }) + + # ------------------------------------------------------------------------- + # 3. Statistiques et agrégats + # ------------------------------------------------------------------------- + total_itvs = len(enriched_interventions) + total_obs = len(enriched_observations) + + itv_ages = [i['age_days'] for i in enriched_interventions] + obs_ages = [o['age_days'] for o in enriched_observations] + + def _calc_median(lst): + if not lst: + return 0 + s = sorted(lst) + n = len(s) + mid = n // 2 + return (s[mid] + s[~mid]) / 2.0 + + def _calc_aging_buckets(lst): + return { + 'less_7d': sum(1 for x in lst if x < 7), + '7_to_30d': sum(1 for x in lst if 7 <= x < 30), + '30_to_90d': sum(1 for x in lst if 30 <= x < 90), + '90_to_180d': sum(1 for x in lst if 90 <= x < 180), + 'more_180d': sum(1 for x in lst if x >= 180), + } + + delayed_itvs = sum(1 for i in enriched_interventions if i['is_delayed']) + paused_itvs = sum(1 for i in enriched_interventions if i['status'] == 'on_pause') + orphan_obs = sum(1 for o in enriched_observations if o['is_orphan']) + desynced_obs = sum(1 for o in enriched_observations if o['is_desynced']) + + # Répartition par statut d'intervention + itv_by_status = {} + for st in PENDING_INTERVENTION_STATUSES: + items = [i for i in enriched_interventions if i['status'] == st] + itv_by_status[st] = { + 'label': INTERVENTION_STATUS_LABELS.get(st, st), + 'count': len(items), + 'pct': round((len(items) / total_itvs * 100), 1) if total_itvs else 0, + 'avg_age': round(sum(i['age_days'] for i in items) / len(items), 1) if items else 0, + 'avg_days_in_status': round(sum(i['days_in_status'] for i in items) / len(items), 1) if items else 0, + } + + # Répartition par motif de pause + pause_reasons_stats = {} + for i in enriched_interventions: + if i['status'] == 'on_pause': + pr = i['pause_reason'] or 'non_specifie' + pr_label = i['pause_reason_label'] or "Non spécifié" + if pr not in pause_reasons_stats: + pause_reasons_stats[pr] = {'label': pr_label, 'count': 0, 'durations': []} + pause_reasons_stats[pr]['count'] += 1 + pause_reasons_stats[pr]['durations'].append(i['days_in_status']) + + for pr, data in pause_reasons_stats.items(): + durations = data['durations'] + data['avg_duration'] = round(sum(durations) / len(durations), 1) if durations else 0 + data['max_duration'] = round(max(durations), 1) if durations else 0 + + # Répartition par statut d'observation + obs_by_status = {} + for st in PENDING_OBSERVATION_STATUSES: + items = [o for o in enriched_observations if o['status'] == st] + obs_by_status[st] = { + 'label': OBSERVATION_STATUS_LABELS.get(st, st), + 'count': len(items), + 'pct': round((len(items) / total_obs * 100), 1) if total_obs else 0, + 'avg_age': round(sum(o['age_days'] for o in items) / len(items), 1) if items else 0, + } + + # Répartition par prestataire & contrat + provider_contract_stats = {} + for i in enriched_interventions: + key = (i['provider_name'], i['contract_number']) + if key not in provider_contract_stats: + provider_contract_stats[key] = { + 'provider': i['provider_name'], + 'contract': i['contract_number'], + 'manager': i['manager_name'], + 'total': 0, + 'delayed': 0, + 'by_status': {st: 0 for st in PENDING_INTERVENTION_STATUSES}, + 'ages': [], + } + provider_contract_stats[key]['total'] += 1 + if i['is_delayed']: + provider_contract_stats[key]['delayed'] += 1 + provider_contract_stats[key]['by_status'][i['status']] += 1 + provider_contract_stats[key]['ages'].append(i['age_days']) + + for key, data in provider_contract_stats.items(): + data['delayed_pct'] = round((data['delayed'] / data['total'] * 100), 1) if data['total'] else 0 + data['avg_age'] = round(sum(data['ages']) / len(data['ages']), 1) if data['ages'] else 0 + + stats = { + 'total_interventions': total_itvs, + 'total_observations': total_obs, + 'avg_itv_age': round(sum(itv_ages) / len(itv_ages), 1) if itv_ages else 0, + 'median_itv_age': round(_calc_median(itv_ages), 1), + 'avg_obs_age': round(sum(obs_ages) / len(obs_ages), 1) if obs_ages else 0, + 'median_obs_age': round(_calc_median(obs_ages), 1), + 'delayed_itvs_count': delayed_itvs, + 'delayed_itvs_pct': round((delayed_itvs / total_itvs * 100), 1) if total_itvs else 0, + 'paused_itvs_count': paused_itvs, + 'orphan_obs_count': orphan_obs, + 'desynced_obs_count': desynced_obs, + 'itv_aging_buckets': _calc_aging_buckets(itv_ages), + 'obs_aging_buckets': _calc_aging_buckets(obs_ages), + 'itv_by_status': itv_by_status, + 'obs_by_status': obs_by_status, + 'pause_reasons_stats': pause_reasons_stats, + 'provider_contract_stats': list(provider_contract_stats.values()), + } + + return { + 'months': months, + 'start_date': start_date, + 'now': now, + 'interventions': enriched_interventions, + 'observations': enriched_observations, + 'stats': stats, + } + + +def generate_pending_analysis_excel(months=24, thematic_code=None, contract_number=None, provider_id=None, output_format='workbook'): + """ + Génère le classeur Excel multi-onglets structuré et stylisé. + + Args: + months (int): Horizon d'antériorité en mois (défaut: 24) + thematic_code (str, optional): Code thématique pour filtrer + contract_number (str, optional): Numéro de contrat pour filtrer + provider_id (int, optional): ID du prestataire pour filtrer + output_format (str): 'workbook' pour l'objet Workbook, 'file' pour BytesIO + + Returns: + openpyxl.Workbook ou io.BytesIO + """ + data = get_pending_analysis_data( + months=months, + thematic_code=thematic_code, + contract_number=contract_number, + provider_id=provider_id, + ) + + wb = openpyxl.Workbook() + # Supprimer la feuille par défaut + if "Sheet" in wb.sheetnames: + wb.remove(wb["Sheet"]) + + # ------------------------------------------------------------------------- + # Palette de styles + # ------------------------------------------------------------------------- + font_title = Font(name="Segoe UI", size=15, bold=True, color="FFFFFF") + font_subtitle = Font(name="Segoe UI", size=10, italic=True, color="E2E8F0") + font_section = Font(name="Segoe UI", size=12, bold=True, color="1A365D") + font_card_title = Font(name="Segoe UI", size=9, bold=True, color="4A5568") + font_card_value = Font(name="Segoe UI", size=18, bold=True, color="1A365D") + font_card_value_alert = Font(name="Segoe UI", size=18, bold=True, color="C53030") + font_card_sub = Font(name="Segoe UI", size=8, italic=True, color="718096") + font_header = Font(name="Segoe UI", size=10, bold=True, color="FFFFFF") + font_body = Font(name="Segoe UI", size=9, color="2D3748") + font_body_bold = Font(name="Segoe UI", size=9, bold=True, color="1A365D") + font_alert_red = Font(name="Segoe UI", size=9, bold=True, color="9B2C2C") + font_alert_orange = Font(name="Segoe UI", size=9, bold=True, color="9C4221") + font_alert_green = Font(name="Segoe UI", size=9, bold=True, color="22543D") + + fill_title = PatternFill(start_color="1A365D", end_color="1A365D", fill_type="solid") + fill_header = PatternFill(start_color="2B6CB0", end_color="2B6CB0", fill_type="solid") + fill_sub_header = PatternFill(start_color="4299E1", end_color="4299E1", fill_type="solid") + fill_card = PatternFill(start_color="F7FAFC", end_color="F7FAFC", fill_type="solid") + fill_card_alert = PatternFill(start_color="FFF5F5", end_color="FFF5F5", fill_type="solid") + fill_zebra = PatternFill(start_color="F8FAFC", end_color="F8FAFC", fill_type="solid") + fill_white = PatternFill(start_color="FFFFFF", end_color="FFFFFF", fill_type="solid") + + fill_red_pastel = PatternFill(start_color="FED7D7", end_color="FED7D7", fill_type="solid") + fill_orange_pastel = PatternFill(start_color="FEEBC8", end_color="FEEBC8", fill_type="solid") + fill_green_pastel = PatternFill(start_color="C6F6D5", end_color="C6F6D5", fill_type="solid") + fill_purple_pastel = PatternFill(start_color="E9D8FD", end_color="E9D8FD", fill_type="solid") + + thin_border_side = Side(border_style="thin", color="CBD5E0") + thick_bottom_side = Side(border_style="medium", color="1A365D") + border_cell = Border(left=thin_border_side, right=thin_border_side, top=thin_border_side, bottom=thin_border_side) + border_card = Border(left=thin_border_side, right=thin_border_side, top=thin_border_side, bottom=thick_bottom_side) + + align_center = Alignment(horizontal="center", vertical="center") + align_left = Alignment(horizontal="left", vertical="center") + align_right = Alignment(horizontal="right", vertical="center") + align_header = Alignment(horizontal="center", vertical="center", wrap_text=True) + + stats = data['stats'] + now_str = data['now'].strftime("%d/%m/%Y %H:%M") + start_date_str = data['start_date'].strftime("%d/%m/%Y") + + # ========================================================================= + # ONGLET 1 : SYNTHÈSE & KPI GLOBAUX + # ========================================================================= + ws1 = wb.create_sheet(title="01_Synthese_KPI") + ws1.views.sheetView[0].showGridLines = True + + # Titre bandeau + ws1.merge_cells("A1:I2") + ws1["A1"] = "LOKO — AUDIT ET ANALYSE DES ENCOURS NON TERMINÉS" + ws1["A1"].font = font_title + ws1["A1"].fill = fill_title + ws1["A1"].alignment = Alignment(horizontal="left", vertical="center", indent=1) + + ws1.merge_cells("A3:I3") + ws1["A3"] = f"Périmètre : Historique {months} mois (depuis le {start_date_str}) | Généré le {now_str} | Base de production" + ws1["A3"].font = font_subtitle + ws1["A3"].fill = fill_title + ws1["A3"].alignment = Alignment(horizontal="left", vertical="center", indent=1) + + # Ligne 5-7 : CARTES KPI + def _create_kpi_card(ws, start_col, title, value, sub="", is_alert=False): + c1, c2 = start_col, chr(ord(start_col) + 1) + ws.merge_cells(f"{c1}5:{c2}5") + ws.merge_cells(f"{c1}6:{c2}6") + ws.merge_cells(f"{c1}7:{c2}7") + + ws[f"{c1}5"] = title + ws[f"{c1}5"].font = font_card_title + ws[f"{c1}5"].alignment = align_center + + ws[f"{c1}6"] = value + ws[f"{c1}6"].font = font_card_value_alert if is_alert else font_card_value + ws[f"{c1}6"].alignment = align_center + + ws[f"{c1}7"] = sub + ws[f"{c1}7"].font = font_card_sub + ws[f"{c1}7"].alignment = align_center + + for r in range(5, 8): + for col_letter in [c1, c2]: + cell = ws[f"{col_letter}{r}"] + cell.fill = fill_card_alert if is_alert else fill_card + cell.border = border_card + + _create_kpi_card(ws1, 'A', "INTERVENTIONS EN COURS", f"{stats['total_interventions']:,}".replace(',', ' '), "< Statut 'Terminé'") + _create_kpi_card(ws1, 'C', "OBSERVATIONS OUVERTES", f"{stats['total_observations']:,}".replace(',', ' '), "En prép / À traiter / En cours") + _create_kpi_card(ws1, 'E', "INTERVENTIONS EN RETARD", f"{stats['delayed_itvs_count']:,} ({stats['delayed_itvs_pct']}%)", "Échéance ou SLA dépassé", is_alert=(stats['delayed_itvs_count'] > 0)) + _create_kpi_card(ws1, 'G', "EN PAUSE / ORPHELINES", f"{stats['paused_itvs_count']} itv / {stats['orphan_obs_count']} obs", f"{stats['desynced_obs_count']} obs désynchronisées", is_alert=(stats['orphan_obs_count'] > 0 or stats['desynced_obs_count'] > 0)) + + # Ligne 9 : Section Pyramide des âges + ws1["A9"] = "1. Pyramide des âges de l'encours" + ws1["A9"].font = font_section + + headers_aging = ["Tranche d'âge", "Interventions en cours", "% Interventions", "Observations ouvertes", "% Observations"] + for col_idx, h in enumerate(headers_aging, 1): + c = ws1.cell(row=10, column=col_idx, value=h) + c.font = font_header + c.fill = fill_header + c.alignment = align_header + c.border = border_cell + + itv_buckets = stats['itv_aging_buckets'] + obs_buckets = stats['obs_aging_buckets'] + t_itv = stats['total_interventions'] or 1 + t_obs = stats['total_observations'] or 1 + + aging_rows = [ + ("< 7 jours (Très récent)", itv_buckets['less_7d'], obs_buckets['less_7d']), + ("7 à 30 jours (Récent)", itv_buckets['7_to_30d'], obs_buckets['7_to_30d']), + ("30 à 90 jours (Moyen - 1 à 3 mois)", itv_buckets['30_to_90d'], obs_buckets['30_to_90d']), + ("90 à 180 jours (Ancien - 3 à 6 mois)", itv_buckets['90_to_180d'], obs_buckets['90_to_180d']), + ("> 180 jours (Très ancien > 6 mois)", itv_buckets['more_180d'], obs_buckets['more_180d']), + ] + + r_idx = 11 + for label, i_cnt, o_cnt in aging_rows: + ws1.cell(row=r_idx, column=1, value=label).alignment = align_left + ws1.cell(row=r_idx, column=2, value=i_cnt).alignment = align_right + ws1.cell(row=r_idx, column=3, value=round(i_cnt / t_itv * 100, 1) / 100).number_format = "0.0%" + ws1.cell(row=r_idx, column=4, value=o_cnt).alignment = align_right + ws1.cell(row=r_idx, column=5, value=round(o_cnt / t_obs * 100, 1) / 100).number_format = "0.0%" + + for col_idx in range(1, 6): + cell = ws1.cell(row=r_idx, column=col_idx) + cell.font = font_body + cell.border = border_cell + if r_idx % 2 == 0: + cell.fill = fill_zebra + r_idx += 1 + + # Ligne Total Aging + ws1.cell(row=r_idx, column=1, value="Total général").font = font_body_bold + ws1.cell(row=r_idx, column=2, value=stats['total_interventions']).font = font_body_bold + ws1.cell(row=r_idx, column=3, value=1.0).number_format = "0.0%" + ws1.cell(row=r_idx, column=4, value=stats['total_observations']).font = font_body_bold + ws1.cell(row=r_idx, column=5, value=1.0).number_format = "0.0%" + for col_idx in range(1, 6): + c = ws1.cell(row=r_idx, column=col_idx) + c.border = border_cell + c.fill = fill_card + + # Section 2 : Répartition par Statut Interventions + r_idx += 3 + ws1.cell(row=r_idx, column=1, value="2. Répartition des Interventions par Statut").font = font_section + r_idx += 1 + + headers_itv_st = ["Statut d'intervention", "Nombre", "% du Total", "Âge moyen (j)", "Séjour moyen statut actuel (j)"] + for col_idx, h in enumerate(headers_itv_st, 1): + c = ws1.cell(row=r_idx, column=col_idx, value=h) + c.font = font_header + c.fill = fill_header + c.alignment = align_header + c.border = border_cell + + r_idx += 1 + for st, row_data in stats['itv_by_status'].items(): + ws1.cell(row=r_idx, column=1, value=row_data['label']).alignment = align_left + ws1.cell(row=r_idx, column=2, value=row_data['count']).alignment = align_right + ws1.cell(row=r_idx, column=3, value=row_data['pct'] / 100.0).number_format = "0.0%" + ws1.cell(row=r_idx, column=4, value=row_data['avg_age']).alignment = align_right + ws1.cell(row=r_idx, column=5, value=row_data['avg_days_in_status']).alignment = align_right + + for col_idx in range(1, 6): + cell = ws1.cell(row=r_idx, column=col_idx) + cell.font = font_body + cell.border = border_cell + if r_idx % 2 == 0: + cell.fill = fill_zebra + r_idx += 1 + + # Section 3 : Répartition par Statut Observations + r_idx += 2 + ws1.cell(row=r_idx, column=1, value="3. Répartition des Observations par Statut").font = font_section + r_idx += 1 + + headers_obs_st = ["Statut d'observation", "Nombre", "% du Total", "Âge moyen (jours)"] + for col_idx, h in enumerate(headers_obs_st, 1): + c = ws1.cell(row=r_idx, column=col_idx, value=h) + c.font = font_header + c.fill = fill_header + c.alignment = align_header + c.border = border_cell + + r_idx += 1 + for st, row_data in stats['obs_by_status'].items(): + ws1.cell(row=r_idx, column=1, value=row_data['label']).alignment = align_left + ws1.cell(row=r_idx, column=2, value=row_data['count']).alignment = align_right + ws1.cell(row=r_idx, column=3, value=row_data['pct'] / 100.0).number_format = "0.0%" + ws1.cell(row=r_idx, column=4, value=row_data['avg_age']).alignment = align_right + + for col_idx in range(1, 5): + cell = ws1.cell(row=r_idx, column=col_idx) + cell.font = font_body + cell.border = border_cell + if r_idx % 2 == 0: + cell.fill = fill_zebra + r_idx += 1 + + # Auto-width ws1 + for col in ws1.columns: + max_len = max(len(str(cell.value or '')) for cell in col) + col_letter = get_column_letter(col[0].column) + ws1.column_dimensions[col_letter].width = max(max_len + 4, 12) + ws1.column_dimensions['A'].width = 38 + + # ========================================================================= + # ONGLET 2 : DIAGNOSTICS & BLOCAGES + # ========================================================================= + ws2 = wb.create_sheet(title="02_Diagnostics_Blocages") + ws2.views.sheetView[0].showGridLines = True + + ws2.merge_cells("A1:G2") + ws2["A1"] = "ANALYSE DÉTAILLÉE DES CAUSES DE BLOCAGE ET ANOMALIES" + ws2["A1"].font = font_title + ws2["A1"].fill = fill_title + ws2["A1"].alignment = Alignment(horizontal="left", vertical="center", indent=1) + + # Section 1 : Motifs de pause + ws2["A4"] = "1. Répartition des Interventions en Pause (on_pause)" + ws2["A4"].font = font_section + + headers_pause = ["Code Motif", "Libellé Motif de Pause", "Nb Interventions", "Durée Moyenne en Pause (j)", "Durée Max en Pause (j)"] + for col_idx, h in enumerate(headers_pause, 1): + c = ws2.cell(row=5, column=col_idx, value=h) + c.font = font_header + c.fill = fill_header + c.alignment = align_header + c.border = border_cell + + r_idx = 6 + for pr_code, p_stat in stats['pause_reasons_stats'].items(): + ws2.cell(row=r_idx, column=1, value=pr_code).alignment = align_left + ws2.cell(row=r_idx, column=2, value=p_stat['label']).alignment = align_left + ws2.cell(row=r_idx, column=3, value=p_stat['count']).alignment = align_right + ws2.cell(row=r_idx, column=4, value=p_stat['avg_duration']).alignment = align_right + ws2.cell(row=r_idx, column=5, value=p_stat['max_duration']).alignment = align_right + + for col_idx in range(1, 6): + cell = ws2.cell(row=r_idx, column=col_idx) + cell.font = font_body + cell.border = border_cell + if r_idx % 2 == 1: + cell.fill = fill_zebra + r_idx += 1 + + if not stats['pause_reasons_stats']: + ws2.cell(row=r_idx, column=1, value="Aucune intervention actuellement en pause.").font = font_body + r_idx += 1 + + # Section 2 : Analyse des Observations en Souffrance + r_idx += 2 + ws2.cell(row=r_idx, column=1, value="2. Diagnostic des Observations en Souffrance").font = font_section + r_idx += 1 + + headers_obs_diag = ["Catégorie d'Anomalie / Blocage", "Nombre d'Observations", "Description & Impact", "Action Recommandée"] + for col_idx, h in enumerate(headers_obs_diag, 1): + c = ws2.cell(row=r_idx, column=col_idx, value=h) + c.font = font_header + c.fill = fill_header + c.alignment = align_header + c.border = border_cell + + obs_diag_rows = [ + ( + "Observations Orphelines ('À traiter' sans intervention)", + stats['orphan_obs_count'], + "Observations prêtes pour travaux mais aucune intervention n'a été créée (absence d'auto-création sur symptôme ou rejet).", + "Créer manuellement l'intervention ou vérifier la configuration du symptôme / contrat par défaut." + ), + ( + "Observations Désynchronisées ('En cours' avec intervention close)", + stats['desynced_obs_count'], + "L'observation est restée bloquée en 'En cours' alors que l'intervention associée est terminée, validée ou annulée.", + "Passer l'observation au statut 'Traité' ou 'Clôturé' (Script de resynchronisation)." + ), + ( + "Brouillons anciens (> 15 jours)", + sum(1 for o in data['observations'] if o['status'] == 'in_preparation' and o['age_days'] > 15), + "Observations saisies en brouillon (souvent sur mobile) mais jamais finalisées ni passées à traiter.", + "Contacter les créateurs ou purger/archiver les brouillons obsolètes." + ), + ( + "Observations sans Coordonnées GPS", + sum(1 for o in data['observations'] if not o['latitude'] or not o['longitude']), + "Observations sans position géographique précise, rendant la localisation terrain difficile.", + "Compléter les coordonnées GPS à partir de l'adresse ou via la carte." + ), + ( + "Observations sans Photos", + sum(1 for o in data['observations'] if o['photos_count'] == 0), + "Observations sans cliché descriptif.", + "Sensibiliser les patrouilleurs à l'ajout systématique de photos." + ), + ] + + r_idx += 1 + for cat, cnt, desc, act in obs_diag_rows: + ws2.cell(row=r_idx, column=1, value=cat).alignment = align_left + ws2.cell(row=r_idx, column=2, value=cnt).alignment = align_right + ws2.cell(row=r_idx, column=3, value=desc).alignment = align_left + ws2.cell(row=r_idx, column=4, value=act).alignment = align_left + + for col_idx in range(1, 5): + cell = ws2.cell(row=r_idx, column=col_idx) + cell.font = font_body + cell.border = border_cell + if r_idx % 2 == 1: + cell.fill = fill_zebra + r_idx += 1 + + # Section 3 : Interventions Silencieuses (> 30 jours sans timeline) + r_idx += 2 + ws2.cell(row=r_idx, column=1, value="3. Interventions Silencieuses (> 30 jours sans événement Timeline)").font = font_section + r_idx += 1 + + headers_silent = ["Code Intervention", "Titre", "Thématique", "Statut", "Prestataire", "Jours d'Inactivité", "Diagnostic"] + for col_idx, h in enumerate(headers_silent, 1): + c = ws2.cell(row=r_idx, column=col_idx, value=h) + c.font = font_header + c.fill = fill_header + c.alignment = align_header + c.border = border_cell + + r_idx += 1 + silent_itvs = [i for i in data['interventions'] if i['days_since_activity'] > 30 and i['status'] != 'on_pause'] + for sitv in silent_itvs[:50]: # Top 50 silencieuses + ws2.cell(row=r_idx, column=1, value=sitv['code']).alignment = align_center + ws2.cell(row=r_idx, column=2, value=_sanitize_cell_value(sitv['title'])).alignment = align_left + ws2.cell(row=r_idx, column=3, value=sitv['thematic_name']).alignment = align_left + ws2.cell(row=r_idx, column=4, value=sitv['status_label']).alignment = align_center + ws2.cell(row=r_idx, column=5, value=sitv['provider_name']).alignment = align_left + ws2.cell(row=r_idx, column=6, value=sitv['days_since_activity']).alignment = align_right + ws2.cell(row=r_idx, column=7, value=sitv['diagnostic']).alignment = align_left + + for col_idx in range(1, 8): + cell = ws2.cell(row=r_idx, column=col_idx) + cell.font = font_body + cell.border = border_cell + if r_idx % 2 == 1: + cell.fill = fill_zebra + r_idx += 1 + + if not silent_itvs: + ws2.cell(row=r_idx, column=1, value="Aucune intervention silencieuse détectée.").font = font_body + + # Auto-width ws2 + for col in ws2.columns: + max_len = max(len(str(cell.value or '')) for cell in col) + col_letter = get_column_letter(col[0].column) + ws2.column_dimensions[col_letter].width = min(max(max_len + 4, 12), 60) + + # ========================================================================= + # ONGLET 3 : ANALYSE PRESTATAIRES & CONTRATS + # ========================================================================= + ws3 = wb.create_sheet(title="03_Prestataires_Contrats") + ws3.views.sheetView[0].showGridLines = True + + ws3.merge_cells("A1:K2") + ws3["A1"] = "MATRICE DES ENCOURS PAR PRESTATAIRE ET PAR CONTRAT" + ws3["A1"].font = font_title + ws3["A1"].fill = fill_title + ws3["A1"].alignment = Alignment(horizontal="left", vertical="center", indent=1) + + ws3.merge_cells("A3:K3") + ws3["A3"] = "Vue consolidée des volumes en cours, des taux de retard et des statuts par entreprise partenaire." + ws3["A3"].font = font_subtitle + ws3["A3"].fill = fill_title + ws3["A3"].alignment = Alignment(horizontal="left", vertical="center", indent=1) + + headers_pc = [ + "Prestataire", "Contrat", "Gestionnaire", "Total en cours", + "En prép.", "À approuver", "À planifier", "À traiter", "Pris en charge", "Démarré", "En pause", "À corriger", + "Nb en Retard", "% en Retard", "Âge moyen (j)" + ] + for col_idx, h in enumerate(headers_pc, 1): + c = ws3.cell(row=5, column=col_idx, value=h) + c.font = font_header + c.fill = fill_header + c.alignment = align_header + c.border = border_cell + + r_idx = 6 + pc_data_sorted = sorted(stats['provider_contract_stats'], key=lambda x: x['total'], reverse=True) + for row in pc_data_sorted: + ws3.cell(row=r_idx, column=1, value=row['provider']).alignment = align_left + ws3.cell(row=r_idx, column=2, value=row['contract']).alignment = align_left + ws3.cell(row=r_idx, column=3, value=row['manager'] or "—").alignment = align_left + ws3.cell(row=r_idx, column=4, value=row['total']).alignment = align_right + + # Statuts + by_st = row['by_status'] + ws3.cell(row=r_idx, column=5, value=by_st.get('in_preparation', 0)).alignment = align_right + ws3.cell(row=r_idx, column=6, value=by_st.get('to_be_approved', 0)).alignment = align_right + ws3.cell(row=r_idx, column=7, value=by_st.get('to_be_planned', 0)).alignment = align_right + ws3.cell(row=r_idx, column=8, value=by_st.get('to_be_processed', 0)).alignment = align_right + ws3.cell(row=r_idx, column=9, value=by_st.get('assigned', 0)).alignment = align_right + ws3.cell(row=r_idx, column=10, value=by_st.get('in_progress', 0)).alignment = align_right + ws3.cell(row=r_idx, column=11, value=by_st.get('on_pause', 0)).alignment = align_right + ws3.cell(row=r_idx, column=12, value=by_st.get('to_be_corrected', 0)).alignment = align_right + + # Retard & Age + c_retard = ws3.cell(row=r_idx, column=13, value=row['delayed']) + c_retard.alignment = align_right + if row['delayed'] > 0: + c_retard.font = font_alert_red + + c_retard_pct = ws3.cell(row=r_idx, column=14, value=row['delayed_pct'] / 100.0) + c_retard_pct.number_format = "0.0%" + c_retard_pct.alignment = align_right + if row['delayed_pct'] > 20: + c_retard_pct.font = font_alert_red + + ws3.cell(row=r_idx, column=15, value=row['avg_age']).alignment = align_right + + for col_idx in range(1, 16): + cell = ws3.cell(row=r_idx, column=col_idx) + cell.border = border_cell + if col_idx not in [13, 14]: + cell.font = font_body + if r_idx % 2 == 1: + cell.fill = fill_zebra + r_idx += 1 + + # Total ligne + ws3.cell(row=r_idx, column=1, value="Total général").font = font_body_bold + ws3.cell(row=r_idx, column=4, value=f"=SUM(D6:D{r_idx-1})").font = font_body_bold + ws3.cell(row=r_idx, column=13, value=f"=SUM(M6:M{r_idx-1})").font = font_body_bold + for col_idx in range(1, 16): + c = ws3.cell(row=r_idx, column=col_idx) + c.border = border_cell + c.fill = fill_card + + # Auto-width ws3 + for col in ws3.columns: + max_len = max(len(str(cell.value or '')) for cell in col) + col_letter = get_column_letter(col[0].column) + ws3.column_dimensions[col_letter].width = max(max_len + 3, 11) + ws3.column_dimensions['A'].width = 28 + ws3.column_dimensions['B'].width = 18 + + # ========================================================================= + # ONGLET 4 : DÉTAIL DES OBSERVATIONS + # ========================================================================= + ws4 = wb.create_sheet(title="04_Detail_Observations") + ws4.views.sheetView[0].showGridLines = True + + headers_obs_detail = [ + "Code Obs.", "Thématique", "Catégorie Asset", "Symptôme", "Statut", + "Date Création", "Âge (jours)", "Créé par", "Type Créateur", "Adresse", + "Latitude", "Longitude", "Nb Photos", "Itv Liée (Code)", "Statut Itv Liée", + "Diagnostic Cause de Blocage", "Description" + ] + for col_idx, h in enumerate(headers_obs_detail, 1): + c = ws4.cell(row=1, column=col_idx, value=h) + c.font = font_header + c.fill = fill_header + c.alignment = align_header + c.border = border_cell + + ws4.freeze_panes = "A2" + + r_idx = 2 + for obs in data['observations']: + ws4.cell(row=r_idx, column=1, value=obs['code']).alignment = align_center + ws4.cell(row=r_idx, column=2, value=obs['thematic_name']).alignment = align_left + ws4.cell(row=r_idx, column=3, value=obs['category_name']).alignment = align_left + ws4.cell(row=r_idx, column=4, value=obs['symptom_name']).alignment = align_left + + c_st = ws4.cell(row=r_idx, column=5, value=obs['status_label']) + c_st.alignment = align_center + if obs['status'] == 'to_process': + c_st.fill = fill_orange_pastel + elif obs['status'] == 'in_progress': + c_st.fill = fill_green_pastel + + ws4.cell(row=r_idx, column=6, value=obs['created_at'].strftime("%d/%m/%Y %H:%M")).alignment = align_center + ws4.cell(row=r_idx, column=7, value=obs['age_days']).alignment = align_right + ws4.cell(row=r_idx, column=8, value=obs['created_by']).alignment = align_left + ws4.cell(row=r_idx, column=9, value=obs['creator_type']).alignment = align_center + ws4.cell(row=r_idx, column=10, value=_sanitize_cell_value(obs['address'])).alignment = align_left + ws4.cell(row=r_idx, column=11, value=obs['latitude']).alignment = align_right + ws4.cell(row=r_idx, column=12, value=obs['longitude']).alignment = align_right + ws4.cell(row=r_idx, column=13, value=obs['photos_count']).alignment = align_right + ws4.cell(row=r_idx, column=14, value=obs['intervention_code']).alignment = align_center + ws4.cell(row=r_idx, column=15, value=obs['intervention_status_label']).alignment = align_center + + c_diag = ws4.cell(row=r_idx, column=16, value=obs['diagnostic']) + c_diag.alignment = align_left + if obs['is_desynced'] or obs['is_orphan']: + c_diag.font = font_alert_red + + ws4.cell(row=r_idx, column=17, value=_sanitize_cell_value(obs['description'])).alignment = align_left + + for col_idx in range(1, 18): + cell = ws4.cell(row=r_idx, column=col_idx) + cell.border = border_cell + if col_idx not in [16] and not cell.font: + cell.font = font_body + if r_idx % 2 == 1 and col_idx != 5: + cell.fill = fill_zebra + r_idx += 1 + + ws4.auto_filter.ref = f"A1:Q{max(r_idx-1, 1)}" + + # Widths ws4 + for col in ws4.columns: + max_len = max(len(str(cell.value or '')) for cell in col) + col_letter = get_column_letter(col[0].column) + ws4.column_dimensions[col_letter].width = min(max(max_len + 3, 11), 50) + ws4.column_dimensions['A'].width = 14 + ws4.column_dimensions['J'].width = 25 + ws4.column_dimensions['P'].width = 40 + ws4.column_dimensions['Q'].width = 35 + + # ========================================================================= + # ONGLET 5 : DÉTAIL DES INTERVENTIONS + # ========================================================================= + ws5 = wb.create_sheet(title="05_Detail_Interventions") + ws5.views.sheetView[0].showGridLines = True + + headers_itv_detail = [ + "Code Itv", "Titre", "Thématique", "Catégorie Asset", "Symptôme", "Priorité", + "Type Maintenance", "Statut Actuel", "Date Création", "Âge Total (j)", + "Date Entrée Statut", "Jours dans Statut", "Date Début Prévue", "Date Fin Attendue", + "Statut Retard / SLA", "Prestataire Assigné", "Contrat", "Équipe", "Gestionnaire", + "Motif Pause", "Détail Pause", "Raison Correction", "Dernière Activité", "Jours Inactivité", + "Nb Docs/Photos", "Diagnostic Cause de Blocage", "Adresse", "Description" + ] + for col_idx, h in enumerate(headers_itv_detail, 1): + c = ws5.cell(row=1, column=col_idx, value=h) + c.font = font_header + c.fill = fill_header + c.alignment = align_header + c.border = border_cell + + ws5.freeze_panes = "A2" + + r_idx = 2 + for itv in data['interventions']: + ws5.cell(row=r_idx, column=1, value=itv['code']).alignment = align_center + ws5.cell(row=r_idx, column=2, value=_sanitize_cell_value(itv['title'])).alignment = align_left + ws5.cell(row=r_idx, column=3, value=itv['thematic_name']).alignment = align_left + ws5.cell(row=r_idx, column=4, value=itv['asset_category']).alignment = align_left + ws5.cell(row=r_idx, column=5, value=itv['symptom_name']).alignment = align_left + ws5.cell(row=r_idx, column=6, value=itv['priority_label']).alignment = align_center + ws5.cell(row=r_idx, column=7, value=itv['maintain_type']).alignment = align_center + + c_st = ws5.cell(row=r_idx, column=8, value=itv['status_label']) + c_st.alignment = align_center + if itv['status'] == 'on_pause': + c_st.fill = fill_purple_pastel + elif itv['status'] == 'to_be_processed': + c_st.fill = fill_orange_pastel + elif itv['status'] == 'in_progress': + c_st.fill = fill_green_pastel + + ws5.cell(row=r_idx, column=9, value=itv['creation_time'].strftime("%d/%m/%Y %H:%M")).alignment = align_center + ws5.cell(row=r_idx, column=10, value=itv['age_days']).alignment = align_right + ws5.cell(row=r_idx, column=11, value=itv['status_entry_date'].strftime("%d/%m/%Y %H:%M") if itv['status_entry_date'] else "").alignment = align_center + ws5.cell(row=r_idx, column=12, value=itv['days_in_status']).alignment = align_right + ws5.cell(row=r_idx, column=13, value=itv['planned_begin_time'].strftime("%d/%m/%Y") if itv['planned_begin_time'] else "").alignment = align_center + ws5.cell(row=r_idx, column=14, value=itv['expected_end_time'].strftime("%d/%m/%Y") if itv['expected_end_time'] else "").alignment = align_center + + c_sla = ws5.cell(row=r_idx, column=15, value=itv['sla_status_label']) + c_sla.alignment = align_center + if itv['is_delayed']: + c_sla.fill = fill_red_pastel + c_sla.font = font_alert_red + + ws5.cell(row=r_idx, column=16, value=itv['provider_name']).alignment = align_left + ws5.cell(row=r_idx, column=17, value=itv['contract_number']).alignment = align_left + ws5.cell(row=r_idx, column=18, value=itv['team_name']).alignment = align_left + ws5.cell(row=r_idx, column=19, value=itv['manager_name']).alignment = align_left + ws5.cell(row=r_idx, column=20, value=itv['pause_reason_label']).alignment = align_left + ws5.cell(row=r_idx, column=21, value=_sanitize_cell_value(itv['pause_reason_other'])).alignment = align_left + ws5.cell(row=r_idx, column=22, value=_sanitize_cell_value(itv['reason_for_correction'])).alignment = align_left + ws5.cell(row=r_idx, column=23, value=itv['last_activity_date'].strftime("%d/%m/%Y %H:%M") if itv['last_activity_date'] else "").alignment = align_center + ws5.cell(row=r_idx, column=24, value=itv['days_since_activity']).alignment = align_right + ws5.cell(row=r_idx, column=25, value=itv['documents_count']).alignment = align_right + + c_diag = ws5.cell(row=r_idx, column=26, value=itv['diagnostic']) + c_diag.alignment = align_left + if itv['is_delayed'] or itv['status'] == 'to_be_corrected': + c_diag.font = font_alert_red + + ws5.cell(row=r_idx, column=27, value=_sanitize_cell_value(itv['address'])).alignment = align_left + ws5.cell(row=r_idx, column=28, value=_sanitize_cell_value(itv['description'])).alignment = align_left + + for col_idx in range(1, 29): + cell = ws5.cell(row=r_idx, column=col_idx) + cell.border = border_cell + if col_idx not in [15, 26] and not cell.font: + cell.font = font_body + if r_idx % 2 == 1 and col_idx not in [8, 15]: + cell.fill = fill_zebra + r_idx += 1 + + ws5.auto_filter.ref = f"A1:AB{max(r_idx-1, 1)}" + + # Widths ws5 + for col in ws5.columns: + max_len = max(len(str(cell.value or '')) for cell in col) + col_letter = get_column_letter(col[0].column) + ws5.column_dimensions[col_letter].width = min(max(max_len + 3, 11), 50) + ws5.column_dimensions['A'].width = 14 + ws5.column_dimensions['B'].width = 30 + ws5.column_dimensions['Z'].width = 45 + + # ========================================================================= + # ONGLET 6 : GUIDE & DÉFINITIONS + # ========================================================================= + ws6 = wb.create_sheet(title="06_Definitions_Guide") + ws6.views.sheetView[0].showGridLines = True + + ws6.merge_cells("A1:E2") + ws6["A1"] = "GUIDE MÉTIER, LEXIQUE ET RECOMMANDATIONS DE TRAITEMENT" + ws6["A1"].font = font_title + ws6["A1"].fill = fill_title + ws6["A1"].alignment = Alignment(horizontal="left", vertical="center", indent=1) + + guide_sections = [ + ( + "1. Périmètre de l'Analyse", + [ + ("Historique", f"Toutes les interventions et observations créées sur les {months} derniers mois."), + ("Interventions ciblées", "Interventions dans les statuts actifs avant réalisation terrain : En préparation, À approuver, À planifier, À traiter, Pris en charge, Démarré, En pause, À corriger (hors 'Terminé' et statuts terminaux)."), + ("Observations ciblées", "Observations non clôturées : En préparation, À traiter, En cours (hors 'Traité', 'Clôturé', 'Non relevant', 'Doublon', 'Archivé')."), + ] + ), + ( + "2. Signification des Diagnostics Automatiques (Observations)", + [ + ("Orpheline ('À traiter' sans intervention)", "L'observation est validée mais n'a déclenché aucune intervention. Souvent dû à un symptôme sans auto-création activée, ou sans contrat par défaut."), + ("Désynchronisée ('En cours' avec intervention terminée/annulée)", "L'intervention associée a suivi son cycle de vie jusqu'à la fin, mais le statut de l'observation n'a pas été basculé en 'Traité' ou 'Clôturé'."), + ("Brouillon abandonné", "Observation créée il y a plus de 15 jours en 'En préparation' sans validation ultérieure."), + ] + ), + ( + "3. Signification des Diagnostics Automatiques (Interventions)", + [ + ("Goulot d'approbation", "Intervention en 'À approuver' en attente de visa par le gestionnaire désigné."), + ("Non prise en charge", "Intervention envoyée en 'À traiter' au prestataire sans action de prise en charge après plus de 10 jours."), + ("Blocage en pause", "Intervention arrêtée temporairement pour commande de matériel, sous-traitant, devis, etc. Nécessite un suivi périodique."), + ("Boucle de correction", "Intervention renvoyée au prestataire après contrôle non satisfaisant ('À corriger')."), + ("Intervention silencieuse", "Aucune mise à jour ou note enregistrée dans la timeline depuis plus de 30 jours."), + ] + ), + ( + "4. Pistes d'Actions Recommandées", + [ + ("Resynchronisation automatique", "Exécuter une procédure pour clôturer automatiquement les observations dont l'intervention liée est terminée."), + ("Nettoyage des brouillons", "Relancer ou purger les brouillons anciens de plus de 30 jours."), + ("Revue des motifs de pause", "Organiser des points hebdomadaires sur les interventions en pause > 30 jours avec les prestataires concernés."), + ("Configuration des symptômes", "Vérifier que tous les symptômes usuels disposent d'un contrat et d'une règle d'auto-création appropriée."), + ] + ) + ] + + r_idx = 4 + for sec_title, items in guide_sections: + ws6.cell(row=r_idx, column=1, value=sec_title).font = font_section + r_idx += 1 + for term, defn in items: + c_term = ws6.cell(row=r_idx, column=1, value=term) + c_term.font = font_body_bold + c_term.alignment = align_left + c_term.border = border_cell + c_term.fill = fill_card + + ws6.merge_cells(start_row=r_idx, start_column=2, end_row=r_idx, end_column=5) + c_defn = ws6.cell(row=r_idx, column=2, value=defn) + c_defn.font = font_body + c_defn.alignment = Alignment(horizontal="left", vertical="center", wrap_text=True) + + for col_idx in range(2, 6): + ws6.cell(row=r_idx, column=col_idx).border = border_cell + + r_idx += 1 + r_idx += 1 + + ws6.column_dimensions['A'].width = 35 + ws6.column_dimensions['B'].width = 25 + ws6.column_dimensions['C'].width = 25 + ws6.column_dimensions['D'].width = 25 + ws6.column_dimensions['E'].width = 25 + + if output_format == 'file': + output = io.BytesIO() + wb.save(output) + output.seek(0) + return output + + return wb diff --git a/loko/interventions/tests/test_pending_analysis.py b/loko/interventions/tests/test_pending_analysis.py new file mode 100644 index 0000000..328452f --- /dev/null +++ b/loko/interventions/tests/test_pending_analysis.py @@ -0,0 +1,233 @@ +""" +Tests unitaires pour le service et la commande d'analyse des encours non terminés. +""" + +import os +import tempfile +from datetime import timedelta +from unittest import mock + +from django.contrib.auth import get_user_model +from django.core.management import call_command +from django.test import TestCase +from django.utils import timezone + +import openpyxl + +from common.models import UserConfig, Role, Thematic +from contracts.models import Company, Contract +from assets.models import AssetCategory +from interventions.models import Intervention, InterventionTimeLine, Symptom +from observations.models import Observation +from interventions.services import get_pending_analysis_data, generate_pending_analysis_excel + + +class PendingAnalysisTests(TestCase): + def setUp(self): + User = get_user_model() + self.user = User.objects.create_user(username='test_user', password='password123') + self.user_config = UserConfig.objects.create(user=self.user, is_intern=True) + + self.thematic = Thematic.objects.create(code='lighting', name_fr='Éclairage public', name_nl='Openbare verlichting') + self.category = AssetCategory.objects.create(name_fr='Candélabre', name_nl='Lantaarnpaal', thematic=self.thematic) + self.company = Company.objects.create(name='Prestataire Lumis') + self.contract = Contract.objects.create( + contract_number='CTR-2025-01', + company=self.company, + start_date=timezone.now().date() - timedelta(days=365), + end_date=timezone.now().date() + timedelta(days=365), + is_active=True + ) + self.symptom = Symptom.objects.create( + name_fr='Ampoule grillée', + name_nl='Lamp kapot', + thematic=self.thematic, + contract=self.contract, + provider=self.company, + auto_create_intervention=True, + ) + self.symptom_manual = Symptom.objects.create( + name_fr='Dégradation sans auto-création', + name_nl='Schade zonder auto', + thematic=self.thematic, + auto_create_intervention=False, + ) + + now = timezone.now() + + # 1. Intervention active en cours + self.itv1 = Intervention.objects.create( + title="Réparation luminaire A", + thematic=self.thematic, + asset_category=self.category, + symptom=self.symptom, + status='in_progress', + contract=self.contract, + assigned_provider=self.company, + created_by=self.user, + expected_end_time=now + timedelta(days=2), + priority='3', + ) + InterventionTimeLine.objects.create( + intervention=self.itv1, + event_user=self.user, + event_time=now - timedelta(days=3), + event_type='creation', + to_status='in_progress' + ) + + # 2. Intervention en pause + self.itv2 = Intervention.objects.create( + title="Réparation luminaire B (En pause)", + thematic=self.thematic, + asset_category=self.category, + symptom=self.symptom, + status='on_pause', + pause_reason='order_material', + contract=self.contract, + assigned_provider=self.company, + created_by=self.user, + priority='2', + ) + InterventionTimeLine.objects.create( + intervention=self.itv2, + event_user=self.user, + event_time=now - timedelta(days=10), + event_type='status_change', + to_status='on_pause' + ) + + # 3. Intervention en retard + self.itv3 = Intervention.objects.create( + title="Réparation urgente C (En retard)", + thematic=self.thematic, + asset_category=self.category, + symptom=self.symptom, + status='to_be_processed', + contract=self.contract, + assigned_provider=self.company, + created_by=self.user, + expected_end_time=now - timedelta(days=5), + priority='1', + ) + + # 4. Intervention terminée (ne doit PAS être incluse dans l'analyse des encours) + self.itv_finished = Intervention.objects.create( + title="Réparation terminée D", + thematic=self.thematic, + status='finished', + created_by=self.user, + ) + + # 5. Observation orpheline (symptôme sans auto-création) + self.obs1 = Observation.objects.create( + description="Poteau tordu", + thematic=self.thematic, + category=self.category, + symptom=self.symptom_manual, + status='to_process', + latitude=50.8503, + longitude=4.3517, + created_by=self.user, + ) + + # 6. Observation en cours liée à itv1 + self.obs2 = Observation.objects.create( + description="Lampe clignotante", + thematic=self.thematic, + category=self.category, + symptom=self.symptom, + status='in_progress', + intervention=self.itv1, + latitude=50.8503, + longitude=4.3517, + created_by=self.user, + ) + + # 7. Observation désynchronisée (intervention terminée) + self.obs3 = Observation.objects.create( + description="Lampe éteinte", + thematic=self.thematic, + category=self.category, + status='in_progress', + intervention=self.itv_finished, + latitude=50.8503, + longitude=4.3517, + created_by=self.user, + ) + + # 8. Observation clôturée (ne doit PAS être incluse) + self.obs_closed = Observation.objects.create( + description="Déjà résolu", + status='closed', + latitude=50.8503, + longitude=4.3517, + created_by=self.user, + ) + + def test_get_pending_analysis_data(self): + data = get_pending_analysis_data(months=24) + + # Vérification des volumes + self.assertEqual(len(data['interventions']), 3) # itv1, itv2, itv3 (itv_finished exclue) + self.assertEqual(len(data['observations']), 3) # obs1, obs2, obs3 (obs_closed exclue) + + stats = data['stats'] + self.assertEqual(stats['total_interventions'], 3) + self.assertEqual(stats['total_observations'], 3) + self.assertEqual(stats['paused_itvs_count'], 1) + self.assertEqual(stats['delayed_itvs_count'], 1) + self.assertEqual(stats['orphan_obs_count'], 1) + self.assertEqual(stats['desynced_obs_count'], 1) + + # Vérification de l'observation désynchronisée + obs3_data = next(o for o in data['observations'] if o['id'] == self.obs3.id) + self.assertTrue(obs3_data['is_desynced']) + self.assertIn("Désynchronisée", obs3_data['diagnostic']) + + # Vérification de l'intervention en retard + itv3_data = next(i for i in data['interventions'] if i['id'] == self.itv3.id) + self.assertTrue(itv3_data['is_delayed']) + self.assertIn("En retard", itv3_data['sla_status_label']) + + def test_generate_pending_analysis_excel(self): + wb = generate_pending_analysis_excel(months=24, output_format='workbook') + + # Vérification des 6 feuilles + expected_sheets = [ + "01_Synthese_KPI", + "02_Diagnostics_Blocages", + "03_Prestataires_Contrats", + "04_Detail_Observations", + "05_Detail_Interventions", + "06_Definitions_Guide", + ] + self.assertEqual(wb.sheetnames, expected_sheets) + + # Vérification du contenu de la feuille de synthèse + ws1 = wb["01_Synthese_KPI"] + self.assertIn("LOKO — AUDIT", ws1["A1"].value) + self.assertEqual(ws1["A5"].value, "INTERVENTIONS EN COURS") + + # Vérification du détail des observations (feuille 4) + ws4 = wb["04_Detail_Observations"] + self.assertEqual(ws4["A1"].value, "Code Obs.") + # 3 lignes de données + 1 en-tête = 4 lignes + self.assertEqual(ws4.max_row, 4) + + # Vérification du détail des interventions (feuille 5) + ws5 = wb["05_Detail_Interventions"] + self.assertEqual(ws5["A1"].value, "Code Itv") + self.assertEqual(ws5.max_row, 4) + + def test_analyze_pending_items_command(self): + with tempfile.TemporaryDirectory() as tmp_dir: + out_file = os.path.join(tmp_dir, "test_pending_report.xlsx") + call_command("analyze_pending_items", months=24, output=out_file) + + self.assertTrue(os.path.exists(out_file)) + self.assertGreater(os.path.getsize(out_file), 5000) + + # Recharger le fichier sauvegardé pour vérifier son intégrité + wb = openpyxl.load_workbook(out_file) + self.assertEqual(len(wb.sheetnames), 6)