1205 lines
52 KiB
Python
1205 lines
52 KiB
Python
"""
|
||
Core Routes - Administration
|
||
GMAO Collège
|
||
"""
|
||
from flask import Blueprint, render_template, redirect, url_for, request, flash
|
||
from pathlib import Path
|
||
import hashlib
|
||
import secrets
|
||
from werkzeug.security import generate_password_hash
|
||
from flask_login import login_required, current_user
|
||
from ...extensions import db
|
||
from ..models.user import User, Staff
|
||
from ..models.college import College, Building, Room, RoomType
|
||
from ..models.company import Company, Service
|
||
from ..models.rbac import Role, Permission, UserRole, RolePermission, UserPermission
|
||
from app_new.constants import INTERVENTION_STATUSES, EQUIPMENT_STATUSES, ROLES
|
||
from ..authorization import PERMISSIONS_BY_ROLE, ROLE_LABELS, canonical_role
|
||
|
||
admin_bp = Blueprint('admin', __name__)
|
||
|
||
|
||
def admin_required(f):
|
||
"""Décorateur pour vérifier les droits admin."""
|
||
from functools import wraps
|
||
@wraps(f)
|
||
def decorated_function(*args, **kwargs):
|
||
if not current_user.is_authenticated or not current_user.is_admin():
|
||
flash('Accès réservé aux administrateurs.', 'danger')
|
||
return redirect(url_for('dashboard.index'))
|
||
return f(*args, **kwargs)
|
||
return decorated_function
|
||
|
||
|
||
@admin_bp.route('/')
|
||
@login_required
|
||
@admin_required
|
||
def index():
|
||
"""Page d'administration."""
|
||
stats = {
|
||
'users_count': User.query.count(),
|
||
'staff_count': Staff.query.count(),
|
||
'colleges_count': College.query.count(),
|
||
'buildings_count': Building.query.count(),
|
||
'rooms_count': Room.query.count(),
|
||
'services_count': Service.query.count(),
|
||
'companies_count': Company.query.count(),
|
||
}
|
||
return render_template('admin/index.html', stats=stats)
|
||
|
||
|
||
@admin_bp.route('/users')
|
||
@login_required
|
||
@admin_required
|
||
def users():
|
||
"""Liste des utilisateurs."""
|
||
page = request.args.get('page', 1, type=int)
|
||
users = User.query.order_by(User.created_at.desc()).paginate(page=page, per_page=20)
|
||
return render_template('admin/users.html', users=users, rbac_roles=Role.query.filter_by(is_active=True).order_by(Role.name).all())
|
||
|
||
|
||
@admin_bp.route('/users/new', methods=['GET', 'POST'])
|
||
@login_required
|
||
@admin_required
|
||
def user_new():
|
||
"""Créer un utilisateur."""
|
||
if request.method == 'POST':
|
||
username = (request.form.get('username') or '').strip()
|
||
email = (request.form.get('email') or '').strip()
|
||
password = request.form.get('password') or ''
|
||
password_confirm = request.form.get('password_confirm') or ''
|
||
role = request.form.get('role', 'demandeur')
|
||
valid_roles = set(ROLES) | {r.slug for r in Role.query.filter_by(is_active=True).all()}
|
||
|
||
if len(username) < 3:
|
||
flash('Le nom d’utilisateur doit contenir au moins 3 caractères.', 'danger')
|
||
elif not email:
|
||
flash('L’adresse email est obligatoire.', 'danger')
|
||
elif len(password) < 12:
|
||
flash('Le mot de passe doit contenir au moins 12 caractères.', 'danger')
|
||
elif password != password_confirm:
|
||
flash('Les mots de passe ne correspondent pas.', 'danger')
|
||
elif role not in valid_roles:
|
||
flash('Rôle utilisateur invalide.', 'danger')
|
||
elif User.query.filter_by(username=username).first():
|
||
flash('Ce nom d\'utilisateur existe déjà.', 'danger')
|
||
elif User.query.filter_by(email=email).first():
|
||
flash('Cet email existe déjà.', 'danger')
|
||
else:
|
||
user = User(username=username, email=email, role=role)
|
||
user.set_password(password)
|
||
db.session.add(user)
|
||
db.session.flush()
|
||
selected_role = Role.query.filter_by(slug=role, is_active=True).first()
|
||
if selected_role:
|
||
db.session.add(UserRole(user_id=user.id, role_id=selected_role.id, assigned_by_id=current_user.id))
|
||
db.session.commit()
|
||
flash('Utilisateur créé avec succès.', 'success')
|
||
return redirect(url_for('admin.users'))
|
||
|
||
roles = {**ROLES, **{r.slug: r.name for r in Role.query.filter_by(is_active=True).all()}}
|
||
return render_template('admin/user_form.html', roles=roles)
|
||
|
||
|
||
@admin_bp.route('/users/<int:user_id>/edit', methods=['GET', 'POST'])
|
||
@login_required
|
||
@admin_required
|
||
def user_edit(user_id):
|
||
"""Modifier un utilisateur."""
|
||
user = User.query.get_or_404(user_id)
|
||
|
||
if request.method == 'POST':
|
||
role = request.form.get('role')
|
||
valid_roles = set(ROLES) | {r.slug for r in Role.query.filter_by(is_active=True).all()}
|
||
if role not in valid_roles:
|
||
flash('Rôle utilisateur invalide.', 'danger')
|
||
return render_template('admin/user_form.html', user=user, roles=ROLES), 400
|
||
user.email = request.form.get('email')
|
||
user.role = role
|
||
user.is_active = request.form.get('is_active') == 'on'
|
||
|
||
new_password = request.form.get('new_password')
|
||
if new_password:
|
||
if len(new_password) < 12:
|
||
flash('Le nouveau mot de passe doit contenir au moins 12 caractères.', 'danger')
|
||
return render_template('admin/user_form.html', user=user), 400
|
||
user.set_password(new_password)
|
||
|
||
selected_role = Role.query.filter_by(slug=role, is_active=True).first()
|
||
if selected_role and not UserRole.query.filter_by(user_id=user.id, role_id=selected_role.id).first():
|
||
db.session.add(UserRole(user_id=user.id, role_id=selected_role.id, assigned_by_id=current_user.id))
|
||
db.session.commit()
|
||
flash('Utilisateur modifié avec succès.', 'success')
|
||
return redirect(url_for('admin.users'))
|
||
|
||
roles = {**ROLES, **{r.slug: r.name for r in Role.query.filter_by(is_active=True).all()}}
|
||
return render_template('admin/user_form.html', user=user, roles=roles)
|
||
|
||
|
||
@admin_bp.route('/permissions')
|
||
@login_required
|
||
@admin_required
|
||
def permissions():
|
||
"""Liste des permissions."""
|
||
from ..models.user import User
|
||
users_by_role = {role: [] for role in ROLES}
|
||
for user in User.query.order_by(User.full_name).all():
|
||
users_by_role[canonical_role(user.role)].append(user)
|
||
return render_template(
|
||
'admin/permissions.html', roles=ROLES, role_labels=ROLE_LABELS,
|
||
users_by_role=users_by_role, permissions_by_role=PERMISSIONS_BY_ROLE,
|
||
)
|
||
|
||
|
||
@admin_bp.route('/roles', methods=['GET', 'POST'])
|
||
@login_required
|
||
@admin_required
|
||
def roles():
|
||
"""Gérer les rôles configurables sans supprimer les rôles historiques."""
|
||
if request.method == 'POST':
|
||
name = (request.form.get('name') or '').strip()
|
||
slug = (request.form.get('slug') or '').strip().lower().replace(' ', '_')
|
||
if not name or len(slug) < 2:
|
||
flash('Le nom et le code du rôle sont obligatoires.', 'danger')
|
||
elif Role.query.filter_by(slug=slug).first():
|
||
flash('Ce code de rôle existe déjà.', 'danger')
|
||
else:
|
||
db.session.add(Role(slug=slug, name=name, description=request.form.get('description'), is_system=False))
|
||
db.session.commit()
|
||
flash('Rôle créé. Configurez maintenant ses permissions.', 'success')
|
||
return redirect(url_for('admin.role_permissions', role_id=Role.query.filter_by(slug=slug).first().id))
|
||
return render_template('admin/roles.html', roles_list=Role.query.order_by(Role.is_system.desc(), Role.name).all())
|
||
|
||
|
||
@admin_bp.route('/roles/<int:role_id>/permissions', methods=['GET', 'POST'])
|
||
@login_required
|
||
@admin_required
|
||
def role_permissions(role_id):
|
||
role = Role.query.get_or_404(role_id)
|
||
permissions_list = Permission.query.filter_by(is_active=True).order_by(Permission.module, Permission.action, Permission.code).all()
|
||
if request.method == 'POST':
|
||
selected = {int(value) for value in request.form.getlist('permission_ids')}
|
||
RolePermission.query.filter_by(role_id=role.id).delete(synchronize_session=False)
|
||
for permission in permissions_list:
|
||
if permission.id in selected:
|
||
db.session.add(RolePermission(role_id=role.id, permission_id=permission.id, effect='allow'))
|
||
db.session.commit()
|
||
flash(f'Permissions du rôle « {role.name} » enregistrées.', 'success')
|
||
return redirect(url_for('admin.roles'))
|
||
selected = {link.permission_id for link in role.permission_links if link.effect == 'allow'}
|
||
return render_template('admin/role_permissions.html', role=role, permissions_list=permissions_list, selected=selected)
|
||
|
||
|
||
@admin_bp.route('/roles/<int:role_id>/archive', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def role_archive(role_id):
|
||
role = Role.query.get_or_404(role_id)
|
||
if role.is_system:
|
||
flash('Les rôles système ne peuvent pas être supprimés.', 'danger')
|
||
else:
|
||
role.is_active = False
|
||
db.session.commit()
|
||
flash(f'Rôle « {role.name} » archivé. Son historique est conservé.', 'success')
|
||
return redirect(url_for('admin.roles'))
|
||
|
||
|
||
@admin_bp.route('/users/<int:user_id>/access', methods=['GET', 'POST'])
|
||
@login_required
|
||
@admin_required
|
||
def user_access(user_id):
|
||
user = User.query.get_or_404(user_id)
|
||
roles_list = Role.query.filter_by(is_active=True).order_by(Role.name).all()
|
||
permissions_list = Permission.query.filter_by(is_active=True).order_by(Permission.module, Permission.action, Permission.code).all()
|
||
if request.method == 'POST':
|
||
selected_roles = {int(value) for value in request.form.getlist('role_ids')}
|
||
UserRole.query.filter_by(user_id=user.id).delete(synchronize_session=False)
|
||
for role in roles_list:
|
||
if role.id in selected_roles:
|
||
db.session.add(UserRole(user_id=user.id, role_id=role.id, assigned_by_id=current_user.id))
|
||
# Les exceptions explicites sont optionnelles et tracées ; une case
|
||
# « refuser » est prioritaire sur les rôles.
|
||
UserPermission.query.filter_by(user_id=user.id).delete(synchronize_session=False)
|
||
for permission in permissions_list:
|
||
effect = request.form.get(f'permission_{permission.id}')
|
||
if effect in ('allow', 'deny'):
|
||
db.session.add(UserPermission(user_id=user.id, permission_id=permission.id, effect=effect, note='Exception administrateur'))
|
||
if selected_roles:
|
||
user.role = next((r.slug for r in roles_list if r.id in selected_roles), user.role)
|
||
db.session.commit()
|
||
flash(f'Accès de {user.username} enregistrés.', 'success')
|
||
return redirect(url_for('admin.users'))
|
||
selected_roles = {link.role_id for link in user.role_links}
|
||
overrides = {link.permission_id: link.effect for link in user.permission_links}
|
||
return render_template('admin/user_access.html', user=user, roles_list=roles_list, permissions_list=permissions_list, selected_roles=selected_roles, overrides=overrides)
|
||
|
||
|
||
@admin_bp.route('/audit')
|
||
@login_required
|
||
@admin_required
|
||
def audit_logs():
|
||
"""Consultation du journal immuable des écritures HTTP."""
|
||
from ..models.audit import AuditLog
|
||
page = request.args.get('page', 1, type=int)
|
||
logs = AuditLog.query.order_by(AuditLog.created_at.desc()).paginate(page=page, per_page=50)
|
||
return render_template('admin/audit_logs.html', logs=logs)
|
||
|
||
|
||
@admin_bp.route('/users/<int:id>/permissions', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def update_user_permissions(id):
|
||
"""Change le rôle d'un utilisateur (les permissions sont définies en code)."""
|
||
user = User.query.get_or_404(id)
|
||
role = request.form.get('role')
|
||
if role not in ROLES:
|
||
flash('Rôle invalide.', 'danger')
|
||
return redirect(url_for('admin.permissions'))
|
||
user.role = role
|
||
db.session.commit()
|
||
flash(f'Rôle de {user.username} mis à jour.', 'success')
|
||
return redirect(url_for('admin.permissions'))
|
||
|
||
|
||
@admin_bp.route('/users/list')
|
||
@login_required
|
||
@admin_required
|
||
def list_users():
|
||
"""Alias pour la liste des utilisateurs."""
|
||
return redirect(url_for('admin.users'))
|
||
|
||
|
||
@admin_bp.route('/users/<int:user_id>/delete', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def user_delete(user_id):
|
||
"""Supprimer un utilisateur."""
|
||
user = User.query.get_or_404(user_id)
|
||
|
||
if user.id == current_user.id:
|
||
flash('Vous ne pouvez pas supprimer votre propre compte.', 'danger')
|
||
else:
|
||
db.session.delete(user)
|
||
db.session.commit()
|
||
flash('Utilisateur supprimé.', 'success')
|
||
|
||
return redirect(url_for('admin.users'))
|
||
|
||
|
||
@admin_bp.route('/staff')
|
||
@login_required
|
||
@admin_required
|
||
def staff():
|
||
"""Liste du personnel."""
|
||
page = request.args.get('page', 1, type=int)
|
||
staff_list = Staff.query.order_by(Staff.last_name).paginate(page=page, per_page=20)
|
||
return render_template('admin/staff.html', staff_list=staff_list)
|
||
|
||
|
||
@admin_bp.route('/room-types')
|
||
@login_required
|
||
@admin_required
|
||
def room_types():
|
||
"""Liste des types de salles."""
|
||
room_types_list = RoomType.query.order_by(RoomType.name).all()
|
||
return render_template('admin/room_types.html', room_types=room_types_list)
|
||
|
||
|
||
@admin_bp.route('/room-types/new', methods=['GET', 'POST'])
|
||
@login_required
|
||
@admin_required
|
||
def room_type_new():
|
||
"""Redirige l'ancien écran vers le référentiel canonique."""
|
||
return redirect(url_for('room_types.new'), code=307 if request.method == 'POST' else 302)
|
||
|
||
|
||
@admin_bp.route('/services')
|
||
@login_required
|
||
@admin_required
|
||
def services():
|
||
"""Liste des services."""
|
||
services_list = Service.query.order_by(Service.name).all()
|
||
return render_template('admin/services.html', services=services_list)
|
||
|
||
|
||
@admin_bp.route('/settings')
|
||
@login_required
|
||
@admin_required
|
||
def settings():
|
||
"""Paramètres de l'application."""
|
||
college = College.query.first()
|
||
from ..models.settings import AppSettings
|
||
openrouter_key = AppSettings.get_openrouter_api_key()
|
||
return render_template('admin/settings.html', college=college, openrouter_key=openrouter_key)
|
||
|
||
|
||
@admin_bp.route('/settings/api', methods=['GET', 'POST'])
|
||
@login_required
|
||
@admin_required
|
||
def settings_api():
|
||
"""API pour les paramètres."""
|
||
from ..models.settings import AppSettings
|
||
from flask import jsonify
|
||
|
||
if request.method == 'POST':
|
||
data = request.get_json()
|
||
key = data.get('key')
|
||
value = data.get('value')
|
||
|
||
if not key:
|
||
return jsonify({'error': 'Clé manquante'}), 400
|
||
|
||
AppSettings.set(key, value)
|
||
return jsonify({'success': True, 'message': f'Paramètre {key} sauvegardé'})
|
||
|
||
# GET - retourne les paramètres
|
||
key = request.args.get('key')
|
||
if key:
|
||
value = AppSettings.get(key)
|
||
return jsonify({'key': key, 'value': value})
|
||
|
||
# Retourner tous les paramètres
|
||
settings_list = AppSettings.query.all()
|
||
return jsonify({
|
||
'settings': [{'key': s.key, 'value': s.value, 'description': s.description}
|
||
for s in settings_list]
|
||
})
|
||
|
||
|
||
@admin_bp.route('/settings/openrouter', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def settings_openrouter():
|
||
"""Sauvegarde la clé API OpenRouter."""
|
||
from ..models.settings import AppSettings
|
||
from flask import jsonify
|
||
|
||
data = request.get_json()
|
||
api_key = data.get('api_key')
|
||
|
||
if not api_key:
|
||
return jsonify({'error': 'Clé API manquante'}), 400
|
||
|
||
AppSettings.set_openrouter_api_key(api_key)
|
||
return jsonify({'success': True, 'message': 'Clé API OpenRouter sauvegardée'})
|
||
|
||
|
||
@admin_bp.route('/settings/api-key', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def generate_api_key():
|
||
"""Génère une clé API affichée une seule fois et stockée hachée."""
|
||
from ..models.settings import AppSettings
|
||
raw_key = f"gmao_{secrets.token_urlsafe(32)}"
|
||
AppSettings.set(
|
||
'api_key_hash', generate_password_hash(raw_key),
|
||
description='Empreinte de la clé API REST GMAO',
|
||
)
|
||
return render_template('admin/api_key_created.html', api_key=raw_key)
|
||
|
||
|
||
@admin_bp.route('/settings/api-key/revoke', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def revoke_api_key():
|
||
from ..models.settings import AppSettings
|
||
AppSettings.set('api_key_hash', '', description='Clé API REST révoquée')
|
||
flash('Clé API révoquée.', 'success')
|
||
return redirect(url_for('admin.settings'))
|
||
|
||
|
||
@admin_bp.route('/setup-wizard')
|
||
@login_required
|
||
@admin_required
|
||
def setup_wizard_redirect():
|
||
"""Redirection vers le setup wizard."""
|
||
return redirect(url_for('setup_wizard.index'))
|
||
|
||
|
||
def _template_inventory():
|
||
"""Inventorie les templates globaux et ceux propres aux modules."""
|
||
from flask import current_app
|
||
from ..models.audit import TemplateAuditMark
|
||
|
||
app_root = Path(current_app.root_path).resolve()
|
||
domain_aliases = {'admin': 'core', 'auth': 'core', 'dashboard': 'core', 'setup_wizard': 'core'}
|
||
page_urls = {
|
||
'chatbot/index.html': '/chatbot/',
|
||
'companies/index.html': '/companies/', 'companies/detail.html': '/companies/2',
|
||
'ent/index.html': '/ent/', 'ent/messages.html': '/ent/messages', 'ent/staff.html': '/ent/staff',
|
||
'equipments/index.html': '/equipments/', 'equipments/detail.html': '/equipments/42',
|
||
'equipments/categories.html': '/equipments/categories', 'equipments/buildings.html': '/equipments/buildings/',
|
||
'equipments/rooms.html': '/equipments/rooms/', 'equipments/room_detail.html': '/equipments/rooms/12',
|
||
'interventions/index.html': '/interventions/', 'interventions/detail.html': '/interventions/39',
|
||
'lots/index.html': '/lots/', 'lots/detail.html': '/lots/3',
|
||
'parts/index.html': '/parts/', 'parts/detail.html': '/parts/4',
|
||
'planning/index.html': '/planning/', 'pronote/index.html': '/pronote/',
|
||
'room_types/index.html': '/room-types/', 'training/index.html': '/training/',
|
||
'training/detail.html': '/training/6', 'yeastar/index.html': '/yeastar/',
|
||
'gmao_config/index.html': '/gmao-config/', 'admin/index.html': '/admin/',
|
||
}
|
||
files = []
|
||
for path in sorted(app_root.rglob('*.html')):
|
||
relative = path.relative_to(app_root).as_posix()
|
||
parts = path.relative_to(app_root).parts
|
||
if 'templates' not in parts:
|
||
continue
|
||
template_index = parts.index('templates')
|
||
inner_parts = parts[template_index + 1:]
|
||
if template_index == 0:
|
||
logical_key = '/'.join(inner_parts)
|
||
source = 'global'
|
||
else:
|
||
module = parts[0]
|
||
logical_key = '/'.join(inner_parts)
|
||
if not logical_key.startswith(f'{module}/'):
|
||
logical_key = f'{module}/{logical_key}'
|
||
source = 'module'
|
||
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
||
files.append({
|
||
'path': relative,
|
||
'logical_key': logical_key,
|
||
'source': source,
|
||
'hash': digest,
|
||
'page_url': page_urls.get(logical_key),
|
||
'target_path': (
|
||
f"app_new/{parts[0]}/templates/{logical_key}"
|
||
if template_index > 0
|
||
else (
|
||
f"app_new/templates/{logical_key}"
|
||
if logical_key in {'base.html'} or logical_key.startswith('errors/') or logical_key.startswith('components/')
|
||
else f"app_new/{domain_aliases.get(logical_key.split('/', 1)[0], logical_key.split('/', 1)[0])}/templates/{logical_key}"
|
||
)
|
||
),
|
||
})
|
||
|
||
by_key = {}
|
||
for item in files:
|
||
by_key.setdefault(item['logical_key'], []).append(item)
|
||
marks = {mark.path: mark for mark in TemplateAuditMark.query.all()}
|
||
for item in files:
|
||
group = by_key[item['logical_key']]
|
||
hashes = {candidate['hash'] for candidate in group}
|
||
item['duplicate_count'] = len(group)
|
||
item['comparison'] = 'identique' if len(group) > 1 and len(hashes) == 1 else ('divergent' if len(group) > 1 else 'unique')
|
||
# Le chargeur global gagne habituellement. Le détail intervention est
|
||
# explicitement chargé depuis son module par interventions.detail.
|
||
item['probable_active'] = (
|
||
item['source'] == 'global' or len(group) == 1
|
||
)
|
||
if item['logical_key'] == 'interventions/detail.html':
|
||
item['probable_active'] = item['path'] == 'interventions/templates/detail.html'
|
||
item['mark'] = marks.get(item['path'])
|
||
return files, by_key
|
||
|
||
|
||
TEMPLATE_ARCHITECTURE = [
|
||
{
|
||
'path': 'app_new/templates/',
|
||
'label': 'Socle partagé',
|
||
'description': 'base.html, erreurs, composants et fragments réutilisables. Aucun écran métier ne doit être dupliqué ici.',
|
||
'children': ['base.html', 'errors/', 'components/'],
|
||
},
|
||
{
|
||
'path': 'app_new/<module>/templates/<module>/',
|
||
'label': 'Templates par domaine',
|
||
'description': 'Un domaine métier possède ses écrans canoniques, regroupés sous son propre préfixe.',
|
||
'children': ['core/', 'equipment/', 'buildings/', 'interventions/', 'planning/', 'lots/', 'companies/', 'training/'],
|
||
},
|
||
{
|
||
'path': 'app_new/<module>/templates/',
|
||
'label': 'Zone de transition',
|
||
'description': 'Ancienne convention actuellement présente dans le projet. À migrer progressivement vers le sous-dossier <module>/.',
|
||
'children': ['detail.html', 'index.html', 'form.html'],
|
||
},
|
||
{
|
||
'path': 'app_new/static/',
|
||
'label': 'Ressources visuelles',
|
||
'description': 'CSS, JavaScript et icônes partagés ; pas de logique métier dans les templates.',
|
||
'children': ['css/', 'js/', 'icons/'],
|
||
},
|
||
]
|
||
|
||
|
||
@admin_bp.route('/debug/templates')
|
||
@login_required
|
||
@admin_required
|
||
def template_debug():
|
||
"""Affiche les doublons et les décisions de nettoyage des templates."""
|
||
files, by_key = _template_inventory()
|
||
selected_filter = request.args.get('filter', 'duplicates')
|
||
treated = lambda item: item['mark'] and item['mark'].status == 'obsolete'
|
||
if selected_filter == 'obsolete':
|
||
visible_files = [item for item in files if item['mark'] and item['mark'].status == 'obsolete']
|
||
elif selected_filter == 'identical':
|
||
visible_files = [item for item in files if not treated(item) and item['duplicate_count'] > 1 and item['comparison'] == 'identique']
|
||
elif selected_filter == 'divergent':
|
||
visible_files = [item for item in files if not treated(item) and item['duplicate_count'] > 1 and item['comparison'] == 'divergent']
|
||
elif selected_filter == 'all':
|
||
visible_files = [item for item in files if not treated(item)]
|
||
else:
|
||
selected_filter = 'duplicates'
|
||
visible_files = [item for item in files if not treated(item) and item['duplicate_count'] > 1]
|
||
stats = {
|
||
'total': len(files),
|
||
'duplicate_groups': sum(1 for group in by_key.values() if len(group) > 1),
|
||
'identical_groups': sum(1 for group in by_key.values() if len(group) > 1 and len({item['hash'] for item in group}) == 1),
|
||
'divergent_groups': sum(1 for group in by_key.values() if len(group) > 1 and len({item['hash'] for item in group}) > 1),
|
||
'obsolete': sum(1 for item in files if item['mark'] and item['mark'].status == 'obsolete'),
|
||
}
|
||
groups = {}
|
||
for item in visible_files:
|
||
groups.setdefault(item['logical_key'], []).append(item)
|
||
return render_template('admin/template_debug.html', groups=groups, stats=stats,
|
||
selected_filter=selected_filter, template_architecture=TEMPLATE_ARCHITECTURE)
|
||
|
||
|
||
@admin_bp.route('/debug/templates/preview')
|
||
@login_required
|
||
@admin_required
|
||
def template_debug_preview():
|
||
"""Prévisualise un template autorisé avec un contexte volontairement neutre."""
|
||
from flask import current_app
|
||
from jinja2 import ChoiceLoader, FileSystemLoader, TemplateNotFound, ChainableUndefined
|
||
|
||
relative_path = (request.args.get('path') or '').strip()
|
||
demo_mode = request.args.get('demo', '1') != '0'
|
||
files, _ = _template_inventory()
|
||
item = next((candidate for candidate in files if candidate['path'] == relative_path), None)
|
||
if not item:
|
||
flash('Template introuvable dans l’inventaire.', 'danger')
|
||
return redirect(url_for('admin.template_debug'))
|
||
|
||
app_root = Path(current_app.root_path).resolve()
|
||
source_path = (app_root / relative_path).resolve()
|
||
source = source_path.read_text(encoding='utf-8', errors='replace')
|
||
rendered = None
|
||
render_error = None
|
||
try:
|
||
parts = Path(relative_path).parts
|
||
template_index = parts.index('templates')
|
||
inner_name = '/'.join(parts[template_index + 1:])
|
||
if item['source'] == 'module':
|
||
module_root = app_root / parts[0] / 'templates'
|
||
loader = ChoiceLoader([FileSystemLoader(str(module_root)), current_app.jinja_loader])
|
||
environment = current_app.jinja_env.overlay(loader=loader)
|
||
else:
|
||
environment = current_app.jinja_env.overlay()
|
||
# Un contexte de prévisualisation peut laisser les variables métier
|
||
# vides : elles doivent produire un rendu partiel plutôt qu'une 500.
|
||
environment.undefined = ChainableUndefined
|
||
template = environment.get_template(inner_name)
|
||
preview_context = {}
|
||
# Reproduire les variables globales Flask (current_user, config,
|
||
# request...) sans injecter d'objet métier réel dans la prévisualisation.
|
||
current_app.update_template_context(preview_context)
|
||
if demo_mode:
|
||
from types import SimpleNamespace
|
||
|
||
class DemoValue(str):
|
||
"""Valeur vide pouvant aussi être appelée par un template."""
|
||
def __call__(self, *args, **kwargs):
|
||
return ''
|
||
|
||
def __getattr__(self, name):
|
||
return self
|
||
|
||
def __int__(self):
|
||
return 0
|
||
|
||
def __float__(self):
|
||
return 0.0
|
||
|
||
def __gt__(self, other):
|
||
return False
|
||
|
||
def __ge__(self, other):
|
||
return False
|
||
|
||
def __lt__(self, other):
|
||
return True
|
||
|
||
def __le__(self, other):
|
||
return True
|
||
|
||
class DemoField(DemoValue):
|
||
@property
|
||
def label(self):
|
||
return DemoField(self or 'Champ')
|
||
|
||
@property
|
||
def errors(self):
|
||
return []
|
||
|
||
def __getattr__(self, name):
|
||
return DemoField('')
|
||
|
||
class DemoObject(SimpleNamespace):
|
||
def __getattr__(self, name):
|
||
return DemoValue('')
|
||
|
||
class DemoForm(DemoObject):
|
||
def __getattr__(self, name):
|
||
return DemoField(name.replace('_', ' ').capitalize())
|
||
|
||
class DemoCollection(list):
|
||
def all(self):
|
||
return self
|
||
|
||
def count(self):
|
||
return len(self)
|
||
|
||
@property
|
||
def pages(self):
|
||
return 1
|
||
|
||
@property
|
||
def page(self):
|
||
return 1
|
||
|
||
@property
|
||
def total(self):
|
||
return len(self)
|
||
|
||
@property
|
||
def has_next(self):
|
||
return False
|
||
|
||
@property
|
||
def has_prev(self):
|
||
return False
|
||
|
||
def iter_pages(self, *args, **kwargs):
|
||
return [1]
|
||
|
||
class DemoMapping(dict):
|
||
def get(self, key, default=None):
|
||
return super().get(key, default)
|
||
|
||
from datetime import date, datetime
|
||
demo_date = date(2026, 8, 21)
|
||
demo_datetime = datetime(2026, 8, 21, 8, 15)
|
||
demo_room = DemoObject(id=12, name='B102', code='B102', full_name='Bâtiment principal — Zone Anglais — B102', equipments=DemoCollection(), rooms=DemoCollection())
|
||
demo_user = DemoObject(id=1, username='demo.admin', full_name='Administrateur démo', role='admin', role_label='Administrateur système', is_authenticated=True, is_admin=lambda: True)
|
||
demo_equipment = DemoObject(
|
||
id=42, name='Luminaire LED central', code='ECL-0042',
|
||
full_path='Bâtiment principal / Zone Anglais / B102 / Éclairage',
|
||
room=demo_room, status='en_service', status_label='En service',
|
||
quantity=1, children=DemoCollection(), documents=DemoCollection(),
|
||
)
|
||
demo_room.equipments.append(demo_equipment)
|
||
demo_meter = DemoObject(id=5, name='Compteur pages', unit='pages', equipment=demo_equipment, equipment_id=demo_equipment.id)
|
||
demo_child = DemoObject(id=43, individual_number=1, code='CH-0043', position='Rangée centrale', status='en_service', has_active_curative=False, interventions=DemoCollection())
|
||
demo_staff = DemoObject(id=8, full_name='Alex Martin', first_name='Alex', last_name='Martin', function='Agent polyvalent', department='Maintenance', email='alex.martin@example.fr', phone='01 60 00 00 00', is_active=True, intervention_count=12, open_intervention_count=2)
|
||
demo_company = DemoObject(id=2, name='Entreprise Démo Maintenance', contact_name='Camille Démo', contact_email='contact@example.fr', contact_phone='01 60 00 00 01', address='1 rue des Ateliers')
|
||
demo_lot = DemoObject(id=3, name='Éclairage intérieur', description='Maintenance des luminaires', category=DemoObject(name='Éclairage'), tasks=DemoCollection(), categories=DemoCollection())
|
||
demo_part = DemoObject(id=4, name='Cartouche noire', reference='DEMO-001', quantity=12, stock_quantity=12, minimum_stock=2)
|
||
demo_training = DemoObject(id=6, title='Habilitation électrique', name='Habilitation électrique', start_date=demo_date, end_date=demo_date, status='planifiee')
|
||
demo_building = DemoObject(id=1, name='Bâtiment principal', rooms=DemoCollection([demo_room]))
|
||
demo_category = DemoObject(id=1, name='Éclairage', equipments=DemoCollection([demo_equipment]))
|
||
demo_mail = DemoObject(id='demo-mail', account_id=1, subject='Maintenance urgente', sender='service@exemple.fr', sender_email='service@exemple.fr', received_at=demo_datetime, is_read=False, importance='normal', has_attachments=False, attachment_count=0, body_content='Message de démonstration.', body_preview='Demande de maintenance fictive.', message_id='demo-mail')
|
||
preview_context.update({
|
||
'current_user': demo_user,
|
||
'stats': DemoObject(alerts_unread=3, interventions_pending=4, equipments=128),
|
||
'unread': 3,
|
||
'alerts': [DemoObject(id=1, title='Luminaire défaillant', message='Une intervention est à traiter.', level='warning')],
|
||
'interventions': DemoCollection([DemoObject(id=39, title='Remplacement du luminaire', status='en_cours', status_label='En cours', room=demo_room, equipment=demo_equipment)]),
|
||
'intervention': DemoObject(id=39, title='Remplacement du luminaire', type='curatif', status='en_cours', status_label='En cours', room=demo_room, equipment=demo_equipment),
|
||
'rooms': DemoCollection([demo_room, DemoObject(id=13, name='B103', code='B103', full_name='Bâtiment principal — Zone Anglais — B103', equipments=DemoCollection())]),
|
||
'equipments': DemoCollection([demo_equipment, DemoObject(id=43, name='Prise murale fond de salle', code='ELEC-0043', room=demo_room)]),
|
||
'categories': DemoCollection([demo_category, DemoObject(id=2, name='Prises', equipments=DemoCollection())]),
|
||
'buildings': DemoCollection([demo_building]),
|
||
'zones': DemoCollection([DemoObject(id=1, name='Zone Anglais')]),
|
||
'messages': DemoCollection([DemoObject(id=1, subject='Demande de réparation', sender='Vie scolaire', sender_name='Vie scolaire', body_preview='Une chaise est cassée en B102.', is_unread=True, date=None)]),
|
||
'mails': DemoCollection([demo_mail]),
|
||
'mail': demo_mail,
|
||
'account': DemoObject(id=1, email='demo@example.fr'),
|
||
'tasks': DemoCollection([DemoObject(id=7, name='Contrôle mensuel des luminaires', periodicity='Mensuelle', duration_minutes=30)]),
|
||
'lots': DemoCollection([demo_lot]),
|
||
'companies': DemoCollection([demo_company]),
|
||
'form': DemoForm(),
|
||
'manual_form': DemoForm(),
|
||
'statuses': {'en_service': 'En service', 'en_panne': 'En panne', 'a_jeter': 'À jeter'},
|
||
'equipment': demo_equipment,
|
||
'meter': demo_meter,
|
||
'readings': DemoCollection([DemoObject(date=None, value=1250, comment='Relevé de démonstration', user=demo_user)]),
|
||
'parts': DemoCollection([demo_part]),
|
||
'group': DemoObject(id=50, name='Chaises salle B102', full_path=demo_room.full_name, children_count=10, total_children=12, tracked_individually=True),
|
||
'children': DemoCollection([demo_child]),
|
||
'staff': demo_staff,
|
||
'company': demo_company,
|
||
'lot': demo_lot,
|
||
'part': demo_part,
|
||
'training': demo_training,
|
||
'trainings': DemoCollection([demo_training]),
|
||
'status_counts': {'brouillon': 1, 'en_cours': 2, 'terminee': 4},
|
||
'priorities': {'basse': 'Basse', 'normale': 'Normale', 'haute': 'Haute'},
|
||
'trashed_count': 1,
|
||
'workflow_types': {'corrective': 'Correctif', 'preventive': 'Préventif', 'travaux': 'Travaux'},
|
||
'week_start': demo_date,
|
||
'week_end': demo_date,
|
||
'prev_week': demo_date,
|
||
'next_week': demo_date,
|
||
'today': demo_date,
|
||
'prev_period': demo_date,
|
||
'next_period': demo_date,
|
||
'days': {0: 'Lundi', 1: 'Mardi', 2: 'Mercredi', 3: 'Jeudi', 4: 'Vendredi'},
|
||
'day_date': demo_date,
|
||
'schedules': DemoMapping(),
|
||
'participant_users': DemoMapping({demo_user.id: True}),
|
||
})
|
||
from flask import url_for as flask_url_for
|
||
def preview_url_for(endpoint, *args, **kwargs):
|
||
try:
|
||
return flask_url_for(endpoint, *args, **kwargs)
|
||
except Exception:
|
||
return '#preview'
|
||
preview_context['url_for'] = preview_url_for
|
||
rendered = template.render(**preview_context)
|
||
except (TemplateNotFound, Exception) as exc:
|
||
render_error = f'{type(exc).__name__}: {exc}'
|
||
|
||
return render_template('admin/template_preview.html', template=item,
|
||
source=source, rendered=rendered, render_error=render_error,
|
||
demo_mode=demo_mode)
|
||
|
||
|
||
@admin_bp.route('/debug/templates/mark', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def template_debug_mark():
|
||
"""Marque un template comme obsolète, ou annule ce marquage."""
|
||
from flask import current_app
|
||
from ..models.audit import TemplateAuditMark
|
||
|
||
relative_path = (request.form.get('path') or '').strip()
|
||
action = request.form.get('action')
|
||
notes = (request.form.get('notes') or '').strip() or None
|
||
app_root = Path(current_app.root_path).resolve()
|
||
candidate = (app_root / relative_path).resolve()
|
||
try:
|
||
candidate.relative_to(app_root)
|
||
except ValueError:
|
||
candidate = None
|
||
if not candidate or not candidate.is_file() or candidate.suffix != '.html':
|
||
flash('Chemin de template invalide.', 'danger')
|
||
return redirect(url_for('admin.template_debug'))
|
||
|
||
mark = TemplateAuditMark.query.filter_by(path=relative_path).first()
|
||
if action == 'obsolete':
|
||
if not mark:
|
||
mark = TemplateAuditMark(path=relative_path)
|
||
db.session.add(mark)
|
||
mark.status = 'obsolete'
|
||
mark.notes = notes
|
||
mark.updated_by_id = current_user.id
|
||
flash(f'Template marqué obsolète : {relative_path}', 'warning')
|
||
elif action == 'restore':
|
||
if mark:
|
||
db.session.delete(mark)
|
||
flash(f'Marquage retiré : {relative_path}', 'success')
|
||
else:
|
||
flash('Action inconnue.', 'danger')
|
||
return redirect(url_for('admin.template_debug'))
|
||
db.session.commit()
|
||
return redirect(url_for('admin.template_debug', filter=request.form.get('return_filter', 'duplicates')))
|
||
|
||
|
||
# ─── Gestion des données (cleanup) ─────────────────────────────────────
|
||
|
||
@admin_bp.route('/data')
|
||
@login_required
|
||
@admin_required
|
||
def data_management():
|
||
"""Page de gestion des données : voir et supprimer en masse."""
|
||
from ..models.maintenance import Intervention
|
||
from ..models.equipment import Equipment
|
||
from ...outlook.models import OutlookMail, OutlookMailInterpretation, OutlookAttachment
|
||
from ...ent.models import EntMessage
|
||
from ...ent.interpretation_models import EntMessageInterpretation
|
||
|
||
stats = {
|
||
'equipments': Equipment.query.count(),
|
||
'interventions': Intervention.query.count(),
|
||
'interventions_planifiees': Intervention.query.filter_by(status='planifiee', is_deleted=False).count(),
|
||
'interventions_terminees': Intervention.query.filter_by(status='terminee').count(),
|
||
'outlook_mails': OutlookMail.query.count(),
|
||
'outlook_interpretations': OutlookMailInterpretation.query.count(),
|
||
'ent_messages': EntMessage.query.count(),
|
||
'ent_interpretations': EntMessageInterpretation.query.count(),
|
||
}
|
||
|
||
# Recuperer les mails Outlook (pagine)
|
||
page = request.args.get('page', 1, type=int)
|
||
page_eq = request.args.get('equipments_page', 1, type=int)
|
||
page_int = request.args.get('interventions_page', 1, type=int)
|
||
missing_only = request.args.get('non_interpretes') == '1'
|
||
if missing_only:
|
||
interpreted_mail_ids = db.session.query(OutlookMailInterpretation.mail_id).distinct()
|
||
mails_query = OutlookMail.query.filter(~OutlookMail.id.in_(interpreted_mail_ids))
|
||
else:
|
||
mails_query = OutlookMail.query
|
||
# Une pagination commune est utilisée par les deux listes. Ne pas
|
||
# renvoyer 404 si la liste ENT contient moins de pages que les mails.
|
||
mails = mails_query.order_by(OutlookMail.received_at.desc()).paginate(page=page, per_page=25, error_out=False)
|
||
|
||
# Interventions recentes (pagine, non supprimees)
|
||
interventions = Intervention.query.filter_by(is_deleted=False).order_by(Intervention.created_at.desc()).paginate(page=page_int, per_page=25)
|
||
|
||
# Équipements (pagine, non supprimes)
|
||
equipments = Equipment.query.filter_by(is_deleted=False).order_by(Equipment.name).paginate(page=page_eq, per_page=25)
|
||
|
||
# Pour chaque mail Outlook, verifier s'il a une interpretation
|
||
outlook_interp_map = {}
|
||
for interp in OutlookMailInterpretation.query.all():
|
||
outlook_interp_map[interp.mail_id] = interp
|
||
|
||
# Recuperer les messages ENT (pagine)
|
||
if missing_only:
|
||
interpreted_ent_ids = db.session.query(EntMessageInterpretation.message_id).distinct()
|
||
ent_query = EntMessage.query.filter(~EntMessage.id.in_(interpreted_ent_ids))
|
||
else:
|
||
ent_query = EntMessage.query
|
||
ent_messages = ent_query.order_by(EntMessage.date.desc()).paginate(page=page, per_page=25, error_out=False)
|
||
|
||
# Pour chaque message ENT, verifier s'il a une interpretation
|
||
ent_interp_map = {}
|
||
for interp in EntMessageInterpretation.query.all():
|
||
ent_interp_map[interp.message_id] = interp
|
||
|
||
# Recuperer le contexte GMAO pour les expediteurs exclus
|
||
from ...outlook.models import GmaoContext
|
||
context = GmaoContext.get_for_user(current_user.id)
|
||
excluded_list = context.get_excluded_senders_list() if context else []
|
||
|
||
return render_template('admin/data_management.html',
|
||
stats=stats,
|
||
mails=mails,
|
||
interventions=interventions,
|
||
equipments=equipments,
|
||
outlook_interp_map=outlook_interp_map,
|
||
ent_messages=ent_messages,
|
||
ent_interp_map=ent_interp_map,
|
||
missing_only=missing_only,
|
||
excluded_list=excluded_list)
|
||
|
||
|
||
@admin_bp.route('/data/delete-all/equipments', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def delete_all_equipments():
|
||
"""Supprime tous les equipements."""
|
||
from ..models.equipment import Equipment
|
||
confirm = request.form.get('confirm_text', '')
|
||
if confirm != 'SUPPRIMER':
|
||
flash('Confirmation requise : tapez "SUPPRIMER" dans le champ de confirmation.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
count = Equipment.query.count()
|
||
# Suppression logique pour eviter les cascades
|
||
from sqlalchemy import update
|
||
db.session.execute(update(Equipment).values(is_deleted=True))
|
||
db.session.commit()
|
||
flash(f'{count} equipement(s) supprime(s) (suppression logique).', 'warning')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
|
||
@admin_bp.route('/data/delete-all/interventions', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def delete_all_interventions():
|
||
"""Supprime toutes les interventions."""
|
||
from ..models.maintenance import Intervention
|
||
confirm = request.form.get('confirm_text', '')
|
||
if confirm != 'SUPPRIMER':
|
||
flash('Confirmation requise : tapez "SUPPRIMER" dans le champ de confirmation.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
count = Intervention.query.count()
|
||
from sqlalchemy import update
|
||
db.session.execute(update(Intervention).values(is_deleted=True, deleted_at=db.func.now()))
|
||
db.session.commit()
|
||
flash(f'{count} intervention(s) supprimee(s) (suppression logique).', 'warning')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
|
||
@admin_bp.route('/data/delete-all/mails', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def delete_all_mails():
|
||
"""Supprime tous les mails Outlook."""
|
||
from ...outlook.models import OutlookMail, OutlookMailInterpretation, OutlookAttachment
|
||
confirm = request.form.get('confirm_text', '')
|
||
if confirm != 'SUPPRIMER':
|
||
flash('Confirmation requise : tapez "SUPPRIMER" dans le champ de confirmation.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
try:
|
||
# Les pièces jointes possèdent une FK vers les mails : elles doivent
|
||
# être supprimées avant leurs mails (le modèle seul ne garantit pas
|
||
# une cascade côté MariaDB).
|
||
OutlookAttachment.query.delete(synchronize_session=False)
|
||
OutlookMailInterpretation.query.delete(synchronize_session=False)
|
||
count = OutlookMail.query.count()
|
||
OutlookMail.query.delete(synchronize_session=False)
|
||
db.session.commit()
|
||
except Exception:
|
||
db.session.rollback()
|
||
raise
|
||
flash(f'{count} mail(s) supprime(s) definitivement.', 'warning')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
|
||
@admin_bp.route('/data/delete-mail/<mail_id>', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def delete_single_mail(mail_id):
|
||
"""Supprime un seul mail Outlook."""
|
||
from ...outlook.models import OutlookMail, OutlookMailInterpretation
|
||
mail = OutlookMail.query.get(mail_id)
|
||
if mail:
|
||
# Supprimer les interpretations associees
|
||
OutlookMailInterpretation.query.filter_by(mail_id=mail_id).delete()
|
||
db.session.delete(mail)
|
||
db.session.commit()
|
||
flash(f'Mail supprime.', 'info')
|
||
else:
|
||
flash('Mail introuvable.', 'danger')
|
||
return redirect(url_for('admin.data_management', page=request.args.get('page', 1)))
|
||
|
||
|
||
@admin_bp.route('/data/delete-all/ent-messages', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def delete_all_ent_messages():
|
||
"""Supprime tous les messages ENT et leurs interpretations."""
|
||
from ...ent.models import EntMessage
|
||
from ...ent.interpretation_models import EntMessageInterpretation
|
||
confirm = request.form.get('confirm_text', '')
|
||
if confirm != 'SUPPRIMER':
|
||
flash('Confirmation requise : tapez "SUPPRIMER" dans le champ de confirmation.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
# Supprimer d'abord les interpretations
|
||
EntMessageInterpretation.query.delete()
|
||
# Supprimer les messages
|
||
count = EntMessage.query.count()
|
||
EntMessage.query.delete()
|
||
db.session.commit()
|
||
flash(f'{count} message(s) ENT supprime(s) definitivement.', 'warning')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
|
||
@admin_bp.route('/data/delete-ent-message/<msg_id>', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def delete_single_ent_message(msg_id):
|
||
"""Supprime un seul message ENT."""
|
||
from ...ent.models import EntMessage
|
||
from ...ent.interpretation_models import EntMessageInterpretation
|
||
msg = EntMessage.query.get(msg_id)
|
||
if msg:
|
||
EntMessageInterpretation.query.filter_by(message_id=msg_id).delete()
|
||
db.session.delete(msg)
|
||
db.session.commit()
|
||
flash('Message ENT supprime.', 'info')
|
||
else:
|
||
flash('Message introuvable.', 'danger')
|
||
return redirect(url_for('admin.data_management', page=request.args.get('page', 1)))
|
||
|
||
|
||
@admin_bp.route('/data/ignore-sender', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def ignore_sender():
|
||
"""Ajoute un expediteur a la liste des expediteurs exclus du GmaoContext."""
|
||
from ...outlook.models import GmaoContext
|
||
sender = request.form.get('sender', '').strip()
|
||
if not sender:
|
||
flash('Expediteur manquant.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
context = GmaoContext.get_for_user(current_user.id)
|
||
current_list = context.excluded_senders or ''
|
||
# Verifier s'il est deja present
|
||
existing = [e.strip().lower() for e in current_list.split('\n') if e.strip()]
|
||
if sender.lower() not in existing:
|
||
if current_list and not current_list.endswith('\n'):
|
||
current_list += '\n'
|
||
current_list += sender + '\n'
|
||
context.excluded_senders = current_list
|
||
db.session.commit()
|
||
flash(f'Expediteur "{sender}" ajoute a la liste d\'exclusion.', 'success')
|
||
else:
|
||
flash(f'Expediteur "{sender}" deja exclu.', 'info')
|
||
return redirect(url_for('admin.data_management', page=request.args.get('page', 1)))
|
||
|
||
|
||
# ─── Suppressions individuelles ────────────────────────────────────────
|
||
|
||
@admin_bp.route('/data/delete-intervention/<int:interv_id>', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def delete_single_intervention(interv_id):
|
||
"""Supprime une intervention."""
|
||
from ..models.maintenance import Intervention
|
||
interv = Intervention.query.get(interv_id)
|
||
if interv:
|
||
interv.is_deleted = True
|
||
interv.deleted_at = db.func.now()
|
||
db.session.commit()
|
||
flash(f'Intervention #{interv_id} supprimée (suppression logique).', 'info')
|
||
else:
|
||
flash('Intervention introuvable.', 'danger')
|
||
return redirect(url_for('admin.data_management', page=request.args.get('page', 1, type=int)))
|
||
|
||
|
||
@admin_bp.route('/data/delete-equipment/<int:equip_id>', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def delete_single_equipment(equip_id):
|
||
"""Supprime un équipement."""
|
||
from ..models.equipment import Equipment
|
||
equip = Equipment.query.get(equip_id)
|
||
if equip:
|
||
equip.is_deleted = True
|
||
db.session.commit()
|
||
flash(f'Équipement "{equip.name}" supprimé (suppression logique).', 'info')
|
||
else:
|
||
flash('Équipement introuvable.', 'danger')
|
||
return redirect(url_for('admin.data_management', page=request.args.get('page', 1, type=int)))
|
||
|
||
|
||
@admin_bp.route('/data/interpret-outlook/<mail_id>', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def interpret_outlook_mail(mail_id):
|
||
"""Lance l'interpretation manuelle d'un mail Outlook."""
|
||
from ...outlook.models import OutlookMail, GmaoContext
|
||
from ...outlook.auto_interpret import analyze_email_with_hermes
|
||
from ...core.models.settings import AppSettings
|
||
|
||
mail = OutlookMail.query.get(mail_id)
|
||
if not mail:
|
||
flash('Mail introuvable.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
context = GmaoContext.get_for_user(current_user.id)
|
||
if not context:
|
||
flash('Contexte GMAO non configure.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
result = analyze_email_with_hermes(mail, context)
|
||
if result.get('error'):
|
||
flash(f'Erreur interpretation: {result["error"]}', 'danger')
|
||
else:
|
||
flash('Mail interprete avec succes.', 'success')
|
||
return redirect(url_for('admin.data_management', page=request.args.get('page', 1)))
|
||
|
||
|
||
@admin_bp.route('/data/interpret-ent/<msg_id>', methods=['POST'])
|
||
@login_required
|
||
@admin_required
|
||
def interpret_ent_message(msg_id):
|
||
"""Lance l'interpretation manuelle d'un message ENT."""
|
||
from ...ent.models import EntMessage
|
||
from ...ent.interpretation_models import EntMessageInterpretation
|
||
from ...outlook.models import GmaoContext
|
||
from ...core.models.settings import AppSettings
|
||
import requests as http_requests
|
||
|
||
message = EntMessage.query.get(msg_id)
|
||
if not message:
|
||
flash('Message introuvable.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
context = GmaoContext.get_for_user(current_user.id)
|
||
if not context:
|
||
flash('Contexte GMAO non configure.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
api_key = AppSettings.get('openrouter_api_key')
|
||
if not api_key:
|
||
flash('Cle API OpenRouter non configuree.', 'danger')
|
||
return redirect(url_for('admin.data_management'))
|
||
|
||
# Construire le prompt
|
||
prompt = f"""Analyse ce message de la plateforme ENT77 et propose une action GMAO.
|
||
|
||
Contexte de l'etablissement:
|
||
{context.establishment_name or 'Non defini'}
|
||
{context.technical_context or ''}
|
||
{context.equipment_context or ''}
|
||
|
||
Message:
|
||
Sujet: {message.subject}
|
||
Expediteur: {message.sender_name}
|
||
Contenu: {message.body[:1000] if message.body else ''}
|
||
|
||
Reponds en JSON avec:
|
||
- type: curative/preventive/administrative/formation/information
|
||
- action: create/update/inform
|
||
- priority: low/medium/high/critical
|
||
- summary: resume court
|
||
- location: salle ou zone concernnee
|
||
- equipment: equipement concerne
|
||
"""
|
||
|
||
try:
|
||
model = AppSettings.get('openrouter_model', 'poolside/laguna-s-2.1:free')
|
||
response = http_requests.post(
|
||
'https://openrouter.ai/api/v1/chat/completions',
|
||
headers={'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'},
|
||
json={'model': model, 'messages': [{'role': 'user', 'content': prompt}], 'max_tokens': 1000},
|
||
timeout=60
|
||
)
|
||
if response.status_code == 200:
|
||
content = response.json().get('choices', [{}])[0].get('message', {}).get('content', '')
|
||
# Parser le JSON
|
||
import json
|
||
if '```json' in content:
|
||
content = content.split('```json')[1].split('```')[0]
|
||
elif '```' in content:
|
||
content = content.split('```')[1].split('```')[0]
|
||
content = content.strip()
|
||
json_start = content.find('{')
|
||
if json_start >= 0:
|
||
content = content[json_start:]
|
||
json_end = content.rfind('}')
|
||
if json_end >= 0:
|
||
content = content[:json_end+1]
|
||
|
||
analysis = json.loads(content) if content else {}
|
||
|
||
# Le modèle peut renvoyer plusieurs équipements sous forme de
|
||
# tableau alors que la colonne est volontairement textuelle.
|
||
equipment = analysis.get('equipment', '')
|
||
if isinstance(equipment, (list, tuple)):
|
||
equipment = ', '.join(str(item) for item in equipment if item)
|
||
elif equipment is None:
|
||
equipment = ''
|
||
|
||
# Supprimer l'ancienne interpretation si elle existe
|
||
EntMessageInterpretation.query.filter_by(message_id=msg_id).delete()
|
||
|
||
# Creer l'interpretation
|
||
interp = EntMessageInterpretation(
|
||
message_id=msg_id,
|
||
analysis_type=analysis.get('type', 'information'),
|
||
analysis_action=analysis.get('action', 'inform'),
|
||
analysis_priority=analysis.get('priority', 'medium'),
|
||
analysis_summary=analysis.get('summary', ''),
|
||
suggested_equipment=str(equipment),
|
||
suggested_description=analysis.get('location', ''),
|
||
status='pending',
|
||
)
|
||
db.session.add(interp)
|
||
db.session.commit()
|
||
flash('Message ENT interprete avec succes.', 'success')
|
||
else:
|
||
flash(f'Erreur API: {response.status_code}', 'danger')
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
flash(f'Erreur interpretation: {str(e)[:200]}', 'danger')
|
||
|
||
return redirect(url_for('admin.data_management', page=request.args.get('page', 1)))
|