402 lines
14 KiB
Python
402 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)
|