677 lines
25 KiB
Python
677 lines
25 KiB
Python
"""
|
||
Core Routes - Administration
|
||
GMAO Collège
|
||
"""
|
||
from flask import Blueprint, render_template, redirect, url_for, request, flash
|
||
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 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)
|
||
|
||
|
||
@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')
|
||
|
||
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 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.commit()
|
||
flash('Utilisateur créé avec succès.', 'success')
|
||
return redirect(url_for('admin.users'))
|
||
|
||
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')
|
||
if role not in 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)
|
||
|
||
db.session.commit()
|
||
flash('Utilisateur modifié avec succès.', 'success')
|
||
return redirect(url_for('admin.users'))
|
||
|
||
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('/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():
|
||
"""Créer un type de salle."""
|
||
if request.method == 'POST':
|
||
from ..models.college import RoomType
|
||
room_type = RoomType(
|
||
name=request.form.get('name'),
|
||
description=request.form.get('description'),
|
||
color=request.form.get('color', '#6c757d'),
|
||
icon=request.form.get('icon')
|
||
)
|
||
db.session.add(room_type)
|
||
db.session.commit()
|
||
flash('Type de salle créé.', 'success')
|
||
return redirect(url_for('admin.room_types'))
|
||
|
||
return render_template('admin/room_type_form.html')
|
||
|
||
|
||
@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('/setup-wizard')
|
||
@login_required
|
||
@admin_required
|
||
def setup_wizard_redirect():
|
||
"""Redirection vers le setup wizard."""
|
||
return redirect(url_for('setup_wizard.index'))
|
||
|
||
|
||
# ─── 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
|
||
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)
|
||
mails = OutlookMail.query.order_by(OutlookMail.received_at.desc()).paginate(page=page, per_page=25)
|
||
|
||
# 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)
|
||
ent_messages = EntMessage.query.order_by(EntMessage.date.desc()).paginate(page=page, per_page=25)
|
||
|
||
# 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,
|
||
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
|
||
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
|
||
OutlookMailInterpretation.query.delete()
|
||
# Supprimer les mails
|
||
count = OutlookMail.query.count()
|
||
OutlookMail.query.delete()
|
||
db.session.commit()
|
||
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 {}
|
||
|
||
# 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=analysis.get('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:
|
||
flash(f'Erreur interpretation: {str(e)[:200]}', 'danger')
|
||
|
||
return redirect(url_for('admin.data_management', page=request.args.get('page', 1)))
|