feat: add management command and service for analyzing and exporting pending interventions and observations
This commit is contained in:
parent
22da9400f3
commit
f72fe1bfd7
4 changed files with 1640 additions and 1 deletions
147
loko/interventions/management/commands/analyze_pending_items.py
Normal file
147
loko/interventions/management/commands/analyze_pending_items.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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',
|
||||
]
|
||||
|
|
|
|||
1254
loko/interventions/services/pending_analysis_service.py
Normal file
1254
loko/interventions/services/pending_analysis_service.py
Normal file
File diff suppressed because it is too large
Load diff
233
loko/interventions/tests/test_pending_analysis.py
Normal file
233
loko/interventions/tests/test_pending_analysis.py
Normal file
|
|
@ -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)
|
||||
Loading…
Reference in a new issue