"""Configuration des horaires et suivi annuel des heures de l'agent.""" from datetime import date, datetime, timedelta from flask import flash, redirect, render_template, request, url_for from flask_login import current_user, login_required from ..extensions import db from ..core.models.planning import ( AnnualTimeConfig, ClosureWorkDay, CollegeClosure, TimeEntry, WorkSchedule, WorkScheduleTemplate, ) from .schedules import planning_bp def _time(value): return datetime.strptime(value, "%H:%M").time() if value else None def _minutes(start, end, lunch_start=None, lunch_end=None): if not start or not end: return 0 value = (end.hour * 60 + end.minute) - (start.hour * 60 + start.minute) if lunch_start and lunch_end: value -= (lunch_end.hour * 60 + lunch_end.minute) - (lunch_start.hour * 60 + lunch_start.minute) return max(value, 0) def _redirect(year=None): return redirect(url_for("planning.time_tracking", year=year or date.today().year)) @planning_bp.route("/time") @login_required def time_tracking(): year = request.args.get("year", date.today().year, type=int) config = AnnualTimeConfig.query.filter_by(user_id=current_user.id, year=year).first() if not config: config = AnnualTimeConfig(user_id=current_user.id, year=year) db.session.add(config) db.session.commit() templates = WorkScheduleTemplate.query.filter_by(user_id=current_user.id, is_active=True).order_by(WorkScheduleTemplate.id).all() assignments = {item.day_of_week: item for item in WorkSchedule.query.filter_by(user_id=current_user.id, is_active=True).all()} closures = CollegeClosure.query.filter( CollegeClosure.end_date >= date(year, 1, 1), CollegeClosure.start_date <= date(year, 12, 31) ).order_by(CollegeClosure.start_date).all() from ..core.models.planning import ClosureSchedule, PersonalLeave, Training, TrainingParticipant from ..core.services.planning_service import get_french_public_holidays closure_ids = [item.id for item in closures] work_days = ClosureWorkDay.query.filter(ClosureWorkDay.closure_id.in_(closure_ids)).all() if closure_ids else [] closure_schedules = ClosureSchedule.query.filter(ClosureSchedule.closure_id.in_(closure_ids), ClosureSchedule.is_active.is_(True)).all() if closure_ids else [] work_day_map = {item.work_date: item for item in work_days} closure_schedule_map = {(item.closure_id, item.day_of_week): item for item in closure_schedules} holidays = {item[0] for item in get_french_public_holidays(year)} leaves = PersonalLeave.query.filter(PersonalLeave.user_id == current_user.id, PersonalLeave.end_date >= date(year, 1, 1), PersonalLeave.start_date <= date(year, 12, 31)).all() leave_dates = {day for leave in leaves for day in (leave.start_date + timedelta(days=i) for i in range((leave.end_date - leave.start_date).days + 1))} entries = TimeEntry.query.filter( TimeEntry.user_id == current_user.id, TimeEntry.work_date >= date(year, 1, 1), TimeEntry.work_date <= date(year, 12, 31), ).order_by(TimeEntry.work_date.desc()).all() planned = 0 current = date(year, 1, 1) while current.year == year: hours = None if current not in holidays and current not in leave_dates: closure = next((item for item in closures if item.start_date <= current <= item.end_date), None) if closure: exceptional = work_day_map.get(current) if exceptional: hours = (exceptional.start_time, exceptional.end_time, exceptional.lunch_start, exceptional.lunch_end) elif closure.work_hours_type == "reduced": hours = (_time("09:00"), _time("12:00"), None, None) elif closure.work_hours_type == "custom": item = closure_schedule_map.get((closure.id, current.weekday())) if item: hours = (item.start_time, item.end_time, item.lunch_start, item.lunch_end) elif current.weekday() < 5 and assignments.get(current.weekday()): assignment = assignments[current.weekday()] source = assignment.template if assignment.template and assignment.template.is_active else assignment hours = (source.start_time, source.end_time, source.lunch_start, source.lunch_end) if hours: planned += _minutes(*hours) current += timedelta(days=1) actual = sum(item.actual_minutes or 0 for item in entries) return render_template( "planning/time_tracking.html", year=year, time_config=config, templates=templates, assignments=assignments, closures=closures, entries=entries, days={0: "Lundi", 1: "Mardi", 2: "Mercredi", 3: "Jeudi", 4: "Vendredi", 5: "Samedi", 6: "Dimanche"}, planned_minutes=planned, actual_minutes=actual, required_minutes=config.minutes_to_work, variance_minutes=actual - config.minutes_to_work, ) @planning_bp.route("/time/config", methods=["POST"]) @login_required def save_time_config(): year = request.form.get("year", date.today().year, type=int) config = AnnualTimeConfig.query.filter_by(user_id=current_user.id, year=year).first() if not config: config = AnnualTimeConfig(user_id=current_user.id, year=year) db.session.add(config) config.target_minutes = round(float(request.form.get("target_hours") or 1607) * 60) config.hardship_deduction_minutes = round(float(request.form.get("hardship_hours") or 0) * 60) config.notes = request.form.get("notes") or None db.session.commit() flash("Objectif annuel et déduction de pénibilité enregistrés.", "success") return _redirect(year) @planning_bp.route("/time/template", methods=["POST"]) @login_required def save_time_template(): count = WorkScheduleTemplate.query.filter_by(user_id=current_user.id, is_active=True).count() if count >= 3: flash("Vous pouvez définir au maximum trois horaires types.", "warning") return _redirect() start, end = _time(request.form.get("start_time")), _time(request.form.get("end_time")) lunch_start, lunch_end = _time(request.form.get("lunch_start")), _time(request.form.get("lunch_end")) if _minutes(start, end, lunch_start, lunch_end) <= 0: flash("Les horaires saisis sont invalides.", "danger") return _redirect() template = WorkScheduleTemplate( user_id=current_user.id, name=(request.form.get("name") or f"Horaire {count + 1}").strip(), start_time=start, end_time=end, lunch_start=lunch_start, lunch_end=lunch_end, ) db.session.add(template) db.session.commit() flash("Horaire type ajouté.", "success") return _redirect() @planning_bp.route("/time/template//delete", methods=["POST"]) @login_required def delete_time_template(id): template = WorkScheduleTemplate.query.filter_by(id=id, user_id=current_user.id).first_or_404() template.is_active = False for assignment in template.assignments: assignment.template_id = None db.session.commit() flash("Horaire type désactivé.", "success") return _redirect() @planning_bp.route("/time/template//edit", methods=["POST"]) @login_required def edit_time_template(id): template = WorkScheduleTemplate.query.filter_by(id=id, user_id=current_user.id, is_active=True).first_or_404() start, end = _time(request.form.get("start_time")), _time(request.form.get("end_time")) lunch_start = _time(request.form.get("lunch_start")) lunch_end = _time(request.form.get("lunch_end")) or template.lunch_end if _minutes(start, end, lunch_start, lunch_end) <= 0: flash("Les horaires saisis sont invalides.", "danger") return _redirect() template.name = (request.form.get("name") or template.name).strip() template.start_time, template.end_time = start, end template.lunch_start, template.lunch_end = lunch_start, lunch_end for assignment in template.assignments: assignment.start_time, assignment.end_time = start, end assignment.lunch_start, assignment.lunch_end = lunch_start, lunch_end db.session.commit() flash("Horaire type modifié.", "success") return _redirect() @planning_bp.route("/time/assignment", methods=["POST"]) @login_required def assign_time_template(): day = request.form.get("day_of_week", type=int) template_id = request.form.get("template_id", type=int) assignment = WorkSchedule.query.filter_by(user_id=current_user.id, day_of_week=day).first() if not template_id: if assignment: db.session.delete(assignment) else: template = WorkScheduleTemplate.query.filter_by(id=template_id, user_id=current_user.id, is_active=True).first_or_404() if not assignment: assignment = WorkSchedule(user_id=current_user.id, day_of_week=day) db.session.add(assignment) assignment.template_id = template.id assignment.start_time, assignment.end_time = template.start_time, template.end_time assignment.lunch_start, assignment.lunch_end = template.lunch_start, template.lunch_end assignment.is_active = True db.session.commit() return _redirect() @planning_bp.route("/time/entry", methods=["POST"]) @login_required def save_time_entry(): work_date = datetime.strptime(request.form["work_date"], "%Y-%m-%d").date() entry = TimeEntry.query.filter_by(user_id=current_user.id, work_date=work_date).first() if not entry: entry = TimeEntry(user_id=current_user.id, work_date=work_date) db.session.add(entry) entry.planned_minutes = int(request.form.get("planned_minutes") or 0) entry.actual_minutes = int(request.form.get("actual_minutes") or 0) entry.entry_type = request.form.get("entry_type") or "travail" entry.note = request.form.get("note") or None db.session.commit() flash("Suivi de la journée enregistré.", "success") return _redirect(work_date.year) @planning_bp.route("/time/import-vacations", methods=["POST"]) @login_required def import_school_vacations(): from app_new.lib_ext.vacances_scolaires import import_vacances_scolaires year = request.form.get("year", date.today().year, type=int) added, ignored, errors = import_vacances_scolaires(request.form.get("zone", "B"), year, current_user.id) flash(f"Vacances importées : {added} ajoutée(s), {ignored} déjà présente(s)." + (f" {len(errors)} erreur(s)." if errors else ""), "warning" if errors else "success") return _redirect(year) @planning_bp.route("/time/import-holidays", methods=["POST"]) @login_required def import_public_holidays(): from app_new.lib_ext.vacances_scolaires import import_jours_feries year = request.form.get("year", date.today().year, type=int) added, ignored = import_jours_feries(year, current_user.id) flash(f"Jours fériés importés : {added} ajouté(s), {ignored} déjà présent(s).", "success") return _redirect(year) @planning_bp.route("/time/permanence", methods=["POST"]) @login_required def add_permanence(): work_date = datetime.strptime(request.form["work_date"], "%Y-%m-%d").date() closure = CollegeClosure.query.filter( CollegeClosure.start_date <= work_date, CollegeClosure.end_date >= work_date, CollegeClosure.closure_type.in_(["vacances", "vacances_scolaires"]), ).first() if not closure: flash("La date doit se situer dans une période de vacances scolaires.", "danger") return _redirect(work_date.year) day = ClosureWorkDay.query.filter_by(closure_id=closure.id, work_date=work_date).first() if not day: day = ClosureWorkDay(closure_id=closure.id, work_date=work_date) db.session.add(day) day.start_time, day.end_time = _time(request.form.get("start_time")), _time(request.form.get("end_time")) day.lunch_start, day.lunch_end = _time(request.form.get("lunch_start")), _time(request.form.get("lunch_end")) day.notes = request.form.get("notes") or "Permanence" closure.work_hours_type = "custom" db.session.commit() flash("Permanence enregistrée.", "success") return _redirect(work_date.year) @planning_bp.route("/time/permanence//delete", methods=["POST"]) @login_required def delete_permanence(id): day = ClosureWorkDay.query.get_or_404(id) year = day.work_date.year db.session.delete(day) db.session.commit() flash("Permanence supprimée.", "success") return _redirect(year)