405 lines
20 KiB
Python
405 lines
20 KiB
Python
"""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 _valid_schedule(start, end, lunch_start=None, lunch_end=None):
|
|
if not start or not end or end <= start:
|
|
return False
|
|
if bool(lunch_start) != bool(lunch_end):
|
|
return False
|
|
if lunch_start and (lunch_start < start or lunch_end > end or lunch_end <= lunch_start):
|
|
return False
|
|
return _minutes(start, end, lunch_start, lunch_end) > 0
|
|
|
|
|
|
def _redirect(year=None):
|
|
return redirect(url_for("planning.time_tracking", year=year or _current_year()))
|
|
|
|
|
|
def _current_year():
|
|
today = date.today()
|
|
return today.year if today.month >= 9 else today.year - 1
|
|
|
|
|
|
def _period(year):
|
|
"""Année de travail : du 1er septembre au 31 août suivant."""
|
|
return date(year, 9, 1), date(year + 1, 8, 31)
|
|
|
|
|
|
def _month_starts(start, end):
|
|
"""Retourne les premiers jours de chaque mois de la période comptable."""
|
|
current = date(start.year, start.month, 1)
|
|
result = []
|
|
while current <= end:
|
|
result.append(current)
|
|
current = date(current.year + (current.month == 12), 1 if current.month == 12 else current.month + 1, 1)
|
|
return result
|
|
|
|
|
|
@planning_bp.route("/time")
|
|
@login_required
|
|
def time_tracking():
|
|
year = request.args.get("year", _current_year(), type=int)
|
|
period_start, period_end = _period(year)
|
|
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()
|
|
all_assignments = WorkSchedule.query.filter_by(user_id=current_user.id, is_active=True).all()
|
|
assignments = {item.day_of_week: item for item in all_assignments if item.academic_year == year}
|
|
# Compatibilité avec les affectations créées avant le suivi par année.
|
|
for item in all_assignments:
|
|
if item.academic_year is None and item.day_of_week not in assignments:
|
|
assignments[item.day_of_week] = item
|
|
closures = CollegeClosure.query.filter(
|
|
CollegeClosure.end_date >= period_start, CollegeClosure.start_date <= period_end
|
|
).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)} | {item[0] for item in get_french_public_holidays(year + 1)}
|
|
leaves = PersonalLeave.query.filter(PersonalLeave.user_id == current_user.id, PersonalLeave.end_date >= period_start, PersonalLeave.start_date <= period_end).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 >= period_start, TimeEntry.work_date <= period_end,
|
|
).order_by(TimeEntry.work_date.desc()).all()
|
|
planned = 0
|
|
planned_minutes_by_month = {}
|
|
permanence_minutes_by_month = {}
|
|
for work_day in work_days:
|
|
if period_start <= work_day.work_date <= period_end:
|
|
month = date(work_day.work_date.year, work_day.work_date.month, 1)
|
|
permanence_minutes_by_month[month] = permanence_minutes_by_month.get(month, 0) + _minutes(
|
|
work_day.start_time, work_day.end_time, work_day.lunch_start, work_day.lunch_end
|
|
)
|
|
current = period_start
|
|
while current <= period_end:
|
|
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:
|
|
day_minutes = _minutes(*hours)
|
|
planned += day_minutes
|
|
month = date(current.year, current.month, 1)
|
|
planned_minutes_by_month[month] = planned_minutes_by_month.get(month, 0) + day_minutes
|
|
current += timedelta(days=1)
|
|
permanence_options = []
|
|
permanence_groups = []
|
|
for closure in closures:
|
|
if closure.closure_type not in ("vacances", "vacances_scolaires"):
|
|
continue
|
|
current_day = max(closure.start_date, period_start)
|
|
end_day = min(closure.end_date, period_end)
|
|
while current_day <= end_day:
|
|
if current_day.weekday() < 5:
|
|
existing = work_day_map.get(current_day)
|
|
permanence_options.append({"date": current_day, "closure": closure, "entry": existing})
|
|
current_day += timedelta(days=1)
|
|
if permanence_options:
|
|
group_options = [item for item in permanence_options if item["closure"].id == closure.id]
|
|
if group_options:
|
|
permanence_groups.append({"closure": closure, "options": group_options})
|
|
actual = sum(item.actual_minutes or 0 for item in entries)
|
|
# Synthèse mensuelle : les permanences sont isolées du temps de travail
|
|
# planifié habituel, et les formations viennent des convocations validées.
|
|
monthly_summary = []
|
|
trainings = Training.query.join(TrainingParticipant).filter(
|
|
TrainingParticipant.user_id == current_user.id,
|
|
TrainingParticipant.is_confirmed.is_(True),
|
|
Training.end_date >= period_start,
|
|
Training.start_date <= period_end,
|
|
).all()
|
|
for month in _month_starts(period_start, period_end):
|
|
month_end = date(month.year + (month.month == 12), 1 if month.month == 12 else month.month + 1, 1) - timedelta(days=1)
|
|
month_entries = [item for item in entries if month <= item.work_date <= month_end]
|
|
training_minutes = 0
|
|
for training in trainings:
|
|
start = max(training.start_date or month, month, period_start)
|
|
end = min(training.end_date or start, month_end, period_end)
|
|
if end < start:
|
|
continue
|
|
day = start
|
|
while day <= end:
|
|
if day.weekday() < 5:
|
|
if training.start_time and training.end_time:
|
|
training_minutes += _minutes(training.start_time, training.end_time)
|
|
elif assignments.get(day.weekday()):
|
|
assignment = assignments[day.weekday()]
|
|
source = assignment.template if assignment.template and assignment.template.is_active else assignment
|
|
training_minutes += _minutes(source.start_time, source.end_time, source.lunch_start, source.lunch_end)
|
|
day += timedelta(days=1)
|
|
monthly_summary.append({
|
|
"month": month,
|
|
"planned": max(planned_minutes_by_month.get(month, 0) - permanence_minutes_by_month.get(month, 0), 0),
|
|
"permanence": permanence_minutes_by_month.get(month, 0),
|
|
"retard": sum(item.actual_minutes or 0 for item in month_entries if item.entry_type == "retard"),
|
|
"supplementaire": sum(item.actual_minutes or 0 for item in month_entries if item.entry_type == "supplementaire"),
|
|
"formation": training_minutes,
|
|
})
|
|
# Solde restant selon la règle de l'agent : objectif - prévu - retards
|
|
# + heures supplémentaires + formations.
|
|
balance_minutes = config.minutes_to_work - sum(item["planned"] for item in monthly_summary) - sum(item["retard"] for item in monthly_summary) + sum(item["supplementaire"] for item in monthly_summary) + sum(item["formation"] for item in monthly_summary)
|
|
return render_template(
|
|
"planning/time_tracking.html", year=year, period_start=period_start, period_end=period_end, time_config=config, templates=templates,
|
|
assignments=assignments, closures=closures, permanence_options=permanence_options, permanence_groups=permanence_groups, entries=entries,
|
|
days={0: "Lundi", 1: "Mardi", 2: "Mercredi", 3: "Jeudi", 4: "Vendredi", 5: "Samedi", 6: "Dimanche"},
|
|
months={1: "Janvier", 2: "Février", 3: "Mars", 4: "Avril", 5: "Mai", 6: "Juin", 7: "Juillet", 8: "Août", 9: "Septembre", 10: "Octobre", 11: "Novembre", 12: "Décembre"},
|
|
planned_minutes=planned, actual_minutes=actual,
|
|
required_minutes=config.minutes_to_work, variance_minutes=actual - config.minutes_to_work,
|
|
balance_minutes=balance_minutes,
|
|
monthly_summary=monthly_summary,
|
|
)
|
|
|
|
|
|
@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 >= 6:
|
|
flash("Vous pouvez définir au maximum six 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 not _valid_schedule(start, end, lunch_start, lunch_end):
|
|
flash("Horaires invalides : la fin doit être après le début et la pause doit être complète et comprise dans la journée.", "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/<int:id>/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/<int:id>/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 not _valid_schedule(start, end, lunch_start, lunch_end):
|
|
flash("Horaires invalides : la fin doit être après le début et la pause doit être complète et comprise dans la journée.", "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)
|
|
year = request.form.get("year", _current_year(), 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, academic_year=year).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, academic_year=year)
|
|
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(year)
|
|
|
|
|
|
@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", "C"), 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/vacation", methods=["POST"])
|
|
@login_required
|
|
def add_manual_vacation():
|
|
"""Ajoute une période de vacances personnalisée."""
|
|
try:
|
|
start_date = datetime.strptime(request.form["start_date"], "%Y-%m-%d").date()
|
|
end_date = datetime.strptime(request.form["end_date"], "%Y-%m-%d").date()
|
|
except (KeyError, ValueError):
|
|
flash("Dates de vacances invalides.", "danger")
|
|
return _redirect(request.form.get("year", type=int))
|
|
if end_date < start_date:
|
|
flash("La date de fin doit être postérieure ou égale à la date de début.", "danger")
|
|
return _redirect(start_date.year)
|
|
db.session.add(CollegeClosure(
|
|
name=(request.form.get("name") or "Vacances personnalisées").strip(),
|
|
start_date=start_date, end_date=end_date,
|
|
closure_type="vacances", work_hours_type="none",
|
|
notes="Période ajoutée manuellement",
|
|
))
|
|
db.session.commit()
|
|
flash("Période de vacances ajoutée.", "success")
|
|
return _redirect(start_date.year)
|
|
|
|
|
|
@planning_bp.route("/time/vacation/<int:id>/delete", methods=["POST"])
|
|
@login_required
|
|
def delete_vacation(id):
|
|
vacation = CollegeClosure.query.filter(
|
|
CollegeClosure.id == id,
|
|
CollegeClosure.closure_type.in_(["vacances", "vacances_scolaires"]),
|
|
).first_or_404()
|
|
db.session.delete(vacation)
|
|
db.session.commit()
|
|
flash("Période de vacances supprimée.", "success")
|
|
return _redirect()
|
|
|
|
|
|
@planning_bp.route("/time/permanences", methods=["POST"])
|
|
@login_required
|
|
def save_permanences():
|
|
year = request.form.get("year", date.today().year, type=int)
|
|
template = WorkScheduleTemplate.query.filter_by(id=request.form.get("template_id", type=int), user_id=current_user.id, is_active=True).first_or_404()
|
|
selected_dates = set(request.form.getlist("work_dates"))
|
|
period_start, period_end = _period(year)
|
|
closures = CollegeClosure.query.filter(
|
|
CollegeClosure.end_date >= period_start, CollegeClosure.start_date <= period_end,
|
|
CollegeClosure.closure_type.in_(["vacances", "vacances_scolaires"]),
|
|
).all()
|
|
for closure in closures:
|
|
current_day = max(closure.start_date, period_start)
|
|
end_day = min(closure.end_date, period_end)
|
|
while current_day <= end_day:
|
|
if current_day.weekday() < 5 and current_day.isoformat() in selected_dates:
|
|
day = ClosureWorkDay.query.filter_by(closure_id=closure.id, work_date=current_day).first()
|
|
if not day:
|
|
day = ClosureWorkDay(closure_id=closure.id, work_date=current_day)
|
|
db.session.add(day)
|
|
day.template_id = template.id
|
|
day.start_time, day.end_time = template.start_time, template.end_time
|
|
day.lunch_start, day.lunch_end = template.lunch_start, template.lunch_end
|
|
day.notes = "Permanence"
|
|
closure.work_hours_type = "custom"
|
|
current_day += timedelta(days=1)
|
|
db.session.commit()
|
|
flash(f"{len(selected_dates)} jour(s) de permanence enregistré(s) avec « {template.name} ».", "success")
|
|
return _redirect(year)
|
|
|
|
|
|
@planning_bp.route("/time/permanence/<int:id>/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)
|