175 lines
7.3 KiB
Python
175 lines
7.3 KiB
Python
"""
|
|
Lots Routes - GMAO Collège
|
|
Gestion des lots de maintenance
|
|
"""
|
|
from flask import Blueprint, render_template, request, flash, redirect, url_for
|
|
from flask_login import login_required
|
|
from app_new.extensions import db
|
|
from ..core.models.maintenance import Lot, LotTask
|
|
from ..core.models.planning import ScheduledTask
|
|
from ..core.services.maintenance_engine import generate_due_tasks
|
|
from ..core.models.company import Company
|
|
from ..core.models.equipment import Equipment
|
|
|
|
lots_bp = Blueprint('lots', __name__)
|
|
|
|
@lots_bp.route('/')
|
|
@login_required
|
|
def index():
|
|
"""Liste des lots."""
|
|
page = request.args.get('page', 1, type=int)
|
|
search = (request.args.get('search') or '').strip()
|
|
company_id = request.args.get('company_id', type=int)
|
|
presence = request.args.get('presence', '')
|
|
|
|
query = Lot.query
|
|
if search:
|
|
query = query.filter(Lot.name.ilike(f'%{search}%'))
|
|
if company_id:
|
|
query = query.filter(Lot.company_id == company_id)
|
|
if presence == 'present':
|
|
query = query.filter(Lot.is_present.is_(True))
|
|
elif presence == 'absent':
|
|
query = query.filter(Lot.is_present.is_(False))
|
|
|
|
lots = query.order_by(Lot.name).paginate(page=page, per_page=50, error_out=False)
|
|
companies = Company.query.order_by(Company.name).all()
|
|
return render_template('lots/index.html', lots=lots, companies=companies)
|
|
|
|
@lots_bp.route('/<int:id>')
|
|
@login_required
|
|
def detail(id):
|
|
"""Détail d'un lot."""
|
|
lot = Lot.query.get_or_404(id)
|
|
# Récupérer les équipements de ce lot
|
|
equipments = Equipment.query.filter_by(lot_id=id).order_by(Equipment.name).all()
|
|
task_ids = [task.id for task in lot.tasks]
|
|
task_counts = {'planned': 0, 'suspended': 0, 'overdue': 0}
|
|
if task_ids:
|
|
occurrences = ScheduledTask.query.filter(ScheduledTask.lot_task_id.in_(task_ids)).all()
|
|
task_counts['planned'] = sum(item.status == 'planned' for item in occurrences)
|
|
task_counts['suspended'] = sum(item.status == 'suspended' for item in occurrences)
|
|
task_counts['overdue'] = sum(item.is_overdue for item in occurrences)
|
|
return render_template('lots/detail.html', lot=lot, equipments=equipments, task_counts=task_counts)
|
|
|
|
|
|
def _apply_task_form(task):
|
|
task.num_tache = (request.form.get('num_tache') or '').strip() or None
|
|
task.tache = (request.form.get('tache') or '').strip() or None
|
|
task.periodicite = (request.form.get('periodicite') or '').strip() or None
|
|
task.jours_entre_interventions = request.form.get('jours_entre_interventions', type=int)
|
|
task.duree_minutes = request.form.get('duree_minutes', type=int)
|
|
task.type_maintenance = (request.form.get('type_maintenance') or 'preventive').strip()
|
|
task.trigger_type = request.form.get('trigger_type') or 'calendar'
|
|
task.trigger_event = (request.form.get('trigger_event') or '').strip() or None
|
|
task.meter_threshold = request.form.get('meter_threshold', type=float)
|
|
task.season_start_month = request.form.get('season_start_month', type=int)
|
|
task.season_end_month = request.form.get('season_end_month', type=int)
|
|
task.advance_days = max(request.form.get('advance_days', type=int) or 0, 0)
|
|
task.application_scope = request.form.get('application_scope') or 'auto'
|
|
task.unavailable_policy = request.form.get('unavailable_policy') or 'suspend'
|
|
task.after_repair_policy = request.form.get('after_repair_policy') or 'preserve'
|
|
task.is_safety_critical = request.form.get('is_safety_critical') == 'on'
|
|
task.is_active = request.form.get('is_active') == 'on'
|
|
|
|
|
|
@lots_bp.route('/<int:lot_id>/tasks/new', methods=['GET', 'POST'])
|
|
@login_required
|
|
def task_create(lot_id):
|
|
lot = Lot.query.get_or_404(lot_id)
|
|
task = LotTask(lot_id=lot.id, is_active=True)
|
|
if request.method == 'POST':
|
|
_apply_task_form(task)
|
|
if not task.tache:
|
|
flash('Le nom de la tâche est obligatoire.', 'danger')
|
|
else:
|
|
db.session.add(task)
|
|
db.session.commit()
|
|
flash('Tâche préventive ajoutée au lot.', 'success')
|
|
return redirect(url_for('lots.detail', id=lot.id))
|
|
return render_template('lots/task_form.html', lot=lot, task=task, creating=True)
|
|
|
|
|
|
@lots_bp.route('/<int:lot_id>/tasks/<int:task_id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def task_edit(lot_id, task_id):
|
|
lot = Lot.query.get_or_404(lot_id)
|
|
task = LotTask.query.filter_by(id=task_id, lot_id=lot.id).first_or_404()
|
|
if request.method == 'POST':
|
|
_apply_task_form(task)
|
|
if not task.tache:
|
|
flash('Le nom de la tâche est obligatoire.', 'danger')
|
|
else:
|
|
db.session.commit()
|
|
flash('Règle préventive mise à jour.', 'success')
|
|
return redirect(url_for('lots.detail', id=lot.id))
|
|
return render_template('lots/task_form.html', lot=lot, task=task, creating=False)
|
|
|
|
|
|
@lots_bp.route('/<int:lot_id>/generate', methods=['POST'])
|
|
@login_required
|
|
def generate_tasks(lot_id):
|
|
lot = Lot.query.get_or_404(lot_id)
|
|
if not lot.is_present:
|
|
flash('Aucune échéance générée : ce lot est marqué non présent.', 'warning')
|
|
return redirect(url_for('lots.detail', id=lot.id))
|
|
created = []
|
|
for task in lot.tasks:
|
|
created.extend(generate_due_tasks(task, commit=False))
|
|
db.session.commit()
|
|
flash(f'{len(created)} nouvelle(s) échéance(s) générée(s), sans doublon.', 'success')
|
|
return redirect(url_for('lots.detail', id=lot.id))
|
|
|
|
@lots_bp.route('/<int:id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def edit(id):
|
|
"""Modifier un lot (formulaire)."""
|
|
lot = Lot.query.get_or_404(id)
|
|
if request.method == 'POST':
|
|
lot.company_id = request.form.get('company_id', type=int)
|
|
lot.absence_reason = (request.form.get('absence_reason') or '').strip() or None
|
|
db.session.commit()
|
|
flash('Lot modifié.', 'success')
|
|
return redirect(url_for('lots.detail', id=id))
|
|
|
|
companies = Company.query.order_by(Company.name).all()
|
|
return render_template('lots/edit.html', lot=lot, companies=companies)
|
|
|
|
|
|
@lots_bp.route('/<int:id>/presence', methods=['POST'])
|
|
@login_required
|
|
def set_presence(id):
|
|
"""Marquer un lot comme présent ou non présent dans l'établissement."""
|
|
lot = Lot.query.get_or_404(id)
|
|
action = request.form.get('action')
|
|
|
|
if action == 'absent':
|
|
if lot.active_equipment_count:
|
|
flash(
|
|
f"Impossible de marquer ce lot non présent : "
|
|
f"{lot.active_equipment_count} équipement(s) y sont encore rattachés.",
|
|
'danger',
|
|
)
|
|
else:
|
|
lot.is_present = False
|
|
lot.absence_reason = (request.form.get('absence_reason') or '').strip() or None
|
|
db.session.commit()
|
|
flash(f"Le lot « {lot.name} » est marqué non présent.", 'success')
|
|
elif action == 'present':
|
|
lot.is_present = True
|
|
lot.absence_reason = None
|
|
db.session.commit()
|
|
flash(f"Le lot « {lot.name} » est marqué présent.", 'success')
|
|
else:
|
|
flash('Action de présence invalide.', 'danger')
|
|
|
|
return redirect(request.referrer or url_for('lots.detail', id=id))
|
|
|
|
@lots_bp.route('/<int:id>/delete', methods=['POST'])
|
|
@login_required
|
|
def delete(id):
|
|
"""Supprimer un lot."""
|
|
lot = Lot.query.get_or_404(id)
|
|
# TODO: suppression avec déplacement vers corbeille
|
|
flash("Fonctionnalité en cours de développement", "info")
|
|
return redirect(url_for('lots.index'))
|