d760842121
Le forfait était suivi brut, sans bonification. Considéré réalisé en
journée du jour où il est attribué, il reçoit désormais le facteur
journée correspondant : ×1 en semaine, ×1,5 le samedi, ×2 le dimanche
ou un jour férié (suit les règles de ratios configurées). Toujours hors
heures travaillées ; affiché pondéré sur la ligne du jour, dans les
totaux du récapitulatif et expliqué dans Réglages.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
1388 lines
64 KiB
Python
1388 lines
64 KiB
Python
import asyncio
|
||
import hashlib
|
||
import hmac
|
||
import html
|
||
import json
|
||
import logging
|
||
import re
|
||
import secrets
|
||
import time
|
||
from collections import defaultdict
|
||
from datetime import datetime, timedelta
|
||
from itertools import groupby
|
||
from urllib.parse import urlsplit
|
||
|
||
import bcrypt
|
||
from fastapi import FastAPI, Form, Request, Response
|
||
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
||
from fastapi.templating import Jinja2Templates
|
||
from fastapi.staticfiles import StaticFiles
|
||
|
||
from mail import send_mail
|
||
from models import (
|
||
load_config, load_week_pointages, save_week_pointages, load_conges,
|
||
toggle_conge, list_all_weeks, toggle_astreinte,
|
||
astreinte_active, load_astreintes_effectives,
|
||
load_feries, toggle_ferie,
|
||
load_notif_config, save_notif_config, find_user_by_token, save_presence,
|
||
regenerate_presence_token,
|
||
load_auth, save_auth, find_user_by_reset_token,
|
||
load_plages, save_plages, heures_jour_min,
|
||
append_log, load_logs, load_or_create_secret,
|
||
load_facteurs_astreinte, save_facteurs_astreinte,
|
||
)
|
||
from calcul import (
|
||
compute_week, minutes_to_hhmm, minutes_to_time, hhmm_to_minutes, heures_dues,
|
||
heures_travaillees, jour_astreinte_qualifie, exclusion_retour, couvert_detail,
|
||
nominaux_du_jour, retour_apres_repos, retour_repos, retour_txt, decompte_astreinte,
|
||
fenetre_astreinte_jour, regle_journee_astreinte,
|
||
)
|
||
from stats import compute_all_stats
|
||
from export import export_semaine_ods, export_annee_ods
|
||
from notifications import reminder_loop, send_ntfy, is_safe_ntfy_server
|
||
|
||
logging.basicConfig(
|
||
level=getattr(logging, load_config().get("log_level", "DEBUG").upper(), logging.DEBUG),
|
||
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
||
)
|
||
logger = logging.getLogger("pointeuse")
|
||
|
||
app = FastAPI(title="Décompte Horaire")
|
||
|
||
|
||
@app.middleware("http")
|
||
async def csrf_origin_check(request: Request, call_next):
|
||
"""Protection CSRF stateless sur les requêtes mutatives.
|
||
|
||
Stratégie défense-en-profondeur combinée au cookie SameSite=Lax :
|
||
si la requête porte un header Origin ou Referer, on vérifie que son host
|
||
correspond à celui attendu (config.base_url si définie, sinon le Host de
|
||
la requête). Une requête cross-site aura un Origin différent.
|
||
L'absence d'Origin est tolérée (cas des clients non-navigateurs comme
|
||
Tasker ou curl, où SameSite=Lax suffit).
|
||
"""
|
||
if request.method in ("POST", "PUT", "DELETE", "PATCH"):
|
||
origin = request.headers.get("origin") or request.headers.get("referer")
|
||
if origin:
|
||
try:
|
||
parsed = urlsplit(origin)
|
||
except ValueError:
|
||
return JSONResponse({"error": "csrf"}, status_code=403)
|
||
origin_host = (parsed.hostname or "").lower()
|
||
configured_base = (load_config().get("base_url") or "").strip()
|
||
if configured_base:
|
||
try:
|
||
expected = (urlsplit(configured_base).hostname or "").lower()
|
||
except ValueError:
|
||
expected = ""
|
||
else:
|
||
expected = (request.headers.get("host") or "").lower().split(":", 1)[0]
|
||
if origin_host and expected and origin_host != expected:
|
||
logger.warning(
|
||
"CSRF: origin rejettée origin=%r host=%r path=%r",
|
||
origin, request.headers.get("host"), request.url.path,
|
||
)
|
||
return JSONResponse({"error": "csrf"}, status_code=403)
|
||
response = await call_next(request)
|
||
# ── En-têtes de sécurité appliqués à toutes les réponses ──
|
||
response.headers["X-Content-Type-Options"] = "nosniff"
|
||
response.headers["X-Frame-Options"] = "DENY"
|
||
response.headers["Referrer-Policy"] = "same-origin"
|
||
# CSP relaxée : on autorise Google Fonts (chargé depuis base.html) et les
|
||
# scripts/styles inline (semaine.html et stats.html ont du JS/CSS inline).
|
||
# 'unsafe-inline' reste nécessaire tant que ces ressources ne sont pas
|
||
# extraites dans des fichiers statiques.
|
||
response.headers["Content-Security-Policy"] = (
|
||
"default-src 'self'; "
|
||
"script-src 'self' 'unsafe-inline'; "
|
||
"style-src 'self' 'unsafe-inline' https://fonts.googleapis.com; "
|
||
"font-src 'self' https://fonts.gstatic.com; "
|
||
"img-src 'self' data:; "
|
||
"connect-src 'self'; "
|
||
"frame-ancestors 'none'; "
|
||
"base-uri 'self'"
|
||
)
|
||
# HSTS uniquement en HTTPS (sinon on risquerait de bloquer un user en HTTP)
|
||
if request.url.scheme == "https":
|
||
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
|
||
return response
|
||
|
||
|
||
@app.on_event("startup")
|
||
async def _log_config_at_startup():
|
||
cfg = load_config()
|
||
smtp = cfg.get("smtp", {})
|
||
logger.info(
|
||
"config chargée: heures_jour=%r allowed_email_domain=%r mail_from=%r "
|
||
"smtp.host=%r smtp.port=%r smtp.user=%r smtp.password=%s smtp.use_tls=%r",
|
||
cfg.get("heures_jour"), cfg.get("allowed_email_domain"), cfg.get("mail_from"),
|
||
smtp.get("host"), smtp.get("port"), smtp.get("user"),
|
||
"***" if smtp.get("password") else "(vide)", smtp.get("use_tls"),
|
||
)
|
||
|
||
|
||
@app.on_event("startup")
|
||
async def _start_reminder_loop():
|
||
asyncio.create_task(reminder_loop())
|
||
templates = Jinja2Templates(directory="templates")
|
||
templates.env.filters["zip"] = zip
|
||
app.mount("/static", StaticFiles(directory="static"), name="static")
|
||
|
||
JOURS_FR = ["Lundi", "Mardi", "Mercredi", "Jeudi", "Vendredi", "Samedi", "Dimanche"]
|
||
_USERNAME_RE = re.compile(r"^[a-zA-Z0-9_.\-]{1,64}$")
|
||
_EMAIL_RE = re.compile(r"^([a-zA-Z0-9_.\-]{1,64})@([a-zA-Z0-9.\-]{1,255})$")
|
||
_HHMM_RE = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$")
|
||
_JOURS_AS_VALIDES = ("semaine", "sam", "dim", "ferie")
|
||
|
||
# Échappement HTML pour les fragments construits en f-string hors templates Jinja
|
||
# (autoescape ne s'applique qu'aux templates). Defense-in-depth contre le XSS stocké.
|
||
_e = html.escape
|
||
|
||
|
||
# ── Auth helpers ──────────────────────────────────────────────────────────────
|
||
|
||
def _sign_user_id(user_id: str) -> str:
|
||
"""Return the signed session cookie value: `<user_id>.<hmac_hex>`.
|
||
|
||
The signature authenticates the user_id without storing server-side session
|
||
state. The HMAC key lives in /data/secret.key (see load_or_create_secret).
|
||
"""
|
||
secret = load_or_create_secret()
|
||
sig = hmac.new(secret, user_id.encode(), hashlib.sha256).hexdigest()
|
||
return f"{user_id}.{sig}"
|
||
|
||
|
||
def _verify_signed_cookie(raw: str) -> str | None:
|
||
"""Return the user_id if the cookie carries a valid signature, else None."""
|
||
if not raw or "." not in raw:
|
||
return None
|
||
user_id, _, sig = raw.rpartition(".")
|
||
if not user_id or not sig or not _USERNAME_RE.match(user_id):
|
||
return None
|
||
secret = load_or_create_secret()
|
||
expected = hmac.new(secret, user_id.encode(), hashlib.sha256).hexdigest()
|
||
# Constant-time comparison to avoid leaking signature bytes via timing.
|
||
if hmac.compare_digest(sig, expected):
|
||
return user_id
|
||
return None
|
||
|
||
|
||
def _get_user(request: Request) -> str | None:
|
||
"""Extract current user from signed cookie."""
|
||
return _verify_signed_cookie(request.cookies.get("user_id", ""))
|
||
|
||
|
||
def _set_auth_cookie(resp: Response, user_id: str, request: Request | None = None) -> None:
|
||
"""Set the signed, HttpOnly, SameSite session cookie.
|
||
|
||
Le drapeau `secure` est positionné automatiquement si la requête courante
|
||
est en HTTPS, ou si config.force_secure_cookies est vrai. Cela évite que
|
||
le cookie ne transite en clair sur un downgrade HTTP.
|
||
"""
|
||
secure = False
|
||
if request is not None and request.url.scheme == "https":
|
||
secure = True
|
||
if load_config().get("force_secure_cookies"):
|
||
secure = True
|
||
resp.set_cookie(
|
||
"user_id", _sign_user_id(user_id),
|
||
httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30,
|
||
secure=secure,
|
||
)
|
||
|
||
|
||
def allowed_email_domain() -> str:
|
||
return load_config().get("allowed_email_domain", "")
|
||
|
||
|
||
def mail_from_address() -> str:
|
||
return load_config().get("mail_from") or "no-reply@pointeuse.local"
|
||
|
||
|
||
def smtp_config() -> dict:
|
||
return load_config().get("smtp", {})
|
||
|
||
|
||
# ── Rate-limiting des routes d'authentification ───────────────────────────────
|
||
# Compteur simple en mémoire, fenêtre glissante de LOGIN_WINDOW_S secondes.
|
||
# Suffit pour bloquer le brute-force / mail-bombing sur une app mono-utilisateur.
|
||
# Reset entre tests via main._LOGIN_ATTEMPTS.clear().
|
||
|
||
_LOGIN_ATTEMPTS: dict[str, list[float]] = defaultdict(list)
|
||
LOGIN_WINDOW_S = 60
|
||
DEFAULT_LOGIN_MAX_PER_MIN = 10
|
||
|
||
|
||
def _check_login_rate_limit(key: str, max_per_min: int | None = None) -> bool:
|
||
"""Retourne True si la tentative est autorisée, False sinon.
|
||
|
||
max_per_min <= 0 désactive la limite (utile pour tests ou config).
|
||
"""
|
||
if max_per_min is None:
|
||
max_per_min = int(load_config().get("login_rate_limit_per_min", DEFAULT_LOGIN_MAX_PER_MIN))
|
||
if max_per_min <= 0:
|
||
return True
|
||
now = time.monotonic()
|
||
cutoff = now - LOGIN_WINDOW_S
|
||
recent = [t for t in _LOGIN_ATTEMPTS[key] if t > cutoff]
|
||
_LOGIN_ATTEMPTS[key] = recent
|
||
if len(recent) >= max_per_min:
|
||
return False
|
||
recent.append(now)
|
||
return True
|
||
|
||
|
||
def _client_ip(request: Request) -> str:
|
||
"""Retourne l'IP du client (sans confiance X-Forwarded-For par défaut)."""
|
||
return request.client.host if request.client else "unknown"
|
||
|
||
|
||
def _safe_base_url(request: Request) -> str | None:
|
||
"""Retourne une URL de base fiable pour construire des liens (emails, etc.).
|
||
|
||
Anti open-redirect / phishing : le header Host est contrôlable par le client
|
||
(curl/DevTools), donc `request.base_url` ne doit pas être utilisée telle
|
||
quelle dans un email envoyé à l'utilisateur, sinon un attaquant peut faire
|
||
envoyer un mail légitime contenant `https://phishing.attacker/set-password/<token>`
|
||
avec un token réellement valide.
|
||
|
||
Stratégie :
|
||
- Si `base_url` est défini en config, l'utiliser (schéma http/https requis).
|
||
- Sinon, accepter le Host seulement s'il correspond à `allowed_email_domain`
|
||
(suffixe strict) ou à localhost (utile pour les tests/dev locaux).
|
||
- Sinon, retourner None : l'appelant doit refuser l'envoi.
|
||
"""
|
||
cfg = load_config()
|
||
configured = (cfg.get("base_url") or "").strip().rstrip("/")
|
||
if configured:
|
||
try:
|
||
parts = urlsplit(configured)
|
||
if parts.scheme in ("http", "https") and parts.hostname:
|
||
return configured
|
||
except ValueError:
|
||
pass
|
||
logger.warning("base_url configurée invalide: %r", configured)
|
||
return None
|
||
|
||
host = (request.headers.get("host") or "").strip()
|
||
if not host:
|
||
return None
|
||
host_l = host.lower()
|
||
hostname = host_l.split(":", 1)[0]
|
||
scheme = "https" if request.url.scheme == "https" else "http"
|
||
|
||
if hostname in ("localhost", "127.0.0.1", "::1"):
|
||
return f"{scheme}://{host}"
|
||
|
||
allowed = (cfg.get("allowed_email_domain") or "").strip().lower()
|
||
if allowed and (host_l == allowed or host_l.endswith("." + allowed)):
|
||
return f"{scheme}://{host}"
|
||
|
||
return None
|
||
|
||
|
||
def _require_user(request: Request):
|
||
"""Return user_id or redirect to login."""
|
||
uid = _get_user(request)
|
||
if not uid:
|
||
return None
|
||
return uid
|
||
|
||
|
||
# ── Shared helpers ────────────────────────────────────────────────────────────
|
||
|
||
def week_dates(year: int, week: int) -> list[str]:
|
||
"""Les 7 jours ISO de la semaine : le week-end est affiché (heures dues nulles)
|
||
pour pouvoir y déclarer des astreintes et y saisir les interventions."""
|
||
first = datetime.strptime(f"{year}-W{week:02d}-1", "%G-W%V-%u").date()
|
||
return [(first + timedelta(days=i)).strftime("%Y-%m-%d") for i in range(7)]
|
||
|
||
|
||
def _weeks_by_year(all_weeks: list) -> dict:
|
||
result = {}
|
||
for yr, wks in groupby(sorted(all_weeks, reverse=True), key=lambda x: x[0]):
|
||
result[yr] = sorted(wks, key=lambda x: x[1], reverse=True)
|
||
return result
|
||
|
||
|
||
def _contexte_semaine_precedente(year: int, week: int, dates: list[str], user_id: str) -> list[dict]:
|
||
"""Pointages de la semaine ISO précédente (jours avant la semaine affichée) :
|
||
la fenêtre glissante de 7 jours du repos hebdomadaire de 35h déborde sur la
|
||
semaine d'avant."""
|
||
prev = (datetime.strptime(dates[0], "%Y-%m-%d").date() - timedelta(days=7)).isocalendar()
|
||
if (prev.year, prev.week) == (year, week):
|
||
return []
|
||
return [r for r in load_week_pointages(prev.year, prev.week, user_id) if r.get("date") < dates[0]]
|
||
|
||
|
||
def _build_soldes_ctx(year, week, conges, h_jour, user_id):
|
||
dates = week_dates(year, week)
|
||
raw = load_week_pointages(year, week, user_id)
|
||
p_by_date = {p["date"]: p for p in raw}
|
||
pointages = [
|
||
p_by_date.get(d, {"date": d, "matin_entree": None, "matin_sortie": None,
|
||
"aprem_entree": None, "aprem_sortie": None})
|
||
for d in dates
|
||
]
|
||
contexte = _contexte_semaine_precedente(year, week, dates, user_id)
|
||
result = compute_week(pointages, conges, h_jour, load_plages(user_id), load_astreintes_effectives(user_id), load_facteurs_astreinte(), load_feries(user_id), contexte)
|
||
result["heures_jour"] = minutes_to_hhmm(h_jour)
|
||
# Panneau « aujourd'hui » : uniquement si le jour courant appartient à la
|
||
# semaine affichée (re-rendu à chaque enregistrement via _soldes.html).
|
||
today_str = datetime.today().date().isoformat()
|
||
jour_today = next((j for j in result["jours"] if j["date"] == today_str), None)
|
||
if jour_today is not None:
|
||
trav_total = sum(j["travaille_min"] for j in result["jours"])
|
||
du_total = trav_total - result["solde_min"]
|
||
pct = min(100, round(trav_total * 100 / du_total)) if du_total > 0 else None
|
||
result["today"] = {
|
||
"nom": JOURS_FR[datetime.strptime(today_str, "%Y-%m-%d").weekday()],
|
||
"jour": jour_today,
|
||
"pct": pct,
|
||
"trav_total": minutes_to_hhmm(trav_total),
|
||
"du_total": minutes_to_hhmm(du_total),
|
||
}
|
||
return {"result": result}
|
||
|
||
|
||
def _build_jour_ctx(date, conges, h_jour, user_id):
|
||
d = datetime.strptime(date, "%Y-%m-%d").date()
|
||
iso = d.isocalendar()
|
||
raw = load_week_pointages(iso.year, iso.week, user_id)
|
||
plages = load_plages(user_id)
|
||
astreintes = set(load_astreintes_effectives(user_id))
|
||
feries = set(load_feries(user_id))
|
||
p_by_date = {p["date"]: p for p in raw}
|
||
# Repos hebdomadaire de 35h (7 jours glissants) : la fenêtre des 6 jours
|
||
# précédents peut déborder sur la semaine ISO précédente.
|
||
debut_fenetre = (d - timedelta(days=6)).isocalendar()
|
||
if (debut_fenetre.year, debut_fenetre.week) != (iso.year, iso.week):
|
||
for row in load_week_pointages(debut_fenetre.year, debut_fenetre.week, user_id):
|
||
p_by_date.setdefault(row.get("date"), row)
|
||
p = p_by_date.get(date, {"date": date, "matin_entree": None, "matin_sortie": None,
|
||
"aprem_entree": None, "aprem_sortie": None})
|
||
conges_date = {c["type"] for c in conges if c["date"] == date}
|
||
jour_ferie = date in feries
|
||
# Retour au bureau après repos, calculé depuis la fin de la dernière
|
||
# activité d'astreinte (11h quotidiennes) et, si activée, la contrainte de
|
||
# repos hebdomadaire de 35h sur 7 jours glissants
|
||
retour_min = retour_repos(date, p_by_date, plages, astreintes)
|
||
matin_ex, aprem_ex = exclusion_retour(conges_date, retour_min, plages, jour_ferie)
|
||
forfait = (hhmm_to_minutes(plages["astreinte_forfait"]) if plages.get("astreinte_forfait") else 0) \
|
||
if date in astreintes else 0
|
||
# Le forfait est considéré réalisé en journée du jour AS déclaré : il reçoit
|
||
# la bonification « journée » de ce jour (×1 semaine, ×1,5 samedi, ×2
|
||
# dimanche/férié par défaut) — pondéré, mais toujours hors heures travaillées.
|
||
regle_journee = regle_journee_astreinte(date, load_facteurs_astreinte(), feries) if forfait else None
|
||
forfait_pondere = int(forfait * (regle_journee or {"facteur": 1.0})["facteur"] + 0.5) if forfait else 0
|
||
# Les heures de présence obligatoire couvertes par le repos de 11h sont
|
||
# comptées comme effectuées : créditées au travaillé, dues complètes (le
|
||
# forfait d'astreinte, lui, reste hors heures travaillées). Sans heures
|
||
# dues (week-end, férié, congé), rien à créditer.
|
||
du_full = heures_dues(date, conges, h_jour, plages, None, jour_ferie)
|
||
couvert = min(matin_ex + aprem_ex, du_full)
|
||
trav = heures_travaillees(p, retour_min) + couvert
|
||
du = du_full
|
||
matin_nom, aprem_nom = nominaux_du_jour(conges_date, du_full, matin_ex, aprem_ex, jour_ferie)
|
||
delta = trav - du
|
||
# Décompte astreinte du jour (compteur séparé, hors heures de semaine) :
|
||
# seulement si des événements sont stockés et que le jour est qualifié.
|
||
events = p.get("astreinte_events") or []
|
||
dec = (decompte_astreinte(date, events, load_facteurs_astreinte(), fenetre_astreinte_jour(date, astreintes, feries), feries)
|
||
if events and jour_astreinte_qualifie(date, astreintes) else None)
|
||
return {
|
||
**p,
|
||
"travaille_min": trav, "du_min": du, "delta_min": delta,
|
||
"travaille": minutes_to_hhmm(trav),
|
||
"du": minutes_to_hhmm(du),
|
||
"delta": minutes_to_hhmm(delta),
|
||
"conge": conges_date,
|
||
"ferie": jour_ferie,
|
||
"forfait_min": forfait,
|
||
"forfait_h": minutes_to_hhmm(forfait) if forfait else "",
|
||
"forfait_pondere_min": forfait_pondere,
|
||
"forfait_pondere_h": minutes_to_hhmm(forfait_pondere) if forfait_pondere else "",
|
||
"forfait_regle": regle_journee["nom"] if regle_journee else "",
|
||
"forfait_facteur_txt": regle_journee["facteur_txt"] if regle_journee else "1",
|
||
"du_full_min": du_full,
|
||
"matin_excl_min": matin_ex, "aprem_excl_min": aprem_ex,
|
||
"couvert_min": couvert,
|
||
"couvert_h": minutes_to_hhmm(couvert) if couvert else "",
|
||
"couvert_detail": couvert_detail(plages, retour_min, matin_ex, aprem_ex) if couvert else "",
|
||
"matin_nominal_min": matin_nom, "aprem_nominal_min": aprem_nom,
|
||
"astreinte": date in astreintes,
|
||
"astreinte_qualifiee": jour_astreinte_qualifie(date, astreintes),
|
||
"retour_min": retour_min,
|
||
"retour": retour_txt(retour_min, date) if retour_min is not None else None,
|
||
"astreinte_min": dec["total_min"] if dec else 0,
|
||
"astreinte_h": dec["total_h"] if dec else "",
|
||
}
|
||
|
||
|
||
def _oob_calc(date, jour):
|
||
trav_inner = f'<span class="calc-num">{"—" if not jour["travaille_min"] else _e(jour["travaille"])}</span>'
|
||
if jour.get("couvert_min"):
|
||
trav_inner += (f'<span class="calc-couvert" title="Présence obligatoire couverte par le repos (11h quotidiennes, ou 35h hebdomadaires si activées) '
|
||
f'({_e(jour["couvert_detail"])}) : comptée comme effectuée">+{_e(jour["couvert_h"])} repos</span>')
|
||
if jour["travaille_min"]:
|
||
cls = "delta-pos" if jour["delta_min"] > 0 else ("delta-neg" if jour["delta_min"] < 0 else "delta-zero")
|
||
sign = "+" if jour["delta_min"] > 0 else ""
|
||
delta_inner = f'<span class="{cls}">{sign}{_e(jour["delta"])}</span>'
|
||
else:
|
||
delta_inner = '<span class="delta-zero">—</span>'
|
||
return (
|
||
f'<span id="calc-{_e(date)}-trav" hx-swap-oob="true">{trav_inner}</span>'
|
||
f'<span id="calc-{_e(date)}-delta" hx-swap-oob="true">{delta_inner}</span>'
|
||
)
|
||
|
||
|
||
def _oob_week_extras(result):
|
||
html = ""
|
||
for jour in result["jours"]:
|
||
date = jour["date"]
|
||
cm = jour["delta_cumul_min"]
|
||
if jour.get("cumul_visible"):
|
||
if cm == 0:
|
||
cumul_inner = '<span class="delta-zero">0h00</span>'
|
||
else:
|
||
cls = "delta-pos" if cm > 0 else "delta-neg"
|
||
sign = "+" if cm > 0 else "-"
|
||
cumul_inner = f'<span class="{cls}">{sign}{jour["delta_cumul"]}</span>'
|
||
else:
|
||
cumul_inner = '<span class="delta-zero">—</span>'
|
||
html += f'<span id="calc-{_e(date)}-cumul" hx-swap-oob="true">{cumul_inner}</span>'
|
||
parts = []
|
||
if jour.get("cible_matin"):
|
||
parts.append(f'<span class="cible-num">M {_e(jour["cible_matin_entree"])} → {_e(jour["cible_matin"])}</span>')
|
||
if jour.get("cible_aprem"):
|
||
parts.append(f'<span class="cible-num">A {_e(jour["cible_aprem_entree"])} → {_e(jour["cible_aprem"])}</span>')
|
||
cible_inner = " ".join(parts) if parts else '<span class="delta-zero">—</span>'
|
||
html += f'<span id="calc-{_e(date)}-cible" hx-swap-oob="true">{cible_inner}</span>'
|
||
return html
|
||
|
||
|
||
def _oob_time_cells(date, jour, plages):
|
||
cg = jour["conge"]
|
||
matin_cg = "matin" in cg or "jour" in cg
|
||
aprem_cg = "aprem" in cg or "jour" in cg
|
||
retour_min = jour.get("retour_min") or 0
|
||
# Une demi-journée entièrement couverte par le repos de 11h n'est plus exigée
|
||
matin_requis = not matin_cg and jour["du_min"] > 0 and jour.get("matin_nominal_min", 0) > 0
|
||
aprem_requis = not aprem_cg and jour["du_min"] > 0 and jour.get("aprem_nominal_min", 0) > 0
|
||
# L'heure de retour décale d'autant la limite d'arrivée admise sans avertissement
|
||
matin_warn = matin_requis and (
|
||
(jour.get("matin_entree") and hhmm_to_minutes(jour["matin_entree"]) > max(hhmm_to_minutes(plages["matin_debut"]), retour_min)) or
|
||
(jour.get("matin_sortie") and hhmm_to_minutes(jour["matin_sortie"]) < hhmm_to_minutes(plages["matin_fin"]))
|
||
)
|
||
aprem_warn = aprem_requis and (
|
||
(jour.get("aprem_entree") and hhmm_to_minutes(jour["aprem_entree"]) > max(hhmm_to_minutes(plages["aprem_debut"]), retour_min)) or
|
||
(jour.get("aprem_sortie") and hhmm_to_minutes(jour["aprem_sortie"]) < hhmm_to_minutes(plages["aprem_fin"]))
|
||
)
|
||
matin_cls = "td-fill cg" if matin_cg else ("td-fill warn" if matin_warn else "td-fill")
|
||
aprem_cls = "td-fill cg" if aprem_cg else ("td-fill warn" if aprem_warn else "td-fill")
|
||
matin_html = (
|
||
f'<div id="cg-matin-{_e(date)}" hx-swap-oob="true" data-label="Matin" class="{matin_cls}">'
|
||
f'<div class="time-pair">'
|
||
f'<input class="time-input" type="time" name="matin_entree" data-date="{_e(date)}" value="{_e(jour["matin_entree"] or "")}">'
|
||
f'<span class="time-sep">→</span>'
|
||
f'<input class="time-input" type="time" name="matin_sortie" data-date="{_e(date)}" value="{_e(jour["matin_sortie"] or "")}">'
|
||
f'</div></div>'
|
||
)
|
||
aprem_html = (
|
||
f'<div id="cg-aprem-{_e(date)}" hx-swap-oob="true" data-label="Après-midi" class="{aprem_cls}">'
|
||
f'<div class="time-pair">'
|
||
f'<input class="time-input" type="time" name="aprem_entree" data-date="{_e(date)}" value="{_e(jour["aprem_entree"] or "")}">'
|
||
f'<span class="time-sep">→</span>'
|
||
f'<input class="time-input" type="time" name="aprem_sortie" data-date="{_e(date)}" value="{_e(jour["aprem_sortie"] or "")}">'
|
||
f'</div></div>'
|
||
)
|
||
return matin_html + aprem_html
|
||
|
||
|
||
def _btns_html(date, jour, as_active=True):
|
||
"""Bouton d'état d'une ligne (cible du swap HTMX) : pastilles résumant
|
||
congés / férié / astreinte, menu déroulant des bascules. Doit rester
|
||
synchronisé avec le rendu Jinja de _ligne.html. as_active masque la
|
||
bascule d'astreinte des comptes qui ne la gèrent pas."""
|
||
cg = jour["conge"]
|
||
chips = []
|
||
if "jour" in cg:
|
||
chips.append('<span class="cg-chip st-jour">J</span>')
|
||
if "matin" in cg:
|
||
chips.append('<span class="cg-chip st-ma">MA</span>')
|
||
if "aprem" in cg:
|
||
chips.append('<span class="cg-chip st-am">AM</span>')
|
||
if jour.get("ferie"):
|
||
chips.append('<span class="cg-chip st-ferie">F</span>')
|
||
if as_active and jour.get("astreinte"):
|
||
chips.append('<span class="cg-chip st-as">AS</span>')
|
||
state = "".join(chips) or '<span class="cg-none">—</span>'
|
||
d = _e(date)
|
||
|
||
def opt(url, chip_cls, chip_txt, label, on, title=""):
|
||
t = f' title="{title}"' if title else ""
|
||
return (f'<button hx-post="{url}" hx-swap="none" class="cg-opt{" on" if on else ""}"{t}>'
|
||
f'<span class="cg-opt-chip {chip_cls}">{chip_txt}</span> {label}</button>')
|
||
|
||
# Bascule d'astreinte absente du menu pour les comptes qui ne la gèrent pas
|
||
as_opt = ""
|
||
if as_active:
|
||
as_opt = opt(f"/astreinte/{d}/toggle", "as", "AS", "Jour d'astreinte", jour.get("astreinte"),
|
||
"Jour d'astreinte (18h \u2192 18h, ou 0h \u2192 0h sam/dim/f\u00e9ri\u00e9)")
|
||
|
||
return (
|
||
f'<div id="cg-btn-{d}" class="cg-cell" hx-swap-oob="true">'
|
||
f'<details class="cg-menu">'
|
||
f'<summary class="cg-trigger" title="Congés, jour férié, astreinte">'
|
||
f'<span class="cg-state">{state}</span>'
|
||
f'<span class="cg-caret">▾</span>'
|
||
f'</summary>'
|
||
f'<div class="cg-pop">'
|
||
+ opt(f"/conge/{d}/matin", "ma", "MA", "Congé matin", "matin" in cg)
|
||
+ opt(f"/conge/{d}/aprem", "am", "AM", "Congé après-midi", "aprem" in cg)
|
||
+ opt(f"/conge/{d}/jour", "jour", "J", "Journée entière", "jour" in cg)
|
||
+ '<div class="cg-opt-sep"></div>'
|
||
+ opt(f"/ferie/{d}/toggle", "ferie", "F", "Jour férié", jour.get("ferie"),
|
||
"Jour férié : dû 0, astreinte comptée 0h00 → 0h00")
|
||
+ as_opt
|
||
+ '</div>'
|
||
f'</details>'
|
||
f'</div>'
|
||
)
|
||
|
||
|
||
def _as_event_html(date, ev):
|
||
"""Une ligne d'intervention d'astreinte : début, fin, description (3 colonnes
|
||
de la grille), suppression."""
|
||
return (
|
||
f'<div class="as-event">'
|
||
f'<input class="time-input" type="time" name="astreinte_debut" data-date="{_e(date)}" value="{_e(ev.get("debut") or "")}">'
|
||
f'<input class="time-input" type="time" name="astreinte_fin" data-date="{_e(date)}" value="{_e(ev.get("fin") or "")}">'
|
||
f'<div class="as-desc-cell">'
|
||
f'<input type="text" class="as-desc" name="astreinte_desc" data-date="{_e(date)}" maxlength="200" placeholder="description" value="{_e(ev.get("desc") or "")}">'
|
||
f'<button type="button" class="as-del" title="Supprimer" onclick="this.closest(\'.as-event\').remove()">×</button>'
|
||
f'</div></div>'
|
||
)
|
||
|
||
|
||
def _as_block_html(date, jour):
|
||
"""Bloc astreinte d'une ligne : les heures d'intervention (début, fin,
|
||
description) sont regroupées dans un menu dépliant fermé par défaut ; seuls
|
||
le bouton, l'heure de retour (11h de repos) et le décompte restent visibles.
|
||
Rendu sur les jours qualifiés (astreinte ou lendemain) ; div vide ailleurs."""
|
||
if not jour.get("astreinte_qualifiee"):
|
||
return f'<div id="as-block-{_e(date)}" class="as-block" hx-swap-oob="true"></div>'
|
||
events = jour.get("astreinte_events") or []
|
||
rows = "".join(_as_event_html(date, ev) for ev in events) or _as_event_html(date, {})
|
||
hint = ""
|
||
if jour.get("retour_min") is not None:
|
||
hint = (f'<span class="as-hint" title="Repos après la dernière activité : 11h consécutives (quotidien), et 35h sur 7 jours glissants si le repos hebdomadaire est activé (Réglages). Toute intervention pendant le repos redémarre le décompte ; retour arrondi au quart d\'horloge supérieur : :00, :15, :30 ou :45">'
|
||
f'retour ≥ {_e(jour["retour"])}</span>')
|
||
if jour.get("couvert_min"):
|
||
hint += (f'<span class="as-hint" title="Présence obligatoire couverte par le repos ({_e(jour["couvert_detail"])}) : comptée comme effectuées dans la colonne Travaillé">'
|
||
f'+{_e(jour["couvert_h"])} de présence comptées</span>')
|
||
if jour.get("forfait_min"):
|
||
f_txt = jour.get("forfait_facteur_txt") or "1"
|
||
f_regle = f" ({_e(jour['forfait_regle'])})" if jour.get("forfait_regle") else ""
|
||
hint += (f'<span class="as-hint" title="Forfait du jour AS déclaré (réglé dans Réglages), considéré réalisé en journée du jour : pondéré ×{f_txt}{f_regle} — suivi dans le récapitulatif d'astreinte, hors heures travaillées">'
|
||
f'forfait {_e(jour["forfait_h"])} → {_e(jour["forfait_pondere_h"])} pondérées</span>')
|
||
total = ""
|
||
if jour.get("astreinte_min"):
|
||
total = (f'<span class="as-total" title="Durée réelle des interventions arrondie au quart d\'heure supérieur (2, 5 ou 12 min → 15 ; 22 ou 28 → 30) — compteur séparé, hors heures de semaine">'
|
||
f'astreinte : {_e(jour["astreinte_h"])}</span>')
|
||
nb = len(events)
|
||
label = f'{nb} intervention' + ('s' if nb != 1 else '')
|
||
return (
|
||
f'<div id="as-block-{_e(date)}" class="as-block" data-date="{_e(date)}" data-label="Astreinte" hx-swap-oob="true">'
|
||
f'<div class="as-head">'
|
||
f'<button type="button" class="as-toggle" data-date="{_e(date)}" aria-expanded="false" onclick="asToggle(this)">'
|
||
f'<span class="as-caret">▸</span> {label}</button>'
|
||
f'{hint}{total}'
|
||
f'</div>'
|
||
f'<div class="as-body">'
|
||
f'<div class="as-grid" id="as-events-{_e(date)}">'
|
||
f'<span class="as-th">Début</span><span class="as-th">Fin</span><span class="as-th">Description</span>'
|
||
f'{rows}'
|
||
f'</div>'
|
||
f'<button type="button" class="as-add" onclick="asAddEvent(\'{_e(date)}\')">+ intervention</button>'
|
||
f'</div>'
|
||
f'</div>'
|
||
)
|
||
|
||
|
||
def _htmx_conge_and_soldes(request, date, d, conges, h_jour, user_id):
|
||
iso = d.isocalendar()
|
||
plages = load_plages(user_id)
|
||
jour = _build_jour_ctx(date, conges, h_jour, user_id)
|
||
btns_html = _btns_html(date, jour, as_active=plages.get("astreinte_active", True))
|
||
soldes_ctx = _build_soldes_ctx(iso.year, iso.week, conges, h_jour, user_id)
|
||
soldes_html = templates.get_template("_soldes.html").render({"request": request, **soldes_ctx})
|
||
return HTMLResponse(
|
||
_oob_time_cells(date, jour, plages)
|
||
+ btns_html
|
||
+ _oob_calc(date, jour)
|
||
+ _oob_week_extras(soldes_ctx["result"])
|
||
+ soldes_html
|
||
)
|
||
|
||
|
||
def _htmx_calc_and_soldes(request, date, d, conges, h_jour, user_id):
|
||
iso = d.isocalendar()
|
||
plages = load_plages(user_id)
|
||
jour = _build_jour_ctx(date, conges, h_jour, user_id)
|
||
# Le retour au plus tôt du lendemain dépend des opérations saisies aujourd'hui
|
||
as_blocks = _as_block_html(date, jour)
|
||
lendemain = d + timedelta(days=1)
|
||
lend_iso = lendemain.isocalendar()
|
||
if (lend_iso.year, lend_iso.week) == (iso.year, iso.week):
|
||
lendemain_str = lendemain.strftime("%Y-%m-%d")
|
||
as_blocks += _as_block_html(lendemain_str, _build_jour_ctx(lendemain_str, conges, h_jour, user_id))
|
||
soldes_ctx = _build_soldes_ctx(iso.year, iso.week, conges, h_jour, user_id)
|
||
soldes_html = templates.get_template("_soldes.html").render({"request": request, **soldes_ctx})
|
||
return HTMLResponse(
|
||
_oob_calc(date, jour)
|
||
+ _oob_time_cells(date, jour, plages)
|
||
+ as_blocks
|
||
+ _oob_week_extras(soldes_ctx["result"])
|
||
+ soldes_html
|
||
)
|
||
|
||
|
||
# ── Login / logout ────────────────────────────────────────────────────────────
|
||
|
||
RESET_TOKEN_VALIDITY_HOURS = 24
|
||
|
||
|
||
def _login_ctx(request, step, email="", error=None):
|
||
return {"request": request, "step": step, "email": email, "error": error}
|
||
|
||
|
||
@app.get("/login", response_class=HTMLResponse)
|
||
def login_page(request: Request):
|
||
if _get_user(request):
|
||
return RedirectResponse("/", status_code=302)
|
||
return templates.TemplateResponse("login.html", _login_ctx(request, "email"))
|
||
|
||
|
||
@app.post("/login", response_class=HTMLResponse)
|
||
def login_submit(request: Request, email: str = Form("")):
|
||
# Rate-limit par IP pour empêcher le mail-bombing SMTP et l'énumération.
|
||
ip = _client_ip(request)
|
||
if not _check_login_rate_limit(f"login:{ip}"):
|
||
logger.warning("rate limit /login déclenchée: ip=%r", ip)
|
||
return templates.TemplateResponse("login.html", _login_ctx(
|
||
request, "email", "",
|
||
"Trop de tentatives. Réessaie dans une minute.",
|
||
))
|
||
email = email.strip().lower()
|
||
m = _EMAIL_RE.match(email)
|
||
domain = allowed_email_domain()
|
||
logger.info("POST /login: email=%r domaine_attendu=%r", email, domain)
|
||
if not m or not domain or m.group(2) != domain:
|
||
logger.warning("login refusé (domaine invalide): email=%r", email)
|
||
return templates.TemplateResponse("login.html", _login_ctx(
|
||
request, "email", email,
|
||
f"Seules les adresses @{domain or '?'} sont autorisées.",
|
||
))
|
||
|
||
user_id = m.group(1)
|
||
auth = load_auth(user_id)
|
||
logger.debug("login: user_id=%r compte_existant=%s", user_id, bool(auth.get("password_hash")))
|
||
|
||
if auth.get("password_hash"):
|
||
return templates.TemplateResponse("login.html", _login_ctx(request, "password", email))
|
||
|
||
# No account yet: (re)send a link to set the password
|
||
base_url = _safe_base_url(request)
|
||
if not base_url:
|
||
logger.warning(
|
||
"login refusé (Host non fiable pour générer un lien email): host=%r email=%r",
|
||
request.headers.get("host"), email,
|
||
)
|
||
return templates.TemplateResponse("login.html", _login_ctx(
|
||
request, "email", email,
|
||
"Domaine non autorisé pour générer un lien de connexion.",
|
||
))
|
||
|
||
token = secrets.token_urlsafe(32)
|
||
expires = (datetime.now() + timedelta(hours=RESET_TOKEN_VALIDITY_HOURS)).isoformat()
|
||
save_auth(user_id, email=email, token=token, token_expires=expires)
|
||
logger.info("nouveau token de création de compte pour %r (expire %s)", user_id, expires)
|
||
|
||
link = f"{base_url}/set-password/{token}"
|
||
try:
|
||
send_mail(
|
||
email,
|
||
"Définis ton mot de passe — Décompte Horaire",
|
||
f"Clique sur ce lien pour définir ton mot de passe (valable {RESET_TOKEN_VALIDITY_HOURS}h) :\n\n{link}\n",
|
||
mail_from_address(),
|
||
smtp_config(),
|
||
)
|
||
except Exception:
|
||
logger.exception("échec envoi email d'invitation à %r", email)
|
||
return templates.TemplateResponse("login.html", _login_ctx(
|
||
request, "email", email, "Impossible d'envoyer l'email pour le moment, réessaie plus tard.",
|
||
))
|
||
|
||
return templates.TemplateResponse("login.html", _login_ctx(request, "sent", email))
|
||
|
||
|
||
@app.post("/login/password", response_class=HTMLResponse)
|
||
def login_password(request: Request, email: str = Form(""), password: str = Form("")):
|
||
# Rate-limit par IP+email pour bloquer le brute-force d'un compte précis.
|
||
email = email.strip().lower()
|
||
ip = _client_ip(request)
|
||
if not _check_login_rate_limit(f"pw:{ip}:{email}"):
|
||
logger.warning("rate limit /login/password déclenchée: ip=%r email=%r", ip, email)
|
||
return templates.TemplateResponse("login.html", _login_ctx(
|
||
request, "password", email,
|
||
"Trop de tentatives. Réessaie dans une minute.",
|
||
))
|
||
m = _EMAIL_RE.match(email)
|
||
domain = allowed_email_domain()
|
||
logger.info("POST /login/password: email=%r", email)
|
||
if not m or not domain or m.group(2) != domain:
|
||
logger.warning("login/password refusé (domaine invalide): email=%r", email)
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
user_id = m.group(1)
|
||
ph = load_auth(user_id).get("password_hash")
|
||
if not ph or not bcrypt.checkpw(password.encode(), ph.encode()):
|
||
logger.warning("mot de passe incorrect pour %r", user_id)
|
||
return templates.TemplateResponse("login.html", _login_ctx(
|
||
request, "password", email, "Mot de passe incorrect.",
|
||
))
|
||
|
||
logger.info("connexion réussie: user_id=%r", user_id)
|
||
resp = RedirectResponse("/", status_code=302)
|
||
_set_auth_cookie(resp, user_id, request)
|
||
return resp
|
||
|
||
|
||
def _valid_reset_token(token: str) -> str | None:
|
||
user_id = find_user_by_reset_token(token)
|
||
if not user_id:
|
||
return None
|
||
expires = load_auth(user_id).get("token_expires")
|
||
if not expires or datetime.fromisoformat(expires) < datetime.now():
|
||
return None
|
||
return user_id
|
||
|
||
|
||
@app.get("/set-password/{token}", response_class=HTMLResponse)
|
||
def set_password_page(request: Request, token: str):
|
||
valid = _valid_reset_token(token) is not None
|
||
return templates.TemplateResponse("set_password.html", {
|
||
"request": request, "token": token, "valid": valid, "error": None,
|
||
})
|
||
|
||
|
||
@app.post("/set-password/{token}", response_class=HTMLResponse)
|
||
def set_password_submit(request: Request, token: str, password: str = Form(""), password2: str = Form("")):
|
||
user_id = _valid_reset_token(token)
|
||
if not user_id:
|
||
logger.warning("POST /set-password: token invalide ou expiré")
|
||
return templates.TemplateResponse("set_password.html", {
|
||
"request": request, "token": token, "valid": False, "error": None,
|
||
})
|
||
if len(password) < 8 or password != password2:
|
||
logger.info("POST /set-password: mot de passe rejeté (règles) pour %r", user_id)
|
||
return templates.TemplateResponse("set_password.html", {
|
||
"request": request, "token": token, "valid": True,
|
||
"error": "Les mots de passe ne correspondent pas ou font moins de 8 caractères.",
|
||
})
|
||
|
||
password_hash = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
||
save_auth(user_id, password_hash=password_hash, token=None, token_expires=None)
|
||
logger.info("mot de passe défini pour %r", user_id)
|
||
|
||
resp = RedirectResponse("/", status_code=302)
|
||
_set_auth_cookie(resp, user_id, request)
|
||
return resp
|
||
|
||
|
||
@app.get("/logout")
|
||
def logout(request: Request):
|
||
logger.info("déconnexion: user_id=%r", _get_user(request))
|
||
resp = RedirectResponse("/login", status_code=302)
|
||
resp.delete_cookie("user_id")
|
||
return resp
|
||
|
||
|
||
# ── Main routes ───────────────────────────────────────────────────────────────
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
def index(request: Request):
|
||
if not _get_user(request):
|
||
return RedirectResponse("/login", status_code=302)
|
||
today = datetime.today()
|
||
iso = today.isocalendar()
|
||
return RedirectResponse(f"/semaine/{iso.year}/{iso.week}")
|
||
|
||
|
||
@app.get("/semaine/{year}/{week}", response_class=HTMLResponse)
|
||
def semaine(request: Request, year: int, week: int):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
h_jour = heures_jour_min()
|
||
plages = load_plages(user_id)
|
||
conges = load_conges(user_id)
|
||
soldes_ctx = _build_soldes_ctx(year, week, conges, h_jour, user_id)
|
||
|
||
dates = week_dates(year, week)
|
||
raw = load_week_pointages(year, week, user_id)
|
||
p_by_date = {p["date"]: p for p in raw}
|
||
pointages = [
|
||
p_by_date.get(d, {"date": d, "matin_entree": None, "matin_sortie": None,
|
||
"aprem_entree": None, "aprem_sortie": None})
|
||
for d in dates
|
||
]
|
||
result = compute_week(pointages, conges, h_jour, plages, load_astreintes_effectives(user_id), load_facteurs_astreinte(), load_feries(user_id), _contexte_semaine_precedente(year, week, dates, user_id))
|
||
|
||
prev_d = datetime.strptime(f"{year}-W{week:02d}-1", "%G-W%V-%u").date() - timedelta(weeks=1)
|
||
next_d = datetime.strptime(f"{year}-W{week:02d}-1", "%G-W%V-%u").date() + timedelta(weeks=1)
|
||
all_weeks = list_all_weeks(user_id)
|
||
|
||
return templates.TemplateResponse("semaine.html", {
|
||
"request": request,
|
||
"year": year, "week": week,
|
||
"result": result,
|
||
**soldes_ctx,
|
||
"plages": plages,
|
||
"prev_year": prev_d.isocalendar().year, "prev_week": prev_d.isocalendar().week,
|
||
"next_year": next_d.isocalendar().year, "next_week": next_d.isocalendar().week,
|
||
"jours_fr": JOURS_FR,
|
||
"all_weeks": all_weeks,
|
||
"weeks_by_year": _weeks_by_year(all_weeks),
|
||
"current_user": user_id,
|
||
})
|
||
|
||
|
||
@app.post("/pointage/{date}", response_class=HTMLResponse)
|
||
def save_pointage(
|
||
request: Request, date: str,
|
||
matin_entree: str = Form(""), matin_sortie: str = Form(""),
|
||
aprem_entree: str = Form(""), aprem_sortie: str = Form(""),
|
||
astreinte_debut: list[str] = Form([]), astreinte_fin: list[str] = Form([]),
|
||
astreinte_desc: list[str] = Form([]),
|
||
):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
def clean(v):
|
||
# Refuse tout ce qui n'est pas un HHMM strict : empêche le XSS stocké car la
|
||
# valeur est réinjectée telle quelle dans des f-strings HTML hors templates.
|
||
v = v.strip()
|
||
return v if v and _HHMM_RE.match(v) else None
|
||
|
||
d = datetime.strptime(date, "%Y-%m-%d").date()
|
||
iso = d.isocalendar()
|
||
raw = load_week_pointages(iso.year, iso.week, user_id)
|
||
p_by_date = {p["date"]: p for p in raw}
|
||
# Les événements d'astreinte ne sont affichés (donc postés) que les jours
|
||
# qualifiés (astreinte ou lendemain) ; ailleurs on conserve ceux déjà saisis
|
||
# au lieu de les écraser à vide. Une ligne en cours de saisie (début sans
|
||
# fin, ou l'inverse) est conservée telle quelle : l'auto-enregistrement ne
|
||
# doit pas effacer une saisie à moitié tapée — les calculs (décompte, repos)
|
||
# ignorent les événements incomplets tant que la paire n'est pas complète.
|
||
if jour_astreinte_qualifie(date, set(load_astreintes_effectives(user_id))):
|
||
events = []
|
||
# Les descriptions sont optionnelles : sans elles (POST programmatique),
|
||
# zip s'arrêterait à la liste la plus courte et perdrait les événements.
|
||
for i, (deb, fin) in enumerate(zip(astreinte_debut, astreinte_fin)):
|
||
deb, fin = clean(deb), clean(fin)
|
||
if deb or fin:
|
||
desc = astreinte_desc[i] if i < len(astreinte_desc) else ""
|
||
events.append({"debut": deb, "fin": fin,
|
||
"desc": (desc or "").strip()[:200] or None})
|
||
else:
|
||
events = p_by_date.get(date, {}).get("astreinte_events")
|
||
p_by_date[date] = {
|
||
"date": date,
|
||
"matin_entree": clean(matin_entree), "matin_sortie": clean(matin_sortie),
|
||
"aprem_entree": clean(aprem_entree), "aprem_sortie": clean(aprem_sortie),
|
||
"astreinte_events": events,
|
||
}
|
||
save_week_pointages(iso.year, iso.week, sorted(p_by_date.values(), key=lambda x: x["date"]), user_id)
|
||
conges = load_conges(user_id)
|
||
return _htmx_calc_and_soldes(request, date, d, conges, heures_jour_min(), user_id)
|
||
|
||
|
||
@app.get("/refresh/{date}", response_class=HTMLResponse)
|
||
def refresh_pointage(request: Request, date: str):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
d = datetime.strptime(date, "%Y-%m-%d").date()
|
||
conges = load_conges(user_id)
|
||
return _htmx_calc_and_soldes(request, date, d, conges, heures_jour_min(), user_id)
|
||
|
||
|
||
@app.post("/conge/{date}/{type_conge}", response_class=HTMLResponse)
|
||
def toggle_conge_route(date: str, type_conge: str, request: Request):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
toggle_conge(date, type_conge, user_id)
|
||
d = datetime.strptime(date, "%Y-%m-%d").date()
|
||
if request.headers.get("HX-Request"):
|
||
conges = load_conges(user_id)
|
||
return _htmx_conge_and_soldes(request, date, d, conges, heures_jour_min(), user_id)
|
||
iso = d.isocalendar()
|
||
return RedirectResponse(f"/semaine/{iso.year}/{iso.week}", status_code=303)
|
||
|
||
|
||
@app.post("/astreinte/{date}/toggle", response_class=HTMLResponse)
|
||
def toggle_astreinte_route(date: str, request: Request):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
d = datetime.strptime(date, "%Y-%m-%d").date()
|
||
iso = d.isocalendar()
|
||
# Préférence personnelle : sans gestion des astreintes, la bascule n'existe
|
||
# pas dans l'interface — on refuse aussi les POST directs.
|
||
if not astreinte_active(user_id):
|
||
logger.warning("bascule astreinte refusée (gestion désactivée): user_id=%r date=%r", user_id, date)
|
||
return RedirectResponse(f"/semaine/{iso.year}/{iso.week}", status_code=303)
|
||
|
||
toggle_astreinte(date, user_id)
|
||
if not request.headers.get("HX-Request"):
|
||
return RedirectResponse(f"/semaine/{iso.year}/{iso.week}", status_code=303)
|
||
|
||
conges = load_conges(user_id)
|
||
h_jour = heures_jour_min()
|
||
plages = load_plages(user_id)
|
||
# Marquer une astreinte qualifie ce jour ET son lendemain (la nuit qui suit) :
|
||
# les deux lignes doivent rafraîchir saisie d'opération, avertissements et cibles.
|
||
dates_a_rafraichir = [date]
|
||
lendemain = d + timedelta(days=1)
|
||
lend_iso = lendemain.isocalendar()
|
||
if (lend_iso.year, lend_iso.week) == (iso.year, iso.week):
|
||
dates_a_rafraichir.append(lendemain.strftime("%Y-%m-%d"))
|
||
html = ""
|
||
for date_str in dates_a_rafraichir:
|
||
jour = _build_jour_ctx(date_str, conges, h_jour, user_id)
|
||
html += _oob_time_cells(date_str, jour, plages)
|
||
html += _as_block_html(date_str, jour)
|
||
html += _oob_calc(date_str, jour)
|
||
html += _btns_html(date, _build_jour_ctx(date, conges, h_jour, user_id))
|
||
soldes_ctx = _build_soldes_ctx(iso.year, iso.week, conges, h_jour, user_id)
|
||
soldes_html = templates.get_template("_soldes.html").render({"request": request, **soldes_ctx})
|
||
return HTMLResponse(html + _oob_week_extras(soldes_ctx["result"]) + soldes_html)
|
||
|
||
|
||
@app.post("/ferie/{date}/toggle", response_class=HTMLResponse)
|
||
def toggle_ferie_route(date: str, request: Request):
|
||
"""Marque/démarque un jour férié : dû à 0, aucune présence obligatoire, et
|
||
fenêtre d'astreinte étendue à la journée entière (0h00 → 0h00)."""
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
toggle_ferie(date, user_id)
|
||
d = datetime.strptime(date, "%Y-%m-%d").date()
|
||
iso = d.isocalendar()
|
||
if request.headers.get("HX-Request"):
|
||
conges = load_conges(user_id)
|
||
return _htmx_conge_and_soldes(request, date, d, conges, heures_jour_min(), user_id)
|
||
return RedirectResponse(f"/semaine/{iso.year}/{iso.week}", status_code=303)
|
||
|
||
|
||
@app.get("/stats", response_class=HTMLResponse)
|
||
def stats_page(request: Request):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
stats = compute_all_stats(user_id)
|
||
data_json = json.dumps({
|
||
"time_stats": {
|
||
slot: {
|
||
"mean": s["mean"], "sigma": s["sigma"],
|
||
"mean_hhmm": s["mean_hhmm"], "sigma_min": s["sigma_min"],
|
||
"n": s["n"], "dated_values": s["dated_values"],
|
||
} if s else None
|
||
for slot, s in stats["time_stats"].items()
|
||
},
|
||
"compliance": stats["compliance"],
|
||
"weekly_balances": stats["weekly_balances"],
|
||
"today": stats["today"],
|
||
})
|
||
return templates.TemplateResponse("stats.html", {
|
||
"request": request,
|
||
"stats": stats,
|
||
"data_json": data_json,
|
||
"current_user": user_id,
|
||
})
|
||
|
||
|
||
ODS_MEDIA_TYPE = "application/vnd.oasis.opendocument.spreadsheet"
|
||
|
||
|
||
@app.get("/export/semaine/{year}/{week}")
|
||
def export_semaine(request: Request, year: int, week: int):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
content = export_semaine_ods(year, week, user_id)
|
||
filename = f"pointages-{user_id}-{year}-S{week:02d}.ods"
|
||
logger.info("export ODS semaine: user_id=%r year=%d week=%d", user_id, year, week)
|
||
return Response(
|
||
content=content, media_type=ODS_MEDIA_TYPE,
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|
||
|
||
|
||
@app.get("/export/annee/{year}")
|
||
def export_annee(request: Request, year: int):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
content = export_annee_ods(year, user_id)
|
||
filename = f"pointages-{user_id}-{year}.ods"
|
||
logger.info("export ODS année: user_id=%r year=%d", user_id, year)
|
||
return Response(
|
||
content=content, media_type=ODS_MEDIA_TYPE,
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|
||
|
||
|
||
@app.get("/settings", response_class=HTMLResponse)
|
||
def settings_page(request: Request):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
notif = load_notif_config(user_id)
|
||
plages = load_plages(user_id)
|
||
# Les URLs Tasker sont affichées à l'utilisateur lui-même : moins sensible qu'un
|
||
# email, mais on utilise quand même le helper (fallback sur base_url request si
|
||
# non configurée, car l'utilisateur voit sa propre URL).
|
||
base_url = _safe_base_url(request) or str(request.base_url).rstrip("/")
|
||
return templates.TemplateResponse("settings.html", {
|
||
"request": request,
|
||
"current_user": user_id,
|
||
"ntfy_topic": notif.get("ntfy_topic") or "",
|
||
"ntfy_server": notif.get("ntfy_server") or "https://ntfy.sh",
|
||
"ntfy_token": notif.get("ntfy_token") or "",
|
||
"rappel_debut_h": notif.get("rappel_debut_h", 7),
|
||
"rappel_fin_h": notif.get("rappel_fin_h", 20),
|
||
"arrivee_url": f"{base_url}/presence/{notif['token']}/arrivee",
|
||
"depart_url": f"{base_url}/presence/{notif['token']}/depart",
|
||
"plages": plages,
|
||
"facteurs": load_facteurs_astreinte(),
|
||
"saved": request.query_params.get("saved") == "1",
|
||
"plages_saved": request.query_params.get("plages_saved") == "1",
|
||
"plages_error": request.query_params.get("plages_error") == "1",
|
||
"forfait_saved": request.query_params.get("forfait_saved") == "1",
|
||
"forfait_error": request.query_params.get("forfait_error") == "1",
|
||
"astreinte_saved": request.query_params.get("astreinte_saved") == "1",
|
||
"repos_hebdo_saved": request.query_params.get("repos_hebdo_saved") == "1",
|
||
"facteurs_saved": request.query_params.get("facteurs_saved") == "1",
|
||
"facteurs_error": request.query_params.get("facteurs_error") == "1",
|
||
})
|
||
|
||
|
||
@app.post("/settings", response_class=HTMLResponse)
|
||
def settings_save(
|
||
request: Request, ntfy_topic: str = Form(""), ntfy_server: str = Form(""),
|
||
rappel_debut_h: int = Form(7), rappel_fin_h: int = Form(20),
|
||
ntfy_token: str = Form(""),
|
||
):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
rappel_debut_h = max(0, min(23, rappel_debut_h))
|
||
rappel_fin_h = max(rappel_debut_h + 1, min(24, rappel_fin_h))
|
||
|
||
ntfy_server = ntfy_server.strip()
|
||
if ntfy_server and not is_safe_ntfy_server(ntfy_server):
|
||
logger.warning("réglages notif rejetés (SSRF): user_id=%r ntfy_server=%r", user_id, ntfy_server)
|
||
return RedirectResponse("/settings?ssrf_error=1", status_code=303)
|
||
|
||
logger.info(
|
||
"réglages notif mis à jour: user_id=%r ntfy_topic=%r ntfy_server=%r plage=%sh-%sh token=%s",
|
||
user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h,
|
||
"(défini)" if ntfy_token.strip() else "(vide)",
|
||
)
|
||
save_notif_config(user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h, ntfy_token)
|
||
append_log(
|
||
user_id, "settings", "Réglages ntfy",
|
||
f"topic={ntfy_topic or '(vide)'} server={ntfy_server} plage={rappel_debut_h}h-{rappel_fin_h}h "
|
||
f"jeton={'défini' if ntfy_token.strip() else '(vide)'}",
|
||
)
|
||
return RedirectResponse("/settings?saved=1", status_code=303)
|
||
|
||
|
||
@app.post("/settings/test-ntfy", response_class=HTMLResponse)
|
||
def settings_test_ntfy(request: Request):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
notif = load_notif_config(user_id)
|
||
topic = notif.get("ntfy_topic")
|
||
server = notif.get("ntfy_server", "https://ntfy.sh")
|
||
|
||
if not topic:
|
||
logger.warning("test ntfy: pas de topic configuré pour %r", user_id)
|
||
return HTMLResponse(
|
||
'<div id="ntfy-test-result" class="settings-error">'
|
||
'Renseigne et enregistre un topic ntfy avant de tester.'
|
||
'</div>'
|
||
)
|
||
|
||
logger.info("test ntfy: envoi de test pour %r sur %r/%r", user_id, server, topic)
|
||
ok = send_ntfy(
|
||
server, topic,
|
||
"Ceci est une notification de test — ta configuration ntfy fonctionne.",
|
||
"Test pointeuse", user_id=user_id, token=notif.get("ntfy_token", ""),
|
||
)
|
||
cls = "settings-saved" if ok else "settings-error"
|
||
msg = ("Notification envoyée. Vérifie ton téléphone, elle doit arriver en quelques secondes."
|
||
if ok else
|
||
"Échec de l'envoi. Vérifie le serveur et le topic, et l'optimisation batterie de l'app ntfy.")
|
||
return HTMLResponse(
|
||
f'<div id="ntfy-test-result" class="{cls}">{msg}</div>'
|
||
)
|
||
|
||
|
||
@app.post("/settings/regenerate-presence-token", response_class=HTMLResponse)
|
||
def settings_regenerate_presence_token(request: Request):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
new_token = regenerate_presence_token(user_id)
|
||
logger.warning("token de présence régénéré pour %r", user_id)
|
||
append_log(user_id, "settings", "Token présence régénéré", "Les URLs Tasker ont été invalidées et remplacées.")
|
||
|
||
base_url = _safe_base_url(request) or str(request.base_url).rstrip("/")
|
||
arrivee_url = f"{base_url}/presence/{new_token}/arrivee"
|
||
depart_url = f"{base_url}/presence/{new_token}/depart"
|
||
return HTMLResponse(
|
||
f'<div id="presence-urls" hx-swap-oob="true">'
|
||
f'<div class="settings-urlrow">'
|
||
f'<span class="settings-urllabel">Arrivée</span>'
|
||
f'<code class="settings-url" id="url-arrivee">{arrivee_url}</code>'
|
||
f'<button type="button" class="settings-copy-btn" data-copy-target="url-arrivee">Copier</button>'
|
||
f'</div>'
|
||
f'<div class="settings-urlrow">'
|
||
f'<span class="settings-urllabel">Départ</span>'
|
||
f'<code class="settings-url" id="url-depart">{depart_url}</code>'
|
||
f'<button type="button" class="settings-copy-btn" data-copy-target="url-depart">Copier</button>'
|
||
f'</div>'
|
||
f'<div class="settings-saved">Token régénéré. Mets à jour les URLs dans Automation.</div>'
|
||
f'</div>'
|
||
)
|
||
|
||
|
||
@app.post("/settings/astreinte-active", response_class=HTMLResponse)
|
||
def settings_astreinte_active(request: Request, astreinte_active: str = Form("")):
|
||
"""Préférence personnelle : ce compte gère-t-il les astreintes ? Désactivée,
|
||
tout ce qui touche aux astreintes disparaît pour ce compte (bouton AS,
|
||
interventions, forfait, repos de 11h, récapitulatif, ratios) ; rien n'est
|
||
supprimé, une réactivation retrouve les données telles quelles."""
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
actif = astreinte_active == "1"
|
||
plages = load_plages(user_id)
|
||
plages["astreinte_active"] = actif
|
||
save_plages(user_id, plages)
|
||
logger.info("gestion des astreintes %s: user_id=%r", "activée" if actif else "désactivée", user_id)
|
||
append_log(user_id, "settings", "Gestion des astreintes",
|
||
"activée" if actif else "désactivée (marquages et interventions conservés, interface masquée)")
|
||
return RedirectResponse("/settings?astreinte_saved=1", status_code=303)
|
||
|
||
|
||
@app.post("/settings/repos-hebdo", response_class=HTMLResponse)
|
||
def settings_repos_hebdo(request: Request, repos_hebdo_active: str = Form("")):
|
||
"""Préférence personnelle : respect du repos hebdomadaire de 35h (24h + 11h
|
||
consécutives, une fois par période de 7 jours glissants). Activée, un
|
||
week-end d'astreinte sans trou de 35h repousse l'heure de retour à dernière
|
||
activité + 35h — toute nouvelle intervention avant cette échéance
|
||
redémarre le décompte."""
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
actif = repos_hebdo_active == "1"
|
||
plages = load_plages(user_id)
|
||
plages["repos_hebdo_active"] = actif
|
||
save_plages(user_id, plages)
|
||
logger.info("repos hebdomadaire 35h %s: user_id=%r", "activé" if actif else "désactivé", user_id)
|
||
append_log(user_id, "settings", "Repos hebdomadaire 35h",
|
||
"activé (retour repoussé à dernière activité + 35h quand la semaine glissante n'a pas son repos)"
|
||
if actif else "désactivé (seul le repos quotidien de 11h décale le retour)")
|
||
return RedirectResponse("/settings?repos_hebdo_saved=1", status_code=303)
|
||
|
||
|
||
@app.post("/settings/plages", response_class=HTMLResponse)
|
||
def settings_plages_save(
|
||
request: Request,
|
||
matin_debut: str = Form(""), matin_fin: str = Form(""),
|
||
aprem_debut: str = Form(""), aprem_fin: str = Form(""),
|
||
pause_dejeuner_fin: str = Form(""),
|
||
arrivee_visee: str = Form(""), depart_vise: str = Form(""),
|
||
):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
fields = {
|
||
"matin_debut": matin_debut, "matin_fin": matin_fin,
|
||
"aprem_debut": aprem_debut, "aprem_fin": aprem_fin,
|
||
"pause_dejeuner_fin": pause_dejeuner_fin,
|
||
# Heures visées, optionnelles (vide = comportement par défaut)
|
||
"arrivee_visee": arrivee_visee, "depart_vise": depart_vise,
|
||
}
|
||
optionnels = {"arrivee_visee", "depart_vise"}
|
||
if not all(_HHMM_RE.match(v) for k, v in fields.items() if v or k not in optionnels):
|
||
logger.warning("réglages plages rejetés (format invalide): user_id=%r valeurs=%r", user_id, fields)
|
||
return RedirectResponse("/settings?plages_error=1", status_code=303)
|
||
if not (fields["matin_debut"] < fields["matin_fin"] < fields["aprem_debut"] < fields["aprem_fin"]):
|
||
logger.warning("réglages plages rejetés (ordre invalide): user_id=%r valeurs=%r", user_id, fields)
|
||
return RedirectResponse("/settings?plages_error=1", status_code=303)
|
||
if (fields["arrivee_visee"] and fields["arrivee_visee"] >= fields["matin_fin"]) or \
|
||
(fields["depart_vise"] and fields["depart_vise"] < fields["aprem_fin"]):
|
||
logger.warning("réglages plages rejetés (heures visées incohérentes): user_id=%r valeurs=%r", user_id, fields)
|
||
return RedirectResponse("/settings?plages_error=1", status_code=303)
|
||
|
||
logger.info("réglages plages mis à jour: user_id=%r plages=%r", user_id, fields)
|
||
# Fusionne sur les réglages existants : plages.json porte aussi le forfait
|
||
# d'astreinte et la préférence de gestion des astreintes, à conserver.
|
||
save_plages(user_id, {**load_plages(user_id), **fields})
|
||
append_log(
|
||
user_id, "settings", "Plages horaires",
|
||
f"matin {fields['matin_debut']}→{fields['matin_fin']}, "
|
||
f"aprem {fields['aprem_debut']}→{fields['aprem_fin']}, "
|
||
f"reprise {fields['pause_dejeuner_fin']}"
|
||
+ (f", visées {fields['arrivee_visee']}→{fields['depart_vise']}" if fields['arrivee_visee'] or fields['depart_vise'] else ""),
|
||
)
|
||
return RedirectResponse("/settings?plages_saved=1", status_code=303)
|
||
|
||
|
||
@app.post("/settings/astreinte-forfait", response_class=HTMLResponse)
|
||
def settings_astreinte_forfait(request: Request, astreinte_forfait: str = Form("")):
|
||
"""Durée forfaitaire attribuée à chaque jour d'astreinte déclaré (réglage par
|
||
utilisateur, stocké avec ses plages) : suivie dans le récapitulatif
|
||
d'astreinte, jamais comptée dans les heures travaillées."""
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
val = astreinte_forfait.strip()
|
||
if val and (not _HHMM_RE.match(val) or hhmm_to_minutes(val) > 12 * 60):
|
||
logger.warning("réglages forfait astreinte rejetés (invalide): user_id=%r valeur=%r", user_id, val)
|
||
return RedirectResponse("/settings?forfait_error=1", status_code=303)
|
||
|
||
plages = load_plages(user_id)
|
||
plages["astreinte_forfait"] = val
|
||
save_plages(user_id, plages)
|
||
logger.info("réglages forfait astreinte mis à jour: user_id=%r durée=%r", user_id, val or "(aucune)")
|
||
append_log(user_id, "settings", "Forfait astreinte", f"durée={val or '(aucune)'}")
|
||
return RedirectResponse("/settings?forfait_saved=1", status_code=303)
|
||
|
||
|
||
@app.post("/settings/astreinte-facteurs", response_class=HTMLResponse)
|
||
async def settings_astreinte_facteurs(request: Request):
|
||
"""Règles de pondération des heures d'astreinte, éditables à la main :
|
||
nom, types de jours, fenêtre horaire et ratio de chaque règle. Une règle au
|
||
nom vide est supprimée ; une règle invalide rejette l'enregistrement."""
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
form = await request.form()
|
||
indices = sorted({int(m.group(1)) for k in form.keys()
|
||
if (m := re.fullmatch(r"f_nom_(\d+)", k))})
|
||
regles = []
|
||
for i in indices:
|
||
nom = str(form.get(f"f_nom_{i}") or "").strip()
|
||
if not nom:
|
||
continue
|
||
debut = str(form.get(f"f_debut_{i}") or "")
|
||
fin = str(form.get(f"f_fin_{i}") or "")
|
||
facteur_txt = str(form.get(f"f_facteur_{i}") or "").strip().replace(",", ".")
|
||
jours = [j for j in form.getlist(f"f_jours_{i}") if j in _JOURS_AS_VALIDES]
|
||
try:
|
||
facteur = float(facteur_txt)
|
||
except ValueError:
|
||
facteur = 0.0
|
||
if not _HHMM_RE.match(debut) or not _HHMM_RE.match(fin) or facteur <= 0 or not jours:
|
||
logger.warning("règle astreinte rejetée (invalide): user_id=%r nom=%r debut=%r fin=%r facteur=%r jours=%r",
|
||
user_id, nom, debut, fin, facteur_txt, jours)
|
||
return RedirectResponse("/settings?facteurs_error=1", status_code=303)
|
||
regles.append({"nom": nom, "jours": jours, "debut": debut, "fin": fin, "facteur": facteur})
|
||
|
||
if not regles:
|
||
logger.warning("règles astreinte vides, rien n'est enregistré: user_id=%r", user_id)
|
||
return RedirectResponse("/settings?facteurs_error=1", status_code=303)
|
||
save_facteurs_astreinte(regles)
|
||
logger.info("règles astreinte mises à jour: user_id=%r n=%d", user_id, len(regles))
|
||
append_log(user_id, "settings", "Ratios astreinte", f"{len(regles)} règle(s)")
|
||
return RedirectResponse("/settings?facteurs_saved=1", status_code=303)
|
||
|
||
|
||
@app.post("/presence/{token}/arrivee")
|
||
def presence_arrivee(request: Request, token: str):
|
||
user_id = find_user_by_token(token)
|
||
if not user_id:
|
||
logger.warning("POST /presence/.../arrivee: token inconnu")
|
||
return JSONResponse({"error": "invalid token"}, status_code=404)
|
||
logger.info("présence: arrivée détectée pour %r", user_id)
|
||
save_presence(user_id, present=True, since=datetime.now().isoformat())
|
||
append_log(user_id, "presence", "arrivée", f"Présence détectée par Tasker (IP {_client_ip(request)})")
|
||
return {"ok": True}
|
||
|
||
|
||
@app.post("/presence/{token}/depart")
|
||
def presence_depart(request: Request, token: str):
|
||
user_id = find_user_by_token(token)
|
||
if not user_id:
|
||
logger.warning("POST /presence/.../depart: token inconnu")
|
||
return JSONResponse({"error": "invalid token"}, status_code=404)
|
||
logger.info("présence: départ détecté pour %r", user_id)
|
||
save_presence(user_id, present=False, since=datetime.now().isoformat())
|
||
append_log(user_id, "presence", "départ", f"Départ détecté par Tasker (IP {_client_ip(request)})")
|
||
return {"ok": True}
|
||
|
||
|
||
@app.get("/logs", response_class=HTMLResponse)
|
||
def logs_page(request: Request):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
logs = load_logs(user_id)
|
||
return templates.TemplateResponse("logs.html", {
|
||
"request": request,
|
||
"current_user": user_id,
|
||
"logs": logs,
|
||
"filter": request.query_params.get("filter", "all"),
|
||
})
|
||
|
||
|
||
@app.get("/aide", response_class=HTMLResponse)
|
||
def aide_page(request: Request):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
return templates.TemplateResponse("aide.html", {
|
||
"request": request,
|
||
"current_user": user_id,
|
||
})
|
||
|
||
|