"""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, timezone from app_new.extensions import db from ..core.models.equipment import Equipment, EquipmentCategory, EquipmentDocument, EquipmentRoomHistory, EquipmentRestriction, EquipmentQuantityMovement 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 from app_new.utils import format_status, status_badge_class def _empty_to_none(v): """Convertit une chaîne vide en None.""" return None if v == '' else v def _date_or_none(value): return datetime.strptime(value, '%Y-%m-%d').date() if value else None 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, EquipmentQuantityMovement 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 # Les descendants sont volontairement chargés par l'endpoint hierarchy # uniquement lorsque l'utilisateur déplie une famille. for parent in parents: parent.zones_list = [] parent.individual_count = parent.quantity or 0 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('//hierarchy') @login_required def hierarchy(id): """Charge à la demande la hiérarchie localisée d'une famille.""" from sqlalchemy import select root = Equipment.query.get_or_404(id) descendants = select( Equipment.id, Equipment.parent_id, Equipment.name, Equipment.code, Equipment.status, Equipment.room_id, Equipment.room_id.label('effective_room_id'), ).where(Equipment.parent_id == root.id, Equipment.is_deleted.is_(False)).cte(recursive=True) descendants = descendants.union_all(select( Equipment.id, Equipment.parent_id, Equipment.name, Equipment.code, Equipment.status, Equipment.room_id, db.func.coalesce(Equipment.room_id, descendants.c.effective_room_id).label('effective_room_id'), ).where(Equipment.parent_id == descendants.c.id, Equipment.is_deleted.is_(False))) rows = db.session.execute( select(descendants.c.id, descendants.c.name, descendants.c.code, descendants.c.status, 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) .order_by(Zone.name, Room.name, descendants.c.name) ).all() zones = {} for row in rows: zone_key = row.zone_id or 0 room_key = row.room_id or 0 zone = zones.setdefault(zone_key, {'name': row.zone_name or 'Sans zone', 'rooms': {}}) room = zone['rooms'].setdefault(room_key, {'name': row.room_name or 'Sans salle', 'items': []}) room['items'].append({ 'id': row.id, 'name': row.name, 'code': row.code, 'status': row.status, 'status_label': format_status(row.status), 'status_class': f"bg-{status_badge_class(row.status)}", 'url': url_for('equipments.detail', id=row.id), }) payload = [] for zone in zones.values(): rooms = list(zone['rooms'].values()) payload.append({'name': zone['name'], 'count': sum(len(room['items']) for room in rooms), 'rooms': rooms}) return jsonify({'root_id': root.id, 'zones': payload}) @main_bp.route('/create', methods=['GET', 'POST']) @login_required def create(): """Créer un équipement.""" if request.method == 'GET' and request.args.get('advanced') != '1' and not request.args.get('parent_id'): return redirect(url_for('wizard.equipment_wizard')) # 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 category_id = request.form.get('category_id', type=int) lot_id = request.form.get('lot_id', type=int) if lot_id: lot = db.session.get(Lot, lot_id) if not category_id or not lot or lot.category_id != category_id: flash('Le lot sélectionné doit appartenir à la catégorie choisie.', 'danger') return redirect(url_for('equipments.create', advanced=1)), 400 equipment = Equipment( name=request.form.get('name'), category_id=category_id, room_id=request.form.get('room_id') or None, lot_id=lot_id, 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, mobility=request.form.get('mobility', 'non_precise'), serial_number=_empty_to_none(request.form.get('serial_number', '').strip()), manufacturer=_empty_to_none(request.form.get('manufacturer', '').strip()), model_reference=_empty_to_none(request.form.get('model_reference', '').strip()), supplier_id=request.form.get('supplier_id') or None, supplier_reference=_empty_to_none(request.form.get('supplier_reference', '').strip()), purchase_price=request.form.get('purchase_price') or None, purchase_date=_date_or_none(request.form.get('purchase_date')), recurrence_monitoring=request.form.get('recurrence_monitoring') == '1', ) 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() from ..core.models.company import Company companies = Company.query.order_by(Company.name).all() return render_template('equipments/form.html', categories=categories, rooms=rooms, buildings=buildings, parent_equipments=parent_equipments, lots=lots, statuses=EQUIPMENT_STATUSES, companies=companies, 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) root_id = equipment.parent_id or equipment.id if equipment.parent and equipment.parent.parent_id: root_id = equipment.parent.parent_id quantity_movements = EquipmentQuantityMovement.query.filter_by( root_equipment_id=root_id ).order_by(EquipmentQuantityMovement.moved_at.desc()).limit(50).all() return render_template('equipments/detail.html', equipment=equipment, quantity_movements=quantity_movements) @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': category_id = request.form.get('category_id', type=int) lot_id = request.form.get('lot_id', type=int) if lot_id: lot = db.session.get(Lot, lot_id) if not category_id or not lot or lot.category_id != category_id: flash('Le lot sélectionné doit appartenir à la catégorie choisie.', 'danger') return redirect(url_for('equipments.edit', id=id)), 400 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 = category_id equipment.lot_id = 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')) equipment.mobility = request.form.get('mobility', 'non_precise') equipment.serial_number = _empty_to_none(request.form.get('serial_number', '').strip()) equipment.manufacturer = _empty_to_none(request.form.get('manufacturer', '').strip()) equipment.model_reference = _empty_to_none(request.form.get('model_reference', '').strip()) equipment.supplier_id = request.form.get('supplier_id') or None equipment.supplier_reference = _empty_to_none(request.form.get('supplier_reference', '').strip()) equipment.purchase_price = request.form.get('purchase_price') or None equipment.purchase_date = _date_or_none(request.form.get('purchase_date')) equipment.recurrence_monitoring = bool(request.form.get('recurrence_monitoring')) # 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() from ..core.models.company import Company companies = Company.query.order_by(Company.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, companies=companies, 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//recurrence-policy', methods=['POST']) @login_required def category_recurrence_policy(id): category = EquipmentCategory.query.get_or_404(id) category.recurrence_alert_enabled = request.form.get('enabled') == 'on' category.recurrence_threshold = max(1, request.form.get('threshold', type=int) or 3) category.recurrence_window_days = max(1, request.form.get('window_days', type=int) or 365) db.session.commit() flash('Politique de récurrence mise à jour.', 'success') return redirect(url_for('equipments.category_detail', id=id)) @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' equipment.lifecycle_status = 'mise_au_rebut_planifiee' equipment.planned_disposal_date = _date_or_none(request.form.get('planned_disposal_date')) equipment.disposal_reason = _empty_to_none(request.form.get('reason', '').strip()) 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' equipment.lifecycle_status = 'en_service' equipment.planned_disposal_date = None 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.lifecycle_status = 'jete' equipment.disposal_date = datetime.now().date() 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éplace une unité ou une quantité, sans casser les totaux des groupes.""" equipment = Equipment.query.get_or_404(id) if equipment.mobility == 'fixe': flash("Cet équipement est fixe. Sa relocalisation nécessite une modification de sa fiche.", 'danger') return redirect(url_for('equipments.detail', id=id)) if request.method == 'POST': target_room_id = request.form.get('room_id', type=int) target_room = db.session.get(Room, target_room_id) if target_room_id else None if target_room is None: flash('La salle cible est invalide.', 'danger') return redirect(url_for('equipments.move_to_room', id=id)) reason = _empty_to_none((request.form.get('reason') or '').strip()) if equipment.is_group: quantity = request.form.get('quantity', 1, type=int) if quantity < 1: flash('La quantité doit être positive.', 'danger') return redirect(url_for('equipments.move_to_room', id=id)) root = equipment.parent if equipment.parent_id else equipment if equipment.parent_id: available = equipment.unindividualized_quantity else: allocated = sum(child.quantity or 0 for child in equipment.children if child.is_group) available = max((equipment.quantity or 0) - allocated, 0) if quantity > available: flash(f'Seulement {available} unité(s) non individualisée(s) sont déplaçables.', 'danger') return redirect(url_for('equipments.move_to_room', id=id)) target = Equipment.query.filter_by( parent_id=root.id, room_id=target_room.id, is_group=True, is_deleted=False ).first() if target is None: target = Equipment( name=equipment.name if equipment.parent_id else root.name, parent_id=root.id, room_id=target_room.id, is_group=True, quantity=0, status='en_service', mobility=root.mobility, ) db.session.add(target) db.session.flush() if equipment.parent_id and target.id == equipment.id: flash('La salle cible est déjà la salle source.', 'warning') return redirect(url_for('equipments.move_to_room', id=id)) source_room_id = equipment.room_id if equipment.parent_id else None if equipment.parent_id: equipment.quantity -= quantity target.quantity = (target.quantity or 0) + quantity db.session.add(EquipmentQuantityMovement( root_equipment_id=root.id, source_equipment_id=equipment.id, target_equipment_id=target.id, source_room_id=source_room_id, target_room_id=target_room.id, quantity=quantity, reason=reason, moved_by_id=current_user.id, )) db.session.commit() flash(f'{quantity} unité(s) déplacée(s) vers {target_room.name}.', 'success') return redirect(url_for('equipments.detail', id=root.id)) old_room_id = equipment.effective_room.id if equipment.effective_room else None if old_room_id == target_room.id: flash('Cet équipement est déjà dans cette salle.', 'warning') return redirect(url_for('equipments.move_to_room', id=id)) equipment.room_id = target_room.id if equipment.parent_id and equipment.parent and equipment.parent.parent_id: root = equipment.parent.parent target_group = Equipment.query.filter_by( parent_id=root.id, room_id=target_room.id, is_group=True, is_deleted=False ).first() if target_group: equipment.parent_id = target_group.id db.session.add(EquipmentRoomHistory( equipment_id=equipment.id, old_room_id=old_room_id, new_room_id=target_room.id, changed_by_id=current_user.id, reason=reason, )) db.session.commit() flash('Équipement déplacé.', 'success') return redirect(url_for('equipments.detail', id=equipment.id)) rooms = Room.query.join(Building).order_by(Building.name, Room.name).all() return render_template( 'equipments/move.html', equipment=equipment, rooms=rooms, current_room=equipment.effective_room, max_quantity=(equipment.unindividualized_quantity if equipment.parent_id else max( (equipment.quantity or 0) - sum(c.quantity or 0 for c in equipment.children if c.is_group), 0 )) if equipment.is_group else 1, ) @main_bp.route('//repair/start', methods=['POST']) @login_required def start_repair(id): equipment = Equipment.query.get_or_404(id) equipment.lifecycle_status = 'en_reparation' equipment.status = 'en_maintenance' equipment.unavailable_since = datetime.now(timezone.utc) db.session.commit() flash('Équipement retiré provisoirement pour réparation.', 'warning') return redirect(url_for('equipments.detail', id=id)) @main_bp.route('//repair/complete', methods=['POST']) @login_required def complete_repair(id): equipment = Equipment.query.get_or_404(id) equipment.lifecycle_status = 'en_service' equipment.status = 'en_service' equipment.unavailable_since = None db.session.commit() flash('Réparation terminée, équipement remis en service.', 'success') return redirect(url_for('equipments.detail', id=id)) # === Gestion des Zones ===