"""Equipments Main Routes - GMAO Collège""" from flask import Blueprint, render_template, redirect, url_for, request, flash, jsonify from flask_login import login_required, current_user from datetime import datetime from app_new.extensions import db from ..core.models.equipment import Equipment, EquipmentCategory, EquipmentDocument, EquipmentRoomHistory, EquipmentRestriction from ..core.models.college import Room, Building, RoomType, Zone from ..core.models.maintenance import Lot, Intervention from ..core.models.planning import Meter, MeterReading, Consumable, EquipmentConsumable, ScheduledTask from app_new.constants import EQUIPMENT_STATUSES def _empty_to_none(v): """Convertit une chaîne vide en None.""" return None if v == '' else v main_bp = Blueprint("equipments", __name__, template_folder='templates') """ Equipments Routes - GMAO Collège Gestion des équipements """ from flask import Blueprint, render_template, redirect, url_for, request, flash, jsonify from flask_login import login_required, current_user from datetime import datetime from app_new.extensions import db from ..core.models.equipment import Equipment, EquipmentCategory, EquipmentDocument, EquipmentRoomHistory, EquipmentRestriction from ..core.models.college import Room, Building, RoomType, Zone from ..core.models.maintenance import Lot, Intervention from ..core.models.planning import Meter, MeterReading, Consumable, EquipmentConsumable from app_new.constants import EQUIPMENT_STATUSES def _empty_to_none(v): """Convertit une chaîne vide en None.""" return None if v == '' else v equipments_bp = Blueprint('equipments', __name__, template_folder='templates') @main_bp.route('/') @login_required def index(): """Liste des équipements - hiérarchie parent > zone > salle > individuel.""" search = request.args.get('search', '') status = request.args.get('status', '') category_id = request.args.get('category', type=int) building_id = request.args.get('building', type=int) # Pagination (page courante, 20 parents par page) page = request.args.get('page', 1, type=int) per_page = 20 # Parents (équipements racine sans parent) parent_query = Equipment.query.filter_by(parent_id=None) if search: parent_query = parent_query.filter( db.or_( Equipment.name.ilike(f'%{search}%'), Equipment.code.ilike(f'%{search}%') ) ) if status: parent_query = parent_query.filter_by(status=status) if category_id: parent_query = parent_query.filter_by(category_id=category_id) if building_id: parent_query = parent_query.join(Room, Equipment.room_id == Room.id).filter(Room.building_id == building_id) parents_pagination = parent_query.order_by(Equipment.name).paginate( page=page, per_page=per_page, error_out=False ) parents = parents_pagination.items parent_ids = [p.id for p in parents] if parent_ids: from sqlalchemy import select # CTE récursive qui remonte le room_id effectif (hérité du parent si l'enfant n'en a pas) descendants = ( select( Equipment.id, Equipment.parent_id, Equipment.name, Equipment.code, Equipment.individual_number, Equipment.status, Equipment.room_id, Equipment.is_group, Equipment.parent_id.label('root_id'), Equipment.room_id.label('effective_room_id') ) .where(Equipment.parent_id.in_(parent_ids)) .cte(recursive=True) ) descendants = descendants.union_all( select( Equipment.id, Equipment.parent_id, Equipment.name, Equipment.code, Equipment.individual_number, Equipment.status, Equipment.room_id, Equipment.is_group, descendants.c.root_id, db.func.coalesce(Equipment.room_id, descendants.c.effective_room_id).label('effective_room_id') ) .where(Equipment.parent_id == descendants.c.id) ) # Requête finale: TOUS les descendants (groupes enfants inclus) avec room/zone effective stmt = select( descendants.c.id.label('id'), descendants.c.name.label('name'), descendants.c.code.label('code'), descendants.c.individual_number.label('individual_number'), descendants.c.status.label('status'), descendants.c.is_group.label('is_group'), descendants.c.root_id.label('root_id'), Room.id.label('room_id'), Room.name.label('room_name'), Zone.id.label('zone_id'), Zone.name.label('zone_name') ).select_from(descendants)\ .join(Room, descendants.c.effective_room_id == Room.id, isouter=True)\ .join(Zone, Room.zone_id == Zone.id, isouter=True) # Pas de filtre is_group : on affiche tous les descendants comme éléments rows = db.session.execute(stmt).all() # Récupérer les IDs des individuels pour les curatifs indiv_ids = [row.id for row in rows] curative_counts = {} if indiv_ids: from ..core.models.maintenance import Intervention counts = db.session.query( Intervention.equipment_id, db.func.count(Intervention.id).label('cnt') ).filter( Intervention.equipment_id.in_(indiv_ids), Intervention.type == 'curatif' ).group_by(Intervention.equipment_id).all() curative_counts = {c.equipment_id: c.cnt for c in counts} # Regrouper par parent > zone > salle data = {} for row in rows: root_id = row.root_id if root_id not in data: data[root_id] = {} zone_id = row.zone_id if row.zone_id else 0 zone_name = row.zone_name if row.zone_name else "Sans zone" if zone_id not in data[root_id]: data[root_id][zone_id] = {'name': zone_name, 'rooms': {}} room_id = row.room_id if row.room_id else 0 room_name = row.room_name if row.room_name else "Sans salle" if room_id not in data[root_id][zone_id]['rooms']: data[root_id][zone_id]['rooms'][room_id] = {'name': room_name, 'individuals': []} indiv = { 'id': row.id, 'name': row.name, 'code': row.code, 'individual_number': row.individual_number, 'status': row.status, 'curative_count': curative_counts.get(row.id, 0) } data[root_id][zone_id]['rooms'][room_id]['individuals'].append(indiv) else: data = {} # Attacher les données aux objets parent for parent in parents: parent.zones_list = [] zones_map = data.get(parent.id, {}) for zone_id in sorted(zones_map.keys(), key=lambda z: zones_map[z]['name']): zone_data = zones_map[zone_id] rooms_list = [] for room_id in sorted(zone_data['rooms'].keys(), key=lambda r: zone_data['rooms'][r]['name']): room_data = zone_data['rooms'][room_id] rooms_list.append({ 'room': {'id': room_id, 'name': room_data['name']}, 'name': room_data['name'], 'individuals': room_data['individuals'] }) parent.zones_list.append({ 'zone': {'id': zone_id, 'name': zone_data['name']}, 'name': zone_data['name'], 'rooms_list': rooms_list }) parent.individual_count = sum( len(room['individuals']) for zone in parent.zones_list for room in zone['rooms_list'] ) categories = EquipmentCategory.query.order_by(EquipmentCategory.name).all() buildings = Building.query.order_by(Building.name).all() return render_template('equipments/index.html', parents=parents, pagination=parents_pagination, categories=categories, buildings=buildings, statuses=EQUIPMENT_STATUSES, filters={'search': search, 'status': status, 'category': category_id, 'building': building_id}) @main_bp.route('/create', methods=['GET', 'POST']) @login_required def create(): """Créer un équipement.""" # Récupérer le parent_id depuis query string ou POST preselected_parent_id = request.args.get('parent_id') if request.method == 'POST': parent_id = request.form.get('parent_select') or request.form.get('parent_id') or None equipment = Equipment( name=request.form.get('name'), category_id=request.form.get('category_id') or None, room_id=request.form.get('room_id') or None, lot_id=request.form.get('lot_id') or None, code=_empty_to_none(request.form.get('code', '').strip()), description=_empty_to_none(request.form.get('description', '').strip()), status=request.form.get('status', 'en_service'), is_group=request.form.get('is_group') == '1', quantity=int(request.form.get('quantity', 1) or 1), parent_id=parent_id ) db.session.add(equipment) db.session.commit() # Créer les tâches planifiées si l'équipement a un lot if equipment.lot_id: equipment.create_scheduled_tasks_from_lot() flash('Équipement créé avec succès.', 'success') return redirect(url_for('equipments.detail', id=equipment.id)) categories = EquipmentCategory.query.order_by(EquipmentCategory.name).all() rooms = Room.query.order_by(Room.name).all() buildings = Building.query.order_by(Building.name).all() parent_equipments = Equipment.query.filter(Equipment.is_group == True, Equipment.parent_id == None).order_by(Equipment.name).all() lots = Lot.query.order_by(Lot.name).all() return render_template('equipments/form.html', categories=categories, rooms=rooms, buildings=buildings, parent_equipments=parent_equipments, lots=lots, statuses=EQUIPMENT_STATUSES, preselected_parent_id=preselected_parent_id) @main_bp.route('/') @login_required def detail(id): """Détail d'un équipement.""" equipment = Equipment.query.get_or_404(id) return render_template('equipments/detail.html', equipment=equipment) @main_bp.route('//edit', methods=['GET', 'POST']) @login_required def edit(id): """Modifier un équipement.""" equipment = Equipment.query.get_or_404(id) if request.method == 'POST': equipment.name = request.form.get('name') equipment.code = _empty_to_none(request.form.get('code')) equipment.description = _empty_to_none(request.form.get('description')) equipment.category_id = _empty_to_none(request.form.get('category_id')) equipment.lot_id = _empty_to_none(request.form.get('lot_id')) equipment.room_id = _empty_to_none(request.form.get('room_id')) equipment.status = request.form.get('status', 'en_service') equipment.quantity = int(request.form.get('quantity', 1)) or 1 equipment.is_group = bool(request.form.get('is_group')) equipment.tracked_individually = bool(request.form.get('tracked_individually')) equipment.position = _empty_to_none(request.form.get('position')) # Position X/Y pos_x = request.form.get('position_x') pos_y = request.form.get('position_y') equipment.position_x = float(pos_x) if pos_x else None equipment.position_y = float(pos_y) if pos_y else None # Dates install_date = request.form.get('install_date') warranty_end = request.form.get('warranty_end') equipment.install_date = datetime.strptime(install_date, '%Y-%m-%d').date() if install_date else None equipment.warranty_end = datetime.strptime(warranty_end, '%Y-%m-%d').date() if warranty_end else None # Parent parent_id = request.form.get('parent_id') equipment.parent_id = int(parent_id) if parent_id else None # Si le lot a changé, créer les tâches planifiées old_lot_id = db.session.query(Equipment.lot_id).filter(Equipment.id == equipment.id).scalar() db.session.commit() if old_lot_id != equipment.lot_id and equipment.lot_id: equipment.create_scheduled_tasks_from_lot() flash('Équipement modifié avec succès.', 'success') return redirect(url_for('equipments.detail', id=equipment.id)) categories = EquipmentCategory.query.order_by(EquipmentCategory.name).all() rooms = Room.query.order_by(Room.name).all() buildings = Building.query.order_by(Building.name).all() parent_equipments = Equipment.query.filter(Equipment.id != id).order_by(Equipment.name).all() lots = Lot.query.order_by(Lot.name).all() # Charger les relations meters = Meter.query.filter_by(equipment_id=id).all() equipment_consumables = EquipmentConsumable.query.filter_by(equipment_id=id).all() all_consumables = Consumable.query.filter_by(is_active=True).order_by(Consumable.name).all() room_history = EquipmentRoomHistory.query.filter_by(equipment_id=id).order_by(EquipmentRoomHistory.changed_at.desc()).all() return render_template('equipments/edit.html', equipment=equipment, categories=categories, rooms=rooms, buildings=buildings, parent_equipments=parent_equipments, lots=lots, meters=meters, equipment_consumables=equipment_consumables, all_consumables=all_consumables, room_history=room_history) @main_bp.route('//delete', methods=['POST']) @login_required def delete(id): """Supprimer un équipement (et ses enfants en cascade).""" equipment = Equipment.query.get_or_404(id) name = equipment.name # Collecter tous les IDs (parent + enfants récursifs) def collect_ids(eq): ids = [eq.id] for child in eq.children: ids.extend(collect_ids(child)) return ids all_ids = collect_ids(equipment) # Supprimer les dépendances en cascade (l'ordre importe peu avec les bonnes FK) EquipmentRoomHistory.query.filter(EquipmentRoomHistory.equipment_id.in_(all_ids)).delete(synchronize_session=False) EquipmentDocument.query.filter(EquipmentDocument.equipment_id.in_(all_ids)).delete(synchronize_session=False) EquipmentConsumable.query.filter(EquipmentConsumable.equipment_id.in_(all_ids)).delete(synchronize_session=False) EquipmentRestriction.query.filter(EquipmentRestriction.equipment_id.in_(all_ids)).delete(synchronize_session=False) Intervention.query.filter(Intervention.equipment_id.in_(all_ids)).delete(synchronize_session=False) # Supprimer les équipements (du plus profond au parent) def delete_recursive(eq): for child in list(eq.children): delete_recursive(child) db.session.delete(eq) delete_recursive(equipment) db.session.commit() flash(f"Équipement '{name}' et ses sous-éléments supprimés.", 'success') return redirect(url_for('equipments.index')) def _cascade_soft_delete(equipment): """Suppression logique en cascade (si is_active existe).""" for child in list(equipment.children): _cascade_soft_delete(child) if hasattr(equipment, 'is_active'): equipment.is_active = False @main_bp.route('//soft_delete', methods=['POST']) @login_required def soft_delete(id): """Suppression logique d'un équipement et de ses enfants.""" equipment = Equipment.query.get_or_404(id) name = equipment.name # Si is_active n'existe pas, supprimer physiquement if not hasattr(equipment, 'is_active'): def delete_recursive(eq): for child in list(eq.children): delete_recursive(child) db.session.delete(eq) delete_recursive(equipment) else: _cascade_soft_delete(equipment) db.session.commit() flash(f"Équipement '{name}' et ses sous-éléments supprimés.", 'success') return redirect(url_for('equipments.index')) @main_bp.route('/categories') @login_required def categories(): """Liste des catégories d'équipements.""" categories = EquipmentCategory.query.order_by(EquipmentCategory.name).all() return render_template('equipments/categories.html', categories=categories) @main_bp.route('/categories/') @login_required def category_detail(id): """Détail d'une catégorie d'équipements.""" category = EquipmentCategory.query.get_or_404(id) equipments = Equipment.query.filter_by(category_id=id).order_by(Equipment.name).all() return render_template('equipments/category_detail.html', category=category, equipments=equipments) @main_bp.route('/categories/new', methods=['GET', 'POST']) @login_required def create_category(): """Créer une catégorie.""" if request.method == 'POST': cat = EquipmentCategory( name=request.form.get('name'), description=request.form.get('description') ) db.session.add(cat) db.session.commit() flash(f"Catégorie '{cat.name}' créée.", 'success') return redirect(url_for('equipments.categories')) return render_template('equipments/category_form.html', title='Nouvelle catégorie') @main_bp.route('//status/', methods=['POST']) @login_required def change_status(id, status): """Changer le statut d'un équipement.""" valid_statuses = ['en_service', 'hs', 'a_jeter', 'jete', 'actif'] if status not in valid_statuses: flash(f'Statut invalide: {status}', 'danger') return redirect(url_for('equipments.detail', id=id)) equipment = Equipment.query.get_or_404(id) equipment.status = status db.session.commit() flash('Statut modifié.', 'success') return redirect(url_for('equipments.detail', id=id)) @main_bp.route('//mark-trash', methods=['POST']) @login_required def mark_as_trash(id): """Marquer un équipement comme à jeter.""" equipment = Equipment.query.get_or_404(id) equipment.status = 'a_jeter' db.session.commit() flash(f"Équipement '{equipment.name}' marqué comme à jeter.", 'warning') return redirect(url_for('equipments.detail', id=equipment.id)) @main_bp.route('//cancel-trash', methods=['POST']) @login_required def cancel_trash(id): """Annuler le statut 'à jeter'.""" equipment = Equipment.query.get_or_404(id) if equipment.status != 'a_jeter': flash("Cet équipement n'est pas marqué comme à jeter.", 'warning') return redirect(url_for('equipments.detail', id=equipment.id)) equipment.status = 'en_service' db.session.commit() flash(f"Équipement '{equipment.name}' remis en service.", 'success') return redirect(url_for('equipments.detail', id=equipment.id)) @main_bp.route('//confirm-trash', methods=['POST']) @login_required def confirm_trashed(id): """Confirmer qu'un équipement a été jeté.""" equipment = Equipment.query.get_or_404(id) equipment.status = 'jete' equipment.room_id = None db.session.commit() flash(f"Équipement '{equipment.name}' marqué comme jeté.", 'success') return redirect(url_for('equipments.trashed')) @main_bp.route('/a-jeter') @login_required def to_trash(): """Liste des équipements à jeter.""" equipments = Equipment.query.filter_by(status='a_jeter').order_by(Equipment.name).all() return render_template('equipments/trash_list.html', equipments=equipments, title='Équipements à jeter') @main_bp.route('/jetes') @login_required def trashed(): """Liste des équipements jetés.""" equipments = Equipment.query.filter_by(status='jete').order_by(Equipment.updated_at.desc()).all() return render_template('equipments/trash_list.html', equipments=equipments, title='Anciens équipements') @main_bp.route('/bulk-confirm-trash', methods=['POST']) @login_required def bulk_confirm_trash(): """Confirmer plusieurs équipements comme jetés.""" ids = request.form.getlist('equipment_ids') for eq_id in ids: equipment = Equipment.query.get(eq_id) if equipment: equipment.status = 'jete' equipment.room_id = None db.session.commit() flash(f'{len(ids)} équipements marqués comme jetés.', 'success') return redirect(url_for('equipments.trashed')) @main_bp.route('//add-elements', methods=['GET', 'POST']) @login_required def add_elements(id): """Ajouter des éléments à un groupe existant.""" equipment = Equipment.query.get_or_404(id) if not equipment.is_group: flash("Cet équipement n'est pas un groupe.", 'warning') return redirect(url_for('equipments.detail', id=equipment.id)) if request.method == 'POST': count = int(request.form.get('count', 1)) if count < 1: flash('La quantite ajoutee doit etre positive.', 'danger') return redirect(url_for('equipments.add_elements', id=equipment.id)) first_number = equipment.individualized_quantity + 1 for i in range(count): new_element = Equipment( name=f"{equipment.name} #{first_number + i}", code=f"{equipment.code}-{first_number + i}" if equipment.code else None, parent_id=equipment.id, individual_number=first_number + i, status='en_service', room_id=equipment.room_id, quantity=1, is_group=False, tracked_individually=True, ) db.session.add(new_element) # L'ajout augmente le patrimoine physique du groupe. equipment.quantity = (equipment.quantity or 0) + count db.session.commit() # Créer les tâches planifiées pour les nouveaux éléments # Récupérer les enfants récemment créés children = Equipment.query.filter_by(parent_id=equipment.id).order_by(Equipment.id.desc()).limit(count).all() for child in children: if child.effective_lot_id: child.create_scheduled_tasks_from_lot() flash(f'{count} élément(s) ajouté(s) au groupe.', 'success') return redirect(url_for('equipments.detail', id=equipment.id)) return render_template('equipments/add_elements.html', equipment=equipment) @main_bp.route('//individualize', methods=['GET', 'POST']) @login_required def individualize(id): """Extraire une unite quantitative d'un groupe vers une fiche individuelle.""" equipment = Equipment.query.get_or_404(id) if not equipment.is_group: flash("Seul un groupe d'équipements peut fournir une unité à individualiser.", 'warning') return redirect(url_for('equipments.detail', id=equipment.id)) if equipment.unindividualized_quantity < 1: flash("Toutes les unités de ce groupe sont déjà individualisées.", 'warning') return redirect(url_for('equipments.detail', id=equipment.id)) if request.method == 'POST': next_number = max( [child.individual_number or 0 for child in equipment.children] or [0] ) + 1 name = (request.form.get('name') or '').strip() or f"{equipment.name} #{next_number}" code = (request.form.get('code') or '').strip() or None position = (request.form.get('position') or '').strip() or None individual = Equipment( name=name, code=code, description=equipment.description, status='en_service', parent_id=equipment.id, is_group=False, quantity=1, tracked_individually=True, individual_number=next_number, position=position, room_id=equipment.room_id, # Le lot et la categorie restent herites du groupe. lot_id=None, category_id=None, ) db.session.add(individual) db.session.commit() if individual.effective_lot_id: individual.create_scheduled_tasks_from_lot() flash( f"Unité #{next_number} individualisée. Le groupe conserve un total de " f"{equipment.quantity} unité(s).", 'success', ) return redirect(url_for('equipments.detail', id=individual.id)) next_number = max( [child.individual_number or 0 for child in equipment.children] or [0] ) + 1 return render_template( 'equipments/individualize.html', equipment=equipment, next_number=next_number, ) @main_bp.route('//children') @login_required def children(id): """Liste tous les enfants d'un groupe.""" equipment = Equipment.query.get_or_404(id) if not equipment.is_group: flash("Cet équipement n'est pas un groupe.", 'warning') return redirect(url_for('equipments.detail', id=equipment.id)) # Récupérer tous les enfants avec pagination page = request.args.get('page', 1, type=int) children = Equipment.query.filter_by(parent_id=id)\ .order_by(Equipment.individual_number)\ .paginate(page=page, per_page=50) return render_template('equipments/children.html', equipment=equipment, children=children) @main_bp.route('//scheduled-tasks/init', methods=['POST']) @login_required def init_scheduled_tasks(id): """Initialise les tâches planifiées depuis le lot et la catégorie hérités.""" equipment = Equipment.query.get_or_404(id) # Créer les tâches planifiées created_tasks = equipment.create_scheduled_tasks_from_lot() if created_tasks: flash(f'{len(created_tasks)} tâche(s) planifiée(s) créée(s).', 'success') else: flash('Aucune tâche planifiée créée (déjà existantes ou pas de lot/catégorie).', 'info') return redirect(url_for('equipments.scheduled_tasks', id=equipment.id)) @main_bp.route('//move', methods=['GET', 'POST']) @login_required def move_to_room(id): """Déplacer un équipement vers une autre salle.""" from ..core.models.college import Building, Room from ..core.models.equipment import EquipmentRoomHistory equipment = Equipment.query.get_or_404(id) if request.method == 'POST': new_room_id = request.form.get('room_id') if new_room_id: old_room_id = equipment.room_id # Créer l'historique du déplacement if old_room_id != int(new_room_id): history = EquipmentRoomHistory( equipment_id=equipment.id, old_room_id=old_room_id, new_room_id=new_room_id ) db.session.add(history) equipment.room_id = new_room_id db.session.commit() flash(f"Équipement déplacé.", 'success') return redirect(url_for('equipments.detail', id=equipment.id)) buildings = Building.query.order_by(Building.name).all() return render_template('equipments/move.html', equipment=equipment, buildings=buildings) # === Gestion des Zones ===