gmao/app_new/core/routes/setup_wizard.py
root 358f946c76
Some checks are pending
CI - Tests et Syntax / lint-and-test (push) Waiting to run
Maintient l'accès administrateur au setup
2026-08-14 19:25:53 +00:00

645 lines
27 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
Core Routes - Setup Wizard
GMAO Collège - Assistant de configuration initiale
"""
import os
import json
import hmac
from datetime import datetime, timezone
from flask import Blueprint, render_template, redirect, url_for, flash, request, jsonify, current_app, session, abort
from flask_login import login_required, current_user
from werkzeug.security import generate_password_hash
from ...extensions import db, csrf
from ..models.user import User
from ..models.college import College, Building, Zone, Room
from ..models.equipment import EquipmentCategory
from ..models.maintenance import Lot
from ..models.planning import WorkSchedule, CollegeClosure, ClosureWorkDay
from ..setup_catalog import EQUIPMENT_CATEGORIES, LOTS, grouped_catalog
setup_wizard_bp = Blueprint('setup_wizard', __name__, url_prefix='/setup-wizard')
# Dossier contenant les données
DATA_DIR = os.path.join(os.path.dirname(__file__), '..', '..', 'instance')
def is_setup_complete():
"""Vérifie si le setup est déjà terminé."""
return os.path.exists(os.path.join(DATA_DIR, '.setup_complete'))
@setup_wizard_bp.before_request
def protect_initial_setup():
"""Exige un jeton d'environnement tant que l'installation est incomplète."""
# Le panneau de gestion reste accessible aux administrateurs connectés,
# même si le marqueur du volume instance a été perdu lors d'un redéploiement.
if request.endpoint == 'setup_wizard.admin_panel' and current_user.is_authenticated and current_user.is_admin():
return None
if is_setup_complete():
return None
expected_token = os.environ.get('SETUP_TOKEN', '')
if len(expected_token) < 32:
current_app.logger.error('SETUP_TOKEN is missing or too short')
return jsonify({'error': 'Installation sécurisée non configurée'}), 503
supplied_token = (
request.headers.get('X-Setup-Token')
or request.args.get('token', '')
)
if supplied_token and hmac.compare_digest(supplied_token, expected_token):
session['setup_authorized'] = True
if request.method == 'GET' and request.args.get('token'):
return redirect(request.path)
if not session.get('setup_authorized'):
return jsonify({'error': 'Jeton dinstallation requis'}), 403
return None
def load_lots_data():
"""Retourne le catalogue embarqué issu du dump SQL de référence."""
return grouped_catalog()
class SetupProgress(db.Model):
"""Progression du setup wizard."""
__tablename__ = 'setup_progress'
id = db.Column(db.Integer, primary_key=True)
# Numéro de la dernière étape validée. Zéro signifie « aucune ».
step = db.Column(db.Integer, default=0)
admin_username = db.Column(db.String(100), nullable=True)
college_name = db.Column(db.String(200), nullable=True)
college_address = db.Column(db.String(255), nullable=True)
college_zip = db.Column(db.String(10), nullable=True)
college_city = db.Column(db.String(100), nullable=True)
buildings_json = db.Column(db.Text, nullable=True)
lots_json = db.Column(db.Text, nullable=True)
categories_json = db.Column(db.Text, nullable=True)
vacations_json = db.Column(db.Text, nullable=True)
vacances_zone = db.Column(db.String(5), nullable=True)
vacances_annee = db.Column(db.Integer, nullable=True)
ent_configured = db.Column(db.Boolean, default=False)
pronote_qr_code = db.Column(db.Text, nullable=True)
pronote_qr_code_id = db.Column(db.String(50), nullable=True)
pronote_pin = db.Column(db.String(100), 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))
def __repr__(self):
return f"<SetupProgress step={self.step}>"
@staticmethod
def get():
"""Récupère ou crée la progression."""
progress = SetupProgress.query.first()
if not progress:
progress = SetupProgress()
db.session.add(progress)
db.session.commit()
return progress
def get_buildings(self):
"""Récupère la liste des bâtiments."""
if not self.buildings_json:
return []
try:
return json.loads(self.buildings_json)
except:
return []
def set_buildings(self, buildings_list):
"""Sauvegarde la liste des bâtiments."""
self.buildings_json = json.dumps(buildings_list)
def get_lots(self):
"""Récupère la liste des lot."""
if not self.lots_json:
return []
try:
return json.loads(self.lots_json)
except:
return []
def set_lots(self, lots_list):
"""Sauvegarde la liste des lots."""
self.lots_json = json.dumps(lots_list)
def to_dict(self):
"""Convertit en dictionnaire pour le template."""
buildings = [
{'id': item.id, 'name': item.name, 'description': item.description or ''}
for item in Building.query.order_by(Building.name, Building.id).all()
]
zones = [
{'id': item.id, 'name': item.name, 'building_id': item.building_id}
for item in Zone.query.order_by(Zone.building_id, Zone.name, Zone.id).all()
]
rooms = [
{
'id': item.id,
'name': item.name,
'code': '' if item.code in (None, 'None') else item.code,
'building_id': item.building_id,
'zone_id': item.zone_id,
'floor': item.floor or 0,
'room_type_id': item.room_type_id,
}
for item in Room.query.order_by(Room.building_id, Room.name, Room.id).all()
]
vacation_periods = [
{
'id': closure.id,
'name': closure.name,
'start_date': closure.start_date.isoformat(),
'end_date': closure.end_date.isoformat(),
'work_days': [
{
'id': work_day.id,
'work_date': work_day.work_date.isoformat(),
'start_time': work_day.start_time.strftime('%H:%M'),
'end_time': work_day.end_time.strftime('%H:%M'),
'lunch_start': work_day.lunch_start.strftime('%H:%M') if work_day.lunch_start else '',
'lunch_end': work_day.lunch_end.strftime('%H:%M') if work_day.lunch_end else '',
'notes': work_day.notes or '',
}
for work_day in closure.work_days
],
}
for closure in CollegeClosure.query.filter_by(closure_type='vacances')
.order_by(CollegeClosure.start_date, CollegeClosure.id).all()
]
return {
'step': self.step,
'data': {
'admin_username': self.admin_username or '',
'college': {
'name': self.college_name or '',
'address': self.college_address or '',
'zip': self.college_zip or '',
'city': self.college_city or '',
},
'buildings': buildings,
'zones': zones,
'rooms': rooms,
'lots': self.get_lots(),
'categories': json.loads(self.categories_json) if self.categories_json else [],
'vacances_zone': self.vacances_zone or '',
'vacances_annee': self.vacances_annee,
'vacation_periods': vacation_periods,
'ent_configured': self.ent_configured,
}
}
@setup_wizard_bp.route('/')
def index():
"""Page principale du wizard - accessible sans auth."""
if is_setup_complete():
return render_template('setup_wizard/setup_complete.html')
progress = SetupProgress.get()
lots_data = load_lots_data()
return render_template('setup_wizard/index.html',
progress=progress.to_dict(),
lots_data=lots_data)
@setup_wizard_bp.route('/api/progress')
@csrf.exempt
def api_progress():
"""API pour récupérer la progression."""
if is_setup_complete():
return jsonify({'complete': True})
progress = SetupProgress.get()
return jsonify({
'complete': False,
'step': progress.step,
'data': progress.to_dict()
})
@setup_wizard_bp.route('/api/step/<int:step>', methods=['POST'])
@csrf.exempt
def api_save_step(step):
"""Sauvegarde une étape du setup."""
if is_setup_complete():
return jsonify({'error': 'Setup déjà terminé'}), 403
try:
data = request.get_json(silent=True)
if not isinstance(data, dict):
return jsonify({'error': 'Corps JSON invalide'}), 400
progress = SetupProgress.get()
if step == 1:
# Étape 1 : Utilisateur admin
username = data.get('username', '').strip()
password = data.get('password', '').strip()
if not username or len(username) < 3:
return jsonify({'error': 'Le nom d\'utilisateur doit faire au moins 3 caractères'}), 400
if not password or len(password) < 12:
return jsonify({'error': 'Le mot de passe doit faire au moins 12 caractères'}), 400
# Supprimer l'utilisateur admin par défaut s'il existe
default_admin = User.query.filter_by(username='admin').first()
if default_admin and username != 'admin':
db.session.delete(default_admin)
# Créer ou mettre à jour l'utilisateur
user = User.query.filter_by(username=username).first()
if user:
user.set_password(password)
user.role = 'admin'
user.is_active = True
else:
user = User(
username=username,
email=f'{username}@gmao.local',
role='admin',
is_active=True
)
user.set_password(password)
db.session.add(user)
progress.admin_username = username
progress.step = 2
db.session.commit()
return jsonify({'success': True, 'step': 2})
elif step == 2:
# Étape 2 : Collège
name = data.get('name', '').strip()
address = data.get('address', '').strip()
zip_code = data.get('zip', '').strip()
city = data.get('city', '').strip()
if not all([name, address, zip_code, city]):
return jsonify({'error': 'Tous les champs sont requis'}), 400
college = College.query.first()
if college:
college.name = name
college.address = address
college.zip_code = zip_code
college.city = city
else:
college = College(
name=name,
address=address,
zip_code=zip_code,
city=city
)
db.session.add(college)
progress.college_name = name
progress.college_address = address
progress.college_zip = zip_code
progress.college_city = city
progress.step = 3
db.session.commit()
return jsonify({'success': True, 'step': 3})
elif step == 3:
# Étape 3 : Bâtiments
buildings = data.get('buildings', [])
if not isinstance(buildings, list) or not buildings:
return jsonify({'error': 'Ajoutez au moins un bâtiment'}), 400
saved = []
for b in buildings:
if not isinstance(b, dict):
return jsonify({'error': 'Bâtiment invalide'}), 400
name = str(b.get('name') or '').strip()
if not name:
return jsonify({'error': 'Le nom de chaque bâtiment est requis'}), 400
building_id = b.get('id')
building = db.session.get(Building, building_id) if building_id else None
if building is None:
building = Building.query.filter(db.func.lower(Building.name) == name.lower()).first()
if building is None:
building = Building(name=name)
db.session.add(building)
db.session.flush()
building.name = name
building.description = str(b.get('description') or '').strip()
saved.append({'id': building.id, 'name': building.name, 'description': building.description or ''})
progress.set_buildings(saved)
progress.step = max(progress.step or 0, 3)
db.session.commit()
return jsonify({'success': True, 'step': 3, 'buildings': saved})
elif step == 4:
# Étape 4 : Zones. Les absences du formulaire ne suppriment rien.
zones = data.get('zones', [])
if not isinstance(zones, list):
return jsonify({'error': 'Liste de zones invalide'}), 400
saved = []
for item in zones:
if not isinstance(item, dict):
return jsonify({'error': 'Zone invalide'}), 400
name = str(item.get('name') or '').strip()
try:
building_id = int(item.get('building_id'))
except (TypeError, ValueError):
return jsonify({'error': f'Bâtiment invalide pour la zone {name or "sans nom"}'}), 400
if not name or db.session.get(Building, building_id) is None:
return jsonify({'error': 'Chaque zone doit avoir un nom et un bâtiment existant'}), 400
zone_id = item.get('id')
zone = db.session.get(Zone, zone_id) if zone_id else None
if zone is None:
zone = Zone.query.filter(
Zone.building_id == building_id,
db.func.lower(Zone.name) == name.lower(),
).first()
if zone is None:
zone = Zone(name=name, building_id=building_id)
db.session.add(zone)
db.session.flush()
zone.name = name
zone.building_id = building_id
saved.append({'id': zone.id, 'name': zone.name, 'building_id': zone.building_id})
progress.step = max(progress.step or 0, 4)
db.session.commit()
return jsonify({'success': True, 'step': 4, 'zones': saved})
elif step == 5:
# Étape 5 : Salles. Mise à jour/création uniquement.
rooms = data.get('rooms', [])
if not isinstance(rooms, list):
return jsonify({'error': 'Liste de salles invalide'}), 400
saved = []
for item in rooms:
if not isinstance(item, dict):
return jsonify({'error': 'Salle invalide'}), 400
name = str(item.get('name') or '').strip()
try:
building_id = int(item.get('building_id'))
floor = int(item.get('floor') or 0)
except (TypeError, ValueError):
return jsonify({'error': f'Bâtiment ou étage invalide pour {name or "la salle"}'}), 400
building = db.session.get(Building, building_id)
if not name or building is None:
return jsonify({'error': 'Chaque salle doit avoir un nom et un bâtiment existant'}), 400
zone_id = item.get('zone_id') or None
if zone_id is not None:
try:
zone_id = int(zone_id)
except (TypeError, ValueError):
return jsonify({'error': f'Zone invalide pour {name}'}), 400
zone = db.session.get(Zone, zone_id)
if zone is None or zone.building_id != building_id:
return jsonify({'error': f'La zone de {name} ne correspond pas à son bâtiment'}), 400
room_id = item.get('id')
room = db.session.get(Room, room_id) if room_id else None
if room is None:
room = Room.query.filter(
Room.building_id == building_id,
db.func.lower(Room.name) == name.lower(),
).first()
if room is None:
room = Room(name=name, building_id=building_id)
db.session.add(room)
db.session.flush()
room.name = name
room.code = str(item.get('code') or '').strip() or None
room.building_id = building_id
room.zone_id = zone_id
room.floor = floor
saved.append({
'id': room.id, 'name': room.name, 'code': room.code or '',
'building_id': room.building_id, 'zone_id': room.zone_id, 'floor': room.floor,
})
progress.step = max(progress.step or 0, 5)
db.session.commit()
return jsonify({'success': True, 'step': 5, 'rooms': saved})
elif step == 6:
# Étape 6 : catalogue fixe des catégories et lots du dump SQL.
categories_by_name = {}
for name in EQUIPMENT_CATEGORIES:
category = EquipmentCategory.query.filter(
db.func.lower(EquipmentCategory.name) == name.lower()
).first()
if category is None:
category = EquipmentCategory(name=name)
db.session.add(category)
db.session.flush()
categories_by_name[name] = category
for lot_name, category_name in LOTS:
lot = Lot.query.filter(db.func.lower(Lot.name) == lot_name.lower()).first()
if lot is None:
lot = Lot(name=lot_name)
db.session.add(lot)
lot.category_id = categories_by_name[category_name].id
progress.categories_json = json.dumps(list(EQUIPMENT_CATEGORIES), ensure_ascii=False)
progress.set_lots([name for name, _category in LOTS])
progress.step = max(progress.step or 0, 6)
db.session.commit()
return jsonify({
'success': True, 'step': 6,
'categories_count': len(EQUIPMENT_CATEGORIES), 'lots_count': len(LOTS),
})
elif step == 7:
# Étape 7 : Vacances et horaires
zone = data.get('zone', 'A')
annee = data.get('annee', datetime.now().year)
import_vacances = data.get('import_vacances', False)
import_feries = data.get('import_feries', False)
vacation_periods = data.get('vacation_periods', [])
if not isinstance(vacation_periods, list):
return jsonify({'error': 'Liste de vacances invalide'}), 400
saved_periods = []
for period in vacation_periods:
if not isinstance(period, dict):
return jsonify({'error': 'Période de vacances invalide'}), 400
name = str(period.get('name') or '').strip()
try:
start_date = datetime.strptime(period.get('start_date', ''), '%Y-%m-%d').date()
end_date = datetime.strptime(period.get('end_date', ''), '%Y-%m-%d').date()
except (TypeError, ValueError):
return jsonify({'error': f'Dates invalides pour {name or "une période"}'}), 400
if not name or end_date < start_date:
return jsonify({'error': 'Chaque période doit avoir un nom et des dates cohérentes'}), 400
closure = db.session.get(CollegeClosure, period.get('id')) if period.get('id') else None
if closure is None:
closure = CollegeClosure.query.filter_by(
name=name, start_date=start_date, end_date=end_date
).first()
if closure is None:
closure = CollegeClosure(name=name, start_date=start_date, end_date=end_date)
db.session.add(closure)
db.session.flush()
closure.name = name
closure.start_date = start_date
closure.end_date = end_date
closure.closure_type = 'vacances'
# Toute la période est fermée, sauf les dates exactes ci-dessous.
closure.work_hours_type = 'none'
work_days = period.get('work_days', [])
if not isinstance(work_days, list):
db.session.rollback()
return jsonify({'error': f'Jours travaillés invalides pour {name}'}), 400
ClosureWorkDay.query.filter_by(closure_id=closure.id).delete()
seen_dates = set()
saved_days = []
for day in work_days:
if not isinstance(day, dict):
db.session.rollback()
return jsonify({'error': f'Jour travaillé invalide pour {name}'}), 400
try:
work_date = datetime.strptime(day.get('work_date', ''), '%Y-%m-%d').date()
start_time = datetime.strptime(day.get('start_time', ''), '%H:%M').time()
end_time = datetime.strptime(day.get('end_time', ''), '%H:%M').time()
lunch_start = datetime.strptime(day['lunch_start'], '%H:%M').time() if day.get('lunch_start') else None
lunch_end = datetime.strptime(day['lunch_end'], '%H:%M').time() if day.get('lunch_end') else None
except (TypeError, ValueError):
db.session.rollback()
return jsonify({'error': f'Date ou horaires invalides pour un jour travaillé de {name}'}), 400
if not start_date <= work_date <= end_date:
db.session.rollback()
return jsonify({'error': f'Le {work_date:%d/%m/%Y} est hors de la période {name}'}), 400
if work_date in seen_dates or end_time <= start_time:
db.session.rollback()
return jsonify({'error': f'Jour dupliqué ou horaires incohérents dans {name}'}), 400
if (lunch_start is None) != (lunch_end is None) or (
lunch_start and not (start_time <= lunch_start < lunch_end <= end_time)
):
db.session.rollback()
return jsonify({'error': f'Pause déjeuner incohérente le {work_date:%d/%m/%Y}'}), 400
seen_dates.add(work_date)
work_day = ClosureWorkDay(
closure_id=closure.id,
work_date=work_date,
start_time=start_time,
end_time=end_time,
lunch_start=lunch_start,
lunch_end=lunch_end,
notes=str(day.get('notes') or '').strip() or None,
)
db.session.add(work_day)
saved_days.append({'work_date': work_date.isoformat()})
saved_periods.append({'id': closure.id, 'name': name, 'work_days': saved_days})
progress.vacances_zone = zone
progress.vacances_annee = annee
progress.vacations_json = json.dumps({
'import_vacances': bool(import_vacances),
'import_feries': bool(import_feries),
'vacation_periods': saved_periods,
}, ensure_ascii=False)
progress.step = max(progress.step or 0, 7)
db.session.commit()
return jsonify({'success': True, 'step': 7, 'vacation_periods': saved_periods})
elif step in (8, 9):
# Intégrations facultatives : le paramétrage complet reste dans leurs pages dédiées.
progress.step = max(progress.step or 0, step)
db.session.commit()
return jsonify({'success': True, 'step': step, 'skipped': True})
elif step == 10:
progress.step = max(progress.step or 0, 10)
db.session.commit()
return jsonify({'success': True, 'step': 10})
else:
return jsonify({'error': 'Étape invalide'}), 400
except Exception:
db.session.rollback()
current_app.logger.exception('Échec lors de la sauvegarde du wizard')
return jsonify({'error': 'Erreur interne lors de la sauvegarde'}), 500
@setup_wizard_bp.route('/api/complete', methods=['POST'])
@csrf.exempt
def api_complete():
"""Marquer le setup comme terminé."""
if is_setup_complete():
return jsonify({'error': 'Setup déjà terminé'}), 403
try:
progress = SetupProgress.get()
admin = User.query.filter_by(role='admin', is_active=True).first()
if not admin:
return jsonify({'error': 'Un administrateur actif est requis'}), 400
if progress.step < 10:
return jsonify({'error': 'Toutes les étapes doivent être terminées'}), 400
# Créer le fichier .setup_complete
with open(os.path.join(DATA_DIR, '.setup_complete'), 'w') as f:
f.write(datetime.now(timezone.utc).isoformat())
return jsonify({'success': True})
except Exception:
current_app.logger.exception('Échec lors de la finalisation du wizard')
return jsonify({'error': 'Erreur interne lors de la finalisation'}), 500
@setup_wizard_bp.route('/api/test-ent', methods=['POST'])
@csrf.exempt
def api_test_ent():
"""Tester la connexion ENT."""
if is_setup_complete():
return jsonify({'error': 'Setup déjà terminé'}), 403
data = request.get_json(silent=True) or {}
username = data.get('username', '')
password = data.get('password', '')
return jsonify({'success': False, 'error': 'Test ENT non implémenté'}), 501
@setup_wizard_bp.route('/api/test-pronote', methods=['POST'])
@csrf.exempt
def api_test_pronote():
"""Tester la connexion Pronote."""
if is_setup_complete():
return jsonify({'error': 'Setup déjà terminé'}), 403
data = request.get_json(silent=True) or {}
qr_code_data = data.get('qr_code', '')
pin = data.get('pin', '')
return jsonify({'success': False, 'error': 'Test Pronote non implémenté'}), 501
@setup_wizard_bp.route('/admin')
@login_required
def admin_panel():
"""Panneau d'administration."""
if not current_user.is_admin():
abort(403)
from ..models.user import User
from ..models.college import Building, Room
from ..models.equipment import Equipment
stats = {
'users': User.query.count(),
'buildings': Building.query.count(),
'rooms': Room.query.count(),
'equipments': Equipment.query.count()
}
return render_template('setup_wizard/admin_panel.html', stats=stats)