gmao/app_new/core/models/maintenance.py
root d18deec5d5
Some checks are pending
CI - Tests et Syntax / lint-and-test (push) Waiting to run
Lier les tâches préventives aux contrats
2026-08-20 22:28:47 +00:00

509 lines
21 KiB
Python

"""
Modèle Maintenance - GMAO Collège
Gestion des interventions et maintenance
Compatible avec la base de données existante
"""
from datetime import datetime, timezone
from ...extensions import db
class Intervention(db.Model):
"""Intervention de maintenance - compatible avec la base existante."""
__tablename__ = "interventions"
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(300), nullable=False)
description = db.Column(db.Text, default="")
# Type : curatif (panne) ou préventif (maintenance planifiée)
type = db.Column(db.String(20), nullable=False, default="curatif")
preventive_type = db.Column(db.String(50), nullable=True)
# Priorité
priority = db.Column(db.String(20), default="normale")
# Statut & Workflow
status = db.Column(db.String(30), default="brouillon")
# Responsable actuel (qui doit agir maintenant)
assigned_to_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
# Technicien assigné à l'exécution
technician_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
# GIMA
gima_number = db.Column(db.String(50), nullable=True)
gima_service = db.Column(db.String(20), nullable=True)
# Budget prévisionnel
needs_budget = db.Column(db.Boolean, default=False)
budget_approved = db.Column(db.Boolean, default=False)
budget_amount = db.Column(db.Float, nullable=True)
# Entreprise extérieure
company_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=True)
# Équipement
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"), nullable=True)
# Salle
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"), nullable=True)
# Quantité affectée (pour équipements multiples)
quantity_affected = db.Column(db.Integer, nullable=True)
individual_numbers = db.Column(db.Text, nullable=True)
# Dates
requested_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
started_at = db.Column(db.DateTime, nullable=True)
completed_at = db.Column(db.DateTime, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc))
# Récurrence
recurrence_month = db.Column(db.Integer, nullable=True)
frequency_days = db.Column(db.Integer, nullable=True)
next_due_date = db.Column(db.DateTime, nullable=True)
is_recurring = db.Column(db.Boolean, default=False)
# Coût & Temps
cost = db.Column(db.Float, nullable=True, default=0.0)
time_spent = db.Column(db.Integer, nullable=True)
# Demandeur externe
requester_name = db.Column(db.String(150), nullable=True)
requester_contact = db.Column(db.String(200), nullable=True)
# Suppression logique
is_deleted = db.Column(db.Boolean, default=False)
deleted_at = db.Column(db.DateTime, nullable=True)
deleted_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
# ENT
ent_message_id = db.Column(db.String(100), nullable=True)
ent_sender = db.Column(db.String(200), nullable=True)
# Lot
lot_id = db.Column(db.Integer, db.ForeignKey("lots.id"), nullable=True)
# Colonnes additionnelles
author_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
scheduled_date = db.Column(db.Date, nullable=True)
scheduled_start = db.Column(db.Time, nullable=True)
scheduled_end = db.Column(db.Time, nullable=True)
completed_date = db.Column(db.Date, nullable=True)
estimated_duration = db.Column(db.Float, nullable=True)
actual_duration = db.Column(db.Float, nullable=True)
notes = db.Column(db.Text, nullable=True)
# Reports
original_scheduled_date = db.Column(db.DateTime, nullable=True)
postpone_count = db.Column(db.Integer, default=0)
last_postpone_reason = db.Column(db.Text, nullable=True)
last_postpone_at = db.Column(db.DateTime, nullable=True)
# Champs migrés depuis ScheduledTask (fusion)
preventive_task_id = db.Column(db.Integer, db.ForeignKey("preventive_tasks.id"), nullable=True)
lot_task_id = db.Column(db.Integer, db.ForeignKey("lot_tasks.id"), nullable=True)
postpone_reason = db.Column(db.Text, nullable=True)
# Relations
parts_used = db.relationship("InterventionPart", backref="intervention",
cascade="all, delete-orphan", lazy="dynamic")
comments = db.relationship("InterventionComment", backref="intervention",
cascade="all, delete-orphan", lazy="dynamic")
documents = db.relationship("InterventionDocument", backref="intervention",
cascade="all, delete-orphan", lazy="dynamic")
status_history = db.relationship("StatusChange", backref="intervention",
cascade="all, delete-orphan", lazy="dynamic")
# User relations - utiliser foreign_keys pour éviter les conflits
assigned_to = db.relationship("User", foreign_keys=[assigned_to_id])
technician = db.relationship("User", foreign_keys=[technician_id])
author = db.relationship("User", foreign_keys=[author_id])
deleted_by = db.relationship("User", foreign_keys=[deleted_by_id])
# Lot relation
lot = db.relationship("Lot", backref="interventions_list")
room = db.relationship("Room", foreign_keys=[room_id])
preventive_task = db.relationship("PreventiveTask", foreign_keys=[preventive_task_id])
lot_task = db.relationship("LotTask", foreign_keys=[lot_task_id])
@property
def status_color(self):
"""Couleur du badge de statut."""
colors = {
'brouillon': 'secondary',
'planifiee': 'info',
'en_attente': 'warning',
'en_cours': 'primary',
'en_attente_pieces': 'warning',
'en_attente_entreprise': 'info',
'terminee': 'success',
'cloturee': 'dark',
'annulee': 'danger',
'ignoree': 'secondary',
}
return colors.get(self.status, 'secondary')
@property
def status_label(self):
"""Label du statut."""
labels = {
'brouillon': 'Brouillon',
'planifiee': 'Planifiee',
'en_attente': 'En attente',
'en_cours': 'En cours',
'en_attente_pieces': 'En attente pieces',
'en_attente_entreprise': 'En attente entreprise',
'terminee': 'Terminee',
'cloturee': 'Cloturee',
'annulee': 'Annulee',
'ignoree': 'Ignoree',
}
return labels.get(self.status, self.status)
@property
def priority_color(self):
"""Couleur du badge de priorité."""
colors = {
'basse': 'secondary',
'normale': 'primary',
'haute': 'warning',
'urgente': 'danger',
}
return colors.get(self.priority, 'secondary')
@property
def priority_label(self):
"""Label de la priorité."""
labels = {
'basse': 'Basse',
'normale': 'Normale',
'haute': 'Haute',
'urgente': 'Urgente',
}
return labels.get(self.priority, self.priority)
@property
def effective_estimated_duration(self):
"""Duree previsionnelle auto-adaptee.
Si actual_duration existe (duree reelle), on l'utilise.
Sinon si estimated_duration existe (duree manuelle), on l'utilise.
Sinon on calcule la moyenne des durees reelles des interventions
passees sur le meme equipement + meme type.
Fallback: 60 minutes par defaut.
"""
if self.actual_duration:
return self.actual_duration
if self.estimated_duration:
return self.estimated_duration
# Moyenne historique : interventions terminees sur le meme equipement + meme type
from sqlalchemy import func
avg = db.session.query(func.avg(Intervention.actual_duration)).filter(
Intervention.equipment_id == self.equipment_id,
Intervention.type == self.type,
Intervention.actual_duration.isnot(None),
Intervention.id != self.id
).scalar()
if avg and avg > 0:
return round(avg, 1)
# Moyenne par lot + type
if self.lot_id:
avg_lot = db.session.query(func.avg(Intervention.actual_duration)).filter(
Intervention.lot_id == self.lot_id,
Intervention.type == self.type,
Intervention.actual_duration.isnot(None),
Intervention.id != self.id
).scalar()
if avg_lot and avg_lot > 0:
return round(avg_lot, 1)
return 60.0 # valeur par defaut (1h)
@property
def is_overdue(self):
"""True si l'intervention est en retard (date planifiee depassee et non terminee)."""
if self.status in ('terminee', 'cloturee', 'annulee', 'ignoree'):
return False
if not self.scheduled_date:
return False
from datetime import date
return self.scheduled_date < date.today()
def can_be_ignored(self):
return self.status == 'planifiee'
def ignore(self, reason=''):
"""Marque l'intervention comme ignoree."""
self.status = 'ignoree'
self.postpone_reason = reason
from ...extensions import db
db.session.commit()
@property
def progress_label(self):
"""Label d'avancement selon le pourcentage."""
if self.status in ('terminee', 'cloturee'):
return 'Terminee'
if self.status == 'annulee':
return 'Annulee'
if self.status == 'brouillon':
return 'Brouillon'
if self.status == 'en_cours':
if not hasattr(self, '_progress') or self._progress is None:
return 'En cours'
return self.status_label
def __repr__(self):
return f"<Intervention #{self.id} {self.title}>"
class StatusChange(db.Model):
"""Historique des changements de statut."""
__tablename__ = "status_changes"
id = db.Column(db.Integer, primary_key=True)
intervention_id = db.Column(db.Integer, db.ForeignKey("interventions.id"), nullable=False)
from_status = db.Column(db.String(30), nullable=True)
to_status = db.Column(db.String(30), nullable=False)
changed_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
comment = db.Column(db.Text, default="")
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
# Relations
changed_by = db.relationship("User", backref="status_changes_list")
def __repr__(self):
return f"<StatusChange {self.from_status}{self.to_status}>"
class InterventionComment(db.Model):
"""Commentaire sur une intervention."""
__tablename__ = "intervention_comments"
id = db.Column(db.Integer, primary_key=True)
intervention_id = db.Column(db.Integer, db.ForeignKey("interventions.id"), nullable=False)
author_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
content = db.Column(db.Text, nullable=False)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
# Relations
author = db.relationship("User", backref="comments_list")
def __repr__(self):
return f"<InterventionComment {self.id}>"
class InterventionDocument(db.Model):
"""Document attaché à une intervention."""
__tablename__ = "intervention_documents"
id = db.Column(db.Integer, primary_key=True)
intervention_id = db.Column(db.Integer, db.ForeignKey("interventions.id"), nullable=False)
filename = db.Column(db.String(255), nullable=False)
filepath = db.Column(db.String(500), nullable=False)
description = db.Column(db.Text, nullable=True)
uploaded_at = db.Column(db.DateTime, nullable=True)
uploaded_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=True)
# Relations
uploader = db.relationship("User", foreign_keys=[uploaded_by_id])
def __repr__(self):
return f"<InterventionDocument {self.filename}>"
class InterventionPart(db.Model):
"""Pièce utilisée dans une intervention."""
__tablename__ = "intervention_parts"
id = db.Column(db.Integer, primary_key=True)
intervention_id = db.Column(db.Integer, db.ForeignKey("interventions.id"), nullable=False)
part_id = db.Column(db.Integer, db.ForeignKey("parts.id"), nullable=False)
quantity = db.Column(db.Integer, nullable=False, default=1)
# Relations - utiliser back_populates pour éviter conflit
part = db.relationship("Part", back_populates="usages_list")
def __repr__(self):
return f"<InterventionPart {self.part_id}x{self.quantity}>"
class Lot(db.Model):
"""Lot de maintenance."""
__tablename__ = "lots"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(500), nullable=False)
company_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=True)
category_id = db.Column(db.Integer, db.ForeignKey("equipment_categories.id"), nullable=True)
is_present = db.Column(db.Boolean, nullable=False, default=True)
absence_reason = db.Column(db.String(500), nullable=True)
# Relations
category = db.relationship("EquipmentCategory", foreign_keys=[category_id])
tasks = db.relationship("LotTask", backref="lot")
@property
def active_equipment_count(self):
"""Equipements relies directement au lot et encore dans le patrimoine."""
from .equipment import Equipment
return Equipment.query.filter(
Equipment.lot_id == self.id,
Equipment.is_deleted.is_(False),
db.or_(Equipment.status.is_(None), Equipment.status != 'jete'),
).count()
@property
def needs_equipment_attention(self):
"""Un lot present sans equipement doit etre signale."""
return self.is_present and self.active_equipment_count == 0
def __repr__(self):
return f"<Lot {self.name}>"
class LotTask(db.Model):
"""Tâche d'un lot de maintenance."""
__tablename__ = "lot_tasks"
id = db.Column(db.Integer, primary_key=True)
lot_id = db.Column(db.Integer, db.ForeignKey("lots.id"), nullable=False)
num_tache = db.Column(db.String(10), nullable=True)
tache = db.Column(db.Text, nullable=True)
periodicite = db.Column(db.String(200), nullable=True)
contrat = db.Column(db.Boolean, default=False)
company_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=True)
contract_id = db.Column(db.Integer, db.ForeignKey("contracts.id"), nullable=True)
gestionnaire = db.Column(db.String(50), nullable=True)
jours_entre_interventions = db.Column(db.Integer, nullable=True)
type_maintenance = db.Column(db.String(50), nullable=True)
duree_minutes = db.Column(db.Integer, nullable=True) # Durée estimée en minutes
trigger_type = db.Column(db.String(20), nullable=False, default="calendar")
trigger_event = db.Column(db.String(80), nullable=True)
meter_threshold = db.Column(db.Float, nullable=True)
season_start_month = db.Column(db.Integer, nullable=True)
season_end_month = db.Column(db.Integer, nullable=True)
advance_days = db.Column(db.Integer, nullable=False, default=0)
is_active = db.Column(db.Boolean, nullable=False, default=True)
application_scope = db.Column(db.String(20), nullable=False, default="auto")
unavailable_policy = db.Column(db.String(20), nullable=False, default="suspend")
after_repair_policy = db.Column(db.String(20), nullable=False, default="preserve")
is_safety_critical = db.Column(db.Boolean, nullable=False, default=False)
company = db.relationship("Company")
contract = db.relationship("Contract")
def effective_duration(self):
"""Duree previsionnelle auto-adaptee pour cette tache de lot.
Si duree_minutes est defini manuellement, on l'utilise.
Sinon on calcule la moyenne des durees reelles des interventions
passees pour ce lot + ce type de tache.
Fallback: 60 minutes par defaut.
"""
from sqlalchemy import func
if self.duree_minutes:
return self.duree_minutes
# Moyenne des interventions de ce lot
avg = db.session.query(func.avg(Intervention.actual_duration)).filter(
Intervention.lot_id == self.lot_id,
Intervention.actual_duration.isnot(None)
).scalar()
if avg and avg > 0:
return int(round(avg))
return 60
def __repr__(self):
return f"<LotTask {self.num_tache or self.id}>"
class WorkRequest(db.Model):
"""Demande de travaux suivie en interne par le technicien."""
__tablename__ = "work_requests"
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(255), nullable=False)
description = db.Column(db.Text, nullable=False)
status = db.Column(db.String(40), nullable=False, default="brouillon", index=True)
execution_mode = db.Column(db.String(30), nullable=False, default="interne")
priority = db.Column(db.String(20), nullable=False, default="normale")
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"), nullable=True)
service_id = db.Column(db.Integer, db.ForeignKey("services.id"), nullable=True)
company_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=True)
created_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
chief_decision = db.Column(db.String(20), nullable=True)
department_decision = db.Column(db.String(20), nullable=True)
chief_decision_date = db.Column(db.Date, nullable=True)
department_decision_date = db.Column(db.Date, nullable=True)
planned_date = db.Column(db.Date, nullable=True)
completed_date = db.Column(db.Date, nullable=True)
actual_minutes = db.Column(db.Integer, nullable=True)
parts_cost = db.Column(db.Numeric(10, 2), nullable=True)
notes = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc))
equipment = db.relationship("Equipment")
service = db.relationship("Service")
company = db.relationship("Company")
created_by = db.relationship("User")
quotes = db.relationship("WorkRequestQuote", back_populates="request", cascade="all, delete-orphan")
events = db.relationship("WorkRequestEvent", back_populates="request", cascade="all, delete-orphan", order_by="WorkRequestEvent.created_at")
class WorkRequestQuote(db.Model):
__tablename__ = "work_request_quotes"
id = db.Column(db.Integer, primary_key=True)
request_id = db.Column(db.Integer, db.ForeignKey("work_requests.id", ondelete="CASCADE"), nullable=False)
company_id = db.Column(db.Integer, db.ForeignKey("companies.id"), nullable=True)
reference = db.Column(db.String(100), nullable=True)
amount_ht = db.Column(db.Numeric(10, 2), nullable=True)
amount_ttc = db.Column(db.Numeric(10, 2), nullable=True)
quote_date = db.Column(db.Date, nullable=True)
valid_until = db.Column(db.Date, nullable=True)
status = db.Column(db.String(20), nullable=False, default="en_attente")
file_path = db.Column(db.String(500), nullable=True)
notes = db.Column(db.Text, nullable=True)
request = db.relationship("WorkRequest", back_populates="quotes")
company = db.relationship("Company")
class WorkRequestEvent(db.Model):
__tablename__ = "work_request_events"
id = db.Column(db.Integer, primary_key=True)
request_id = db.Column(db.Integer, db.ForeignKey("work_requests.id", ondelete="CASCADE"), nullable=False)
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
old_status = db.Column(db.String(40), nullable=True)
new_status = db.Column(db.String(40), nullable=False)
comment = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
request = db.relationship("WorkRequest", back_populates="events")
user = db.relationship("User")
class LotService(db.Model):
"""Association entre lots et services."""
__tablename__ = "lot_services"
__table_args__ = {'extend_existing': True}
id = db.Column(db.Integer, primary_key=True)
lot_id = db.Column(db.Integer, db.ForeignKey("lots.id"), nullable=False)
service_id = db.Column(db.Integer, db.ForeignKey("services.id"), nullable=False)
assigned_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
def __repr__(self):
return f"<LotService {self.lot_id} - {self.service_id}>"
class WatchdogLog(db.Model):
"""Log centralise des evenements des watchdogs."""
__tablename__ = "watchdog_logs"
id = db.Column(db.Integer, primary_key=True)
watchdog_name = db.Column(db.String(50), nullable=False, index=True)
level = db.Column(db.String(20), nullable=False, default='info') # debug, info, warning, error
message = db.Column(db.Text, nullable=False)
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), index=True)
def __repr__(self):
return f"<WatchdogLog {self.watchdog_name} {self.level}>"