275 lines
10 KiB
Python
275 lines
10 KiB
Python
|
from flask import current_app, session
|
||
|
|
from app_new.extensions import db
|
||
|
|
from app_new.core.models import Alert, Company, Equipment, Part, Intervention, INTERVENTION_STATUSES, User
|
||
|
|
import json
|
||
|
|
from datetime import datetime, timezone, timedelta
|
||
|
|
|
||
|
|
|
||
|
|
def format_datetime(value):
|
||
|
|
"""Formate une date/heure pour l'affichage"""
|
||
|
|
if not value:
|
||
|
|
return ""
|
||
|
|
try:
|
||
|
|
return value.strftime('%d/%m/%Y %H:%M')
|
||
|
|
except:
|
||
|
|
return str(value)
|
||
|
|
|
||
|
|
|
||
|
|
def format_date(value):
|
||
|
|
"""Formate une date pour l'affichage"""
|
||
|
|
if not value:
|
||
|
|
return ""
|
||
|
|
try:
|
||
|
|
return value.strftime('%d/%m/%Y')
|
||
|
|
except:
|
||
|
|
return str(value)
|
||
|
|
|
||
|
|
|
||
|
|
def get_user_session():
|
||
|
|
"""Récupérer l'utilisateur de la session"""
|
||
|
|
if 'user_id' in session:
|
||
|
|
return User.query.get(session['user_id'])
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def requires_permission(user, permission):
|
||
|
|
"""Vérifier si l'utilisateur a une permission"""
|
||
|
|
if not user:
|
||
|
|
return False
|
||
|
|
|
||
|
|
# Vérifier les permissions basées sur le rôle
|
||
|
|
if user.role == 'admin':
|
||
|
|
return True
|
||
|
|
|
||
|
|
if user.role == 'chef' and permission in ['can_create_interventions', 'can_view_statistics']:
|
||
|
|
return True
|
||
|
|
|
||
|
|
if user.role == 'technician' and permission in ['can_create_interventions', 'can_view_equipments']:
|
||
|
|
return True
|
||
|
|
|
||
|
|
if user.role == 'requester' and permission == 'can_create_interventions':
|
||
|
|
return True
|
||
|
|
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def log_action(user, action, data=None):
|
||
|
|
"""Loguer une action utilisateur"""
|
||
|
|
try:
|
||
|
|
# Log simple pour l'instant
|
||
|
|
current_app.logger.info(f"User action: {user.username if user else 'anonymous'} - {action} - {data}")
|
||
|
|
except Exception as e:
|
||
|
|
current_app.logger.error(f"Error logging action: {str(e)}")
|
||
|
|
|
||
|
|
|
||
|
|
def log_user_action(user, action, data=None):
|
||
|
|
"""Loguer une action utilisateur avec plus de détails"""
|
||
|
|
try:
|
||
|
|
log_data = {
|
||
|
|
'user_id': user.id if user else None,
|
||
|
|
'username': user.username if user else 'anonymous',
|
||
|
|
'action': action,
|
||
|
|
'data': data,
|
||
|
|
'timestamp': datetime.now(timezone.utc).isoformat()
|
||
|
|
}
|
||
|
|
current_app.logger.info(f"User action: {json.dumps(log_data)}")
|
||
|
|
except Exception as e:
|
||
|
|
current_app.logger.error(f"Error logging user action: {str(e)}")
|
||
|
|
|
||
|
|
|
||
|
|
def generate_alerts():
|
||
|
|
"""Génère les alertes automatiques : récurrence, stock bas, échéances préventif, workflow."""
|
||
|
|
new_alerts = []
|
||
|
|
|
||
|
|
# 1) Alertes récurrence : 3+ interventions curatives sur un même équipement
|
||
|
|
equipments = Equipment.query.all()
|
||
|
|
for eq in equipments:
|
||
|
|
curative_count = eq.interventions.filter_by(type="curatif").count()
|
||
|
|
existing = Alert.query.filter_by(
|
||
|
|
equipment_id=eq.id, type="recurrence_equipement"
|
||
|
|
).first()
|
||
|
|
|
||
|
|
if curative_count >= 3:
|
||
|
|
msg = (f"⚠️ {eq.name} : {curative_count} interventions curatives. "
|
||
|
|
f"Travaux profonds recommandés.")
|
||
|
|
if existing:
|
||
|
|
existing.message = msg
|
||
|
|
existing.severity = "critical"
|
||
|
|
else:
|
||
|
|
alert = Alert(type="recurrence_equipement", severity="critical",
|
||
|
|
message=msg, equipment_id=eq.id)
|
||
|
|
new_alerts.append(alert)
|
||
|
|
elif existing:
|
||
|
|
db.session.delete(existing)
|
||
|
|
|
||
|
|
# 2) Alertes stock bas
|
||
|
|
parts = Part.query.all()
|
||
|
|
for part in parts:
|
||
|
|
if part.is_low_stock:
|
||
|
|
existing = Alert.query.filter_by(
|
||
|
|
part_id=part.id, type="stock_bas"
|
||
|
|
).first()
|
||
|
|
msg = f"📦 Stock bas : {part.name} — {part.quantity}/{part.min_quantity} {part.unit}"
|
||
|
|
if existing:
|
||
|
|
existing.message = msg
|
||
|
|
existing.severity = "warning"
|
||
|
|
else:
|
||
|
|
alert = Alert(type="stock_bas", severity="warning",
|
||
|
|
message=msg, part_id=part.id)
|
||
|
|
new_alerts.append(alert)
|
||
|
|
|
||
|
|
# 3) Alertes échéances préventif (uniquement si échéance dans les 7 prochains jours)
|
||
|
|
now = datetime.now(timezone.utc)
|
||
|
|
week_later = now + timedelta(days=7)
|
||
|
|
upcoming = Intervention.query.filter(
|
||
|
|
Intervention.type == "preventif",
|
||
|
|
Intervention.is_recurring == True,
|
||
|
|
Intervention.next_due_date != None,
|
||
|
|
Intervention.next_due_date <= week_later,
|
||
|
|
Intervention.status.in_(["brouillon", "en_cours", "attente_chef"]),
|
||
|
|
).all()
|
||
|
|
for interv in upcoming:
|
||
|
|
existing = Alert.query.filter_by(
|
||
|
|
intervention_id=interv.id, type="preventive_echeance"
|
||
|
|
).first()
|
||
|
|
if not existing:
|
||
|
|
msg = (f"📅 Maintenance préventive à échéance : {interv.title} "
|
||
|
|
f"(équipement : {interv.equipment.name})")
|
||
|
|
alert = Alert(type="preventive_echeance", severity="info",
|
||
|
|
message=msg, intervention_id=interv.id,
|
||
|
|
equipment_id=interv.equipment_id)
|
||
|
|
new_alerts.append(alert)
|
||
|
|
|
||
|
|
# 4) Alertes interventions en attente chef depuis +3 jours
|
||
|
|
three_days_ago = datetime.now(timezone.utc) - timedelta(days=3)
|
||
|
|
pending_chef = Intervention.query.filter(
|
||
|
|
Intervention.status == "attente_chef",
|
||
|
|
Intervention.requested_at <= three_days_ago
|
||
|
|
).all()
|
||
|
|
for interv in pending_chef:
|
||
|
|
existing = Alert.query.filter_by(
|
||
|
|
intervention_id=interv.id, type="attente_chef"
|
||
|
|
).first()
|
||
|
|
if not existing:
|
||
|
|
msg = (f"⏳ Intervention #{interv.id} en attente du chef depuis +3 jours : {interv.title}")
|
||
|
|
alert = Alert(type="attente_chef", severity="warning",
|
||
|
|
message=msg, intervention_id=interv.id)
|
||
|
|
new_alerts.append(alert)
|
||
|
|
|
||
|
|
# 5) Alertes interventions en demande GIMA depuis +7 jours
|
||
|
|
seven_days_ago = datetime.now(timezone.utc) - timedelta(days=7)
|
||
|
|
pending_gima = Intervention.query.filter(
|
||
|
|
Intervention.status == "demande_gima",
|
||
|
|
Intervention.updated_at <= seven_days_ago
|
||
|
|
).all()
|
||
|
|
for interv in pending_gima:
|
||
|
|
existing = Alert.query.filter_by(
|
||
|
|
intervention_id=interv.id, type="gima_relance"
|
||
|
|
).first()
|
||
|
|
if not existing:
|
||
|
|
msg = (f"🏛️ Relance GIMA : Intervention #{interv.id} en demande GIMA depuis +7 jours : {interv.title}")
|
||
|
|
alert = Alert(type="gima_relance", severity="info",
|
||
|
|
message=msg, intervention_id=interv.id)
|
||
|
|
new_alerts.append(alert)
|
||
|
|
|
||
|
|
db.session.add_all(new_alerts)
|
||
|
|
db.session.commit()
|
||
|
|
return new_alerts
|
||
|
|
|
||
|
|
|
||
|
|
def get_dashboard_stats():
|
||
|
|
"""Statistiques pour le tableau de bord."""
|
||
|
|
stats = {}
|
||
|
|
stats["interventions_total"] = Intervention.query.count()
|
||
|
|
|
||
|
|
# Compter par statut du nouveau workflow
|
||
|
|
for status_key in INTERVENTION_STATUSES:
|
||
|
|
stats[f"interventions_{status_key}"] = Intervention.query.filter_by(status=status_key).count()
|
||
|
|
|
||
|
|
# Stats rapides pour les cartes
|
||
|
|
stats["interventions_en_attente_chef"] = Intervention.query.filter_by(status="attente_chef").count()
|
||
|
|
stats["interventions_en_cours"] = Intervention.query.filter_by(status="en_cours").count()
|
||
|
|
stats["interventions_demande_gima"] = Intervention.query.filter_by(status="demande_gima").count()
|
||
|
|
stats["interventions_curatif"] = Intervention.query.filter_by(type="curatif").count()
|
||
|
|
stats["interventions_preventif"] = Intervention.query.filter_by(type="preventif").count()
|
||
|
|
stats["interventions_recurrentes"] = Intervention.query.filter_by(is_recurring=True).count()
|
||
|
|
|
||
|
|
# Échéances préventif
|
||
|
|
now = datetime.now(timezone.utc)
|
||
|
|
week_later = now + timedelta(days=7)
|
||
|
|
month_later = now + timedelta(days=30)
|
||
|
|
stats["preventif_overdue"] = Intervention.query.filter(
|
||
|
|
Intervention.is_recurring == True,
|
||
|
|
Intervention.next_due_date < now
|
||
|
|
).count()
|
||
|
|
stats["preventif_this_week"] = Intervention.query.filter(
|
||
|
|
Intervention.is_recurring == True,
|
||
|
|
Intervention.next_due_date >= now,
|
||
|
|
Intervention.next_due_date <= week_later
|
||
|
|
).count()
|
||
|
|
stats["preventif_this_month"] = Intervention.query.filter(
|
||
|
|
Intervention.is_recurring == True,
|
||
|
|
Intervention.next_due_date > week_later,
|
||
|
|
Intervention.next_due_date <= month_later
|
||
|
|
).count()
|
||
|
|
|
||
|
|
stats["equipments_total"] = Equipment.query.count()
|
||
|
|
stats["equipments_hors_service"] = Equipment.query.filter(
|
||
|
|
Equipment.status.in_(["hors_service", "out_of_service"])
|
||
|
|
).count()
|
||
|
|
stats["parts_low_stock"] = Part.query.filter(Part.quantity <= Part.min_quantity).count()
|
||
|
|
stats["alerts_unread"] = Alert.query.filter_by(is_read=False).count()
|
||
|
|
stats["alerts_total"] = Alert.query.count()
|
||
|
|
stats["entreprises_actives"] = Company.query.filter_by(is_active=True).count()
|
||
|
|
return stats
|
||
|
|
|
||
|
|
|
||
|
|
def format_status(status):
|
||
|
|
"""Convertit le statut technique en label lisible."""
|
||
|
|
from app_new.core.models import INTERVENTION_STATUSES, EQUIPMENT_STATUSES
|
||
|
|
if status in INTERVENTION_STATUSES:
|
||
|
|
return INTERVENTION_STATUSES[status]["label"]
|
||
|
|
if status in EQUIPMENT_STATUSES:
|
||
|
|
return EQUIPMENT_STATUSES[status]["label"]
|
||
|
|
# Fallback pour statuts inconnus
|
||
|
|
mapping = {
|
||
|
|
"en_service": "En service",
|
||
|
|
"hors_service": "Hors service",
|
||
|
|
"en_reparation": "En réparation",
|
||
|
|
"remplace": "Remplacé",
|
||
|
|
}
|
||
|
|
return mapping.get(status, status)
|
||
|
|
|
||
|
|
|
||
|
|
def status_badge_class(status):
|
||
|
|
"""Retourne la classe CSS Bootstrap pour un statut."""
|
||
|
|
from app_new.core.models import INTERVENTION_STATUSES, EQUIPMENT_STATUSES
|
||
|
|
if status in INTERVENTION_STATUSES:
|
||
|
|
color = INTERVENTION_STATUSES[status].get("color", "secondary")
|
||
|
|
if color == "purple":
|
||
|
|
return "bg-purple"
|
||
|
|
return f"bg-{color}"
|
||
|
|
if status in EQUIPMENT_STATUSES:
|
||
|
|
color = EQUIPMENT_STATUSES[status].get("color", "secondary")
|
||
|
|
return f"bg-{color}"
|
||
|
|
mapping = {
|
||
|
|
"en_service": "bg-success",
|
||
|
|
"hors_service": "bg-danger",
|
||
|
|
"en_reparation": "bg-warning text-dark",
|
||
|
|
"remplace": "bg-secondary",
|
||
|
|
}
|
||
|
|
return mapping.get(status, "bg-secondary")
|
||
|
|
|
||
|
|
|
||
|
|
def priority_badge_class(priority):
|
||
|
|
"""Retourne la classe CSS Bootstrap pour une priorité."""
|
||
|
|
from app_new.core.models import PRIORITIES
|
||
|
|
if priority in PRIORITIES:
|
||
|
|
return f"bg-{PRIORITIES[priority]['color']}"
|
||
|
|
mapping = {
|
||
|
|
"basse": "bg-light text-dark",
|
||
|
|
"normale": "bg-info",
|
||
|
|
"haute": "bg-warning text-dark",
|
||
|
|
"urgente": "bg-danger",
|
||
|
|
}
|
||
|
|
return mapping.get(priority, "bg-secondary")
|