230 lines
11 KiB
Python
230 lines
11 KiB
Python
"""Routes C2 : règles, échéances et tournées de relevés."""
|
||
|
||
from datetime import date, datetime
|
||
from pathlib import Path
|
||
from uuid import uuid4
|
||
|
||
from flask import abort, current_app, flash, jsonify, redirect, render_template, request, url_for
|
||
from flask_login import current_user, login_required
|
||
from werkzeug.utils import secure_filename
|
||
|
||
from ..extensions import db
|
||
from ..core.models.planning import (
|
||
Meter, MeterReadingOccurrence, MeterReadingRound, MeterReadingSchedule,
|
||
)
|
||
from ..core.models.user import User
|
||
from ..core.services.meter_reading_planning import (
|
||
FREQUENCIES, NO_READING_REASONS, MeterDomainError, add_round_member, create_or_update_schedule,
|
||
create_round, ensure_occurrences_for_operational_date, generate_occurrences,
|
||
record_occurrence_reading, record_occurrence_without_reading,
|
||
)
|
||
from ..core.services.meter_analytics import consumption_intervals
|
||
from ..core.models.planning import MeterAlert, MeterAlertRule
|
||
from .schedules import planning_bp
|
||
|
||
|
||
def _admin_required():
|
||
if not current_user.is_admin():
|
||
abort(403)
|
||
|
||
|
||
@planning_bp.route('/meter-schedules')
|
||
@login_required
|
||
def meter_schedules():
|
||
_admin_required()
|
||
schedules = MeterReadingSchedule.query.order_by(MeterReadingSchedule.id.desc()).all()
|
||
return render_template('planning/meter_schedules.html', schedules=schedules)
|
||
|
||
|
||
@planning_bp.route('/meter-schedules/new', methods=['GET', 'POST'])
|
||
@login_required
|
||
def new_meter_schedule():
|
||
_admin_required()
|
||
meters = Meter.query.filter(Meter.status != 'replaced', Meter.is_active.is_(True)).order_by(Meter.name).all()
|
||
users = User.query.filter_by(is_active=True).order_by(User.full_name, User.username).all()
|
||
if request.method == 'POST':
|
||
meter = db.session.get(Meter, request.form.get('meter_id', type=int))
|
||
if not meter:
|
||
flash('Compteur invalide.', 'danger')
|
||
else:
|
||
try:
|
||
reference_date = date.fromisoformat(request.form.get('reference_date'))
|
||
target_time = datetime.strptime(request.form['target_time'], '%H:%M').time() if request.form.get('target_time') else None
|
||
create_or_update_schedule(
|
||
meter=meter, frequency=request.form.get('frequency'), reference_date=reference_date,
|
||
target_time=target_time, duration_minutes=request.form.get('duration_minutes', type=int, default=5),
|
||
assigned_to_id=request.form.get('assigned_to_id', type=int),
|
||
calendar_scope=request.form.get('calendar_scope') or None,
|
||
fixed_month=request.form.get('fixed_month', type=int), fixed_day=request.form.get('fixed_day', type=int),
|
||
)
|
||
flash('Règle de relevé enregistrée.', 'success')
|
||
return redirect(url_for('planning.meter_schedules'))
|
||
except (ValueError, TypeError) as exc:
|
||
flash(str(exc) or 'Date ou heure invalide.', 'danger')
|
||
return render_template('planning/meter_schedule_form.html', meters=meters, users=users, frequencies=sorted(FREQUENCIES))
|
||
|
||
|
||
@planning_bp.route('/meter-occurrences/<int:occurrence_id>')
|
||
@login_required
|
||
def meter_occurrence_detail(occurrence_id):
|
||
occurrence = MeterReadingOccurrence.query.get_or_404(occurrence_id)
|
||
if occurrence.assigned_to_id not in (None, current_user.id) and not current_user.is_admin():
|
||
abort(403)
|
||
return render_template(
|
||
'planning/meter_occurrence.html', occurrence=occurrence,
|
||
no_reading_reasons=NO_READING_REASONS,
|
||
)
|
||
|
||
|
||
def _optional_photo():
|
||
photo = request.files.get('photo')
|
||
if not photo or not photo.filename:
|
||
return None, None
|
||
allowed = {'jpg', 'jpeg', 'png', 'gif'}
|
||
if '.' not in photo.filename or photo.filename.rsplit('.', 1)[1].lower() not in allowed:
|
||
raise ValueError('La photo doit être une image JPG, PNG ou GIF.')
|
||
safe_name = secure_filename(photo.filename)
|
||
upload_dir = Path(current_app.config.get('UPLOAD_FOLDER', 'uploads')) / 'meter-readings'
|
||
upload_dir.mkdir(parents=True, exist_ok=True)
|
||
path = upload_dir / f'{uuid4().hex}_{safe_name}'
|
||
photo.save(path)
|
||
return path.name, str(path)
|
||
|
||
|
||
@planning_bp.route('/meter-occurrences/<int:occurrence_id>/reading', methods=['POST'])
|
||
@login_required
|
||
def meter_occurrence_reading(occurrence_id):
|
||
photo_path = None
|
||
try:
|
||
photo_filename, photo_path = _optional_photo()
|
||
record_occurrence_reading(
|
||
occurrence_id=occurrence_id, value=request.form.get('value'), user_id=current_user.id,
|
||
notes=request.form.get('notes'), is_reset=request.form.get('is_reset') == '1',
|
||
reset_reason=request.form.get('reset_reason'), photo_filename=photo_filename, photo_path=photo_path,
|
||
)
|
||
flash('Relevé enregistré et échéance traitée.', 'success')
|
||
except (ValueError, MeterDomainError) as exc:
|
||
db.session.rollback()
|
||
if photo_path:
|
||
Path(photo_path).unlink(missing_ok=True)
|
||
flash(str(exc), 'danger')
|
||
return redirect(url_for('planning.meter_occurrence_detail', occurrence_id=occurrence_id))
|
||
|
||
|
||
@planning_bp.route('/meter-occurrences/<int:occurrence_id>/without-reading', methods=['POST'])
|
||
@login_required
|
||
def meter_occurrence_without_reading(occurrence_id):
|
||
try:
|
||
record_occurrence_without_reading(
|
||
occurrence_id=occurrence_id, reason=request.form.get('reason'),
|
||
user_id=current_user.id, comment=request.form.get('comment'),
|
||
)
|
||
flash('Échéance traitée sans relevé.', 'success')
|
||
except MeterDomainError as exc:
|
||
db.session.rollback()
|
||
flash(str(exc), 'danger')
|
||
return redirect(url_for('planning.meter_occurrence_detail', occurrence_id=occurrence_id))
|
||
|
||
|
||
@planning_bp.route('/meter-monitoring')
|
||
@login_required
|
||
def meter_monitoring():
|
||
"""Tableau de surveillance C3, sans recalculer ni modifier les relevés."""
|
||
from sqlalchemy.orm import selectinload
|
||
meters = Meter.query.filter(Meter.is_active.is_(True)).options(selectinload(Meter.alerts)).order_by(Meter.name).all()
|
||
alerts = MeterAlert.query.filter(MeterAlert.status.in_(("OPEN", "INVESTIGATING"))).order_by(MeterAlert.level.desc(), MeterAlert.opened_at.desc()).all()
|
||
return render_template('planning/meter_monitoring.html', meters=meters, alerts=alerts)
|
||
|
||
|
||
@planning_bp.route('/meter-analytics/<int:meter_id>')
|
||
@login_required
|
||
def meter_analytics(meter_id):
|
||
meter = Meter.query.get_or_404(meter_id)
|
||
intervals = consumption_intervals(meter)
|
||
return render_template('planning/meter_analytics.html', meter=meter, intervals=intervals)
|
||
|
||
|
||
@planning_bp.route('/meter-alerts/<int:alert_id>/create-intervention', methods=['POST'])
|
||
@login_required
|
||
def create_intervention_from_meter_alert(alert_id):
|
||
from ..core.models.maintenance import Intervention
|
||
alert = MeterAlert.query.get_or_404(alert_id)
|
||
if alert.intervention_id:
|
||
return redirect(url_for('interventions.detail', id=alert.intervention_id))
|
||
meter = alert.meter
|
||
interval_text = f"Valeur observée : {alert.observed_value}; référence : {alert.reference_value}" if alert.reference_value is not None else f"Valeur observée : {alert.observed_value}"
|
||
intervention = Intervention(
|
||
title=request.form.get('title') or f"Analyser l'alerte compteur — {meter.name}",
|
||
description=(request.form.get('description') or f"Alerte {alert.alert_type} ({alert.level}). {interval_text}"),
|
||
type='curatif', workflow_type='corrective', status='brouillon', priority='haute' if alert.level == 'CRITICAL' else 'normale',
|
||
equipment_id=meter.equipment_id, room_id=meter.room_id, author_id=current_user.id,
|
||
)
|
||
db.session.add(intervention)
|
||
db.session.flush()
|
||
alert.intervention_id = intervention.id
|
||
db.session.commit()
|
||
flash('Intervention créée depuis l’alerte.', 'success')
|
||
return redirect(url_for('interventions.detail', id=intervention.id))
|
||
|
||
|
||
@planning_bp.route('/meter-rounds')
|
||
@login_required
|
||
def meter_rounds():
|
||
_admin_required()
|
||
rounds = MeterReadingRound.query.order_by(MeterReadingRound.name).all()
|
||
return render_template('planning/meter_rounds.html', rounds=rounds)
|
||
|
||
|
||
@planning_bp.route('/meter-rounds/new', methods=['GET', 'POST'])
|
||
@login_required
|
||
def new_meter_round():
|
||
_admin_required()
|
||
schedules = MeterReadingSchedule.query.filter_by(is_active=True).order_by(MeterReadingSchedule.id).all()
|
||
users = User.query.filter_by(is_active=True).order_by(User.full_name, User.username).all()
|
||
if request.method == 'POST':
|
||
try:
|
||
round_ = create_round(
|
||
name=request.form.get('name'), default_assigned_to_id=request.form.get('assigned_to_id', type=int),
|
||
estimated_duration_minutes=request.form.get('estimated_duration_minutes', type=int),
|
||
commit=False,
|
||
)
|
||
selected = request.form.getlist('schedule_ids')
|
||
for position, schedule_id in enumerate(selected, 1):
|
||
schedule = db.session.get(MeterReadingSchedule, int(schedule_id))
|
||
if schedule:
|
||
add_round_member(round_=round_, schedule=schedule, position=position, commit=False)
|
||
db.session.commit()
|
||
flash('Tournée enregistrée.', 'success')
|
||
return redirect(url_for('planning.meter_round_detail', round_id=round_.id))
|
||
except (ValueError, TypeError, MeterDomainError) as exc:
|
||
db.session.rollback()
|
||
flash(str(exc), 'danger')
|
||
return render_template('planning/meter_round_form.html', schedules=schedules, users=users)
|
||
|
||
|
||
@planning_bp.route('/meter-rounds/<int:round_id>')
|
||
@login_required
|
||
def meter_round_detail(round_id):
|
||
round_ = MeterReadingRound.query.get_or_404(round_id)
|
||
if not current_user.is_admin():
|
||
assigned = {member.schedule.assigned_to_id for member in round_.members}
|
||
if round_.default_assigned_to_id != current_user.id and current_user.id not in assigned:
|
||
abort(403)
|
||
target_date = request.args.get('date', date.today().isoformat())
|
||
try:
|
||
target_date = date.fromisoformat(target_date)
|
||
except ValueError:
|
||
target_date = date.today()
|
||
ensure_occurrences_for_operational_date(
|
||
target_date, round_id=round_.id,
|
||
schedule_ids=[member.schedule_id for member in round_.members],
|
||
)
|
||
round_occurrence = next((item for item in round_.occurrences if item.operational_date == target_date), None)
|
||
members = []
|
||
if round_occurrence:
|
||
members = sorted(round_occurrence.occurrences, key=lambda item: item.schedule.meter.name)
|
||
members.sort(key=lambda item: next((m.position for m in round_.members if m.schedule_id == item.schedule_id), 9999))
|
||
return render_template(
|
||
'planning/meter_round_detail.html', round=round_, round_occurrence=round_occurrence,
|
||
members=members, target_date=target_date,
|
||
)
|