403 lines
14 KiB
Python
403 lines
14 KiB
Python
|
"""
|
|||
|
|
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
|
|||
|
|
from ..models.maintenance import Lot
|
|||
|
|
from ..models.planning import WorkSchedule, CollegeClosure
|
|||
|
|
|
|||
|
|
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."""
|
|||
|
|
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 d’installation requis'}), 403
|
|||
|
|
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
|
|||
|
|
def load_lots_data():
|
|||
|
|
"""Charge les données des lots depuis le fichier JSON."""
|
|||
|
|
json_path = os.path.join(os.path.dirname(__file__), '..', '..', 'data', 'lots_categories_tasks.json')
|
|||
|
|
try:
|
|||
|
|
with open(json_path, 'r', encoding='utf-8') as f:
|
|||
|
|
return json.load(f)
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"Erreur chargement lots: {e}")
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
|
|||
|
|
class SetupProgress(db.Model):
|
|||
|
|
"""Progression du setup wizard."""
|
|||
|
|
__tablename__ = 'setup_progress'
|
|||
|
|
|
|||
|
|
id = db.Column(db.Integer, primary_key=True)
|
|||
|
|
step = db.Column(db.Integer, default=1)
|
|||
|
|
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."""
|
|||
|
|
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': self.get_buildings(),
|
|||
|
|
'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,
|
|||
|
|
'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()
|
|||
|
|
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', [])
|
|||
|
|
|
|||
|
|
# Supprimer les anciens bâtiments
|
|||
|
|
Building.query.delete()
|
|||
|
|
|
|||
|
|
for b in buildings:
|
|||
|
|
building = Building(
|
|||
|
|
name=b.get('name'),
|
|||
|
|
code=b.get('code', ''),
|
|||
|
|
floors=b.get('floors', 1)
|
|||
|
|
)
|
|||
|
|
db.session.add(building)
|
|||
|
|
|
|||
|
|
progress.set_buildings(buildings)
|
|||
|
|
progress.step = 4
|
|||
|
|
db.session.commit()
|
|||
|
|
|
|||
|
|
return jsonify({'success': True, 'step': 4})
|
|||
|
|
|
|||
|
|
elif step == 4:
|
|||
|
|
# Étape 4 : Lots/Catégories
|
|||
|
|
categories = data.get('categories', [])
|
|||
|
|
|
|||
|
|
# Créer les lots sélectionnés
|
|||
|
|
for cat in categories:
|
|||
|
|
lot = Lot.query.filter_by(name=cat).first()
|
|||
|
|
if not lot:
|
|||
|
|
lot = Lot(name=cat, description=f'Lot {cat}')
|
|||
|
|
db.session.add(lot)
|
|||
|
|
|
|||
|
|
progress.categories_json = json.dumps(categories)
|
|||
|
|
progress.step = 5
|
|||
|
|
db.session.commit()
|
|||
|
|
|
|||
|
|
return jsonify({'success': True, 'step': 5})
|
|||
|
|
|
|||
|
|
elif step == 5:
|
|||
|
|
# Étape 5 : Vacances et horaires
|
|||
|
|
zone = data.get('zone', 'A')
|
|||
|
|
annee = data.get('annee', 2024)
|
|||
|
|
import_vacances = data.get('import_vacances', False)
|
|||
|
|
import_feries = data.get('import_feries', False)
|
|||
|
|
work_schedule = data.get('work_schedule', {})
|
|||
|
|
custom_closures = data.get('custom_closures', [])
|
|||
|
|
personal_leaves = data.get('personal_leaves', [])
|
|||
|
|
|
|||
|
|
progress.vacances_zone = zone
|
|||
|
|
progress.vacances_annee = annee
|
|||
|
|
progress.vacations_json = json.dumps({
|
|||
|
|
'work_schedule': work_schedule,
|
|||
|
|
'custom_closures': custom_closures,
|
|||
|
|
'personal_leaves': personal_leaves
|
|||
|
|
})
|
|||
|
|
progress.step = 6
|
|||
|
|
db.session.commit()
|
|||
|
|
|
|||
|
|
return jsonify({'success': True, 'step': 6})
|
|||
|
|
|
|||
|
|
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 < 6:
|
|||
|
|
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)
|