255 lines
12 KiB
Python
255 lines
12 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 _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/<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 _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/<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)
|