1ab1cf27c1
Notifie l'utilisateur toutes les 5 min s'il est présent dans les locaux sans avoir pointé le matin, ou parti sans avoir dépointé la sortie.
439 lines
17 KiB
Python
439 lines
17 KiB
Python
import asyncio
|
||
import json
|
||
import re
|
||
from datetime import datetime, timedelta
|
||
from itertools import groupby
|
||
|
||
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 models import (
|
||
load_config, load_week_pointages, save_week_pointages, load_conges,
|
||
toggle_conge, list_all_weeks, needs_migration, migrate_legacy_to_user,
|
||
load_notif_config, save_notif_config, find_user_by_token, save_presence,
|
||
)
|
||
from calcul import compute_week, minutes_to_hhmm, hhmm_to_minutes, heures_dues
|
||
from stats import compute_all_stats
|
||
from classement import compute_classement
|
||
from notifications import reminder_loop
|
||
|
||
app = FastAPI(title="Décompte Horaire")
|
||
|
||
|
||
@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"]
|
||
_USERNAME_RE = re.compile(r"^[a-zA-Z0-9_\-]{1,32}$")
|
||
|
||
|
||
# ── Auth helpers ──────────────────────────────────────────────────────────────
|
||
|
||
def _get_user(request: Request) -> str | None:
|
||
"""Extract current user from cookie.
|
||
TODO: replace with SSO token validation when integrating corporate SSO.
|
||
"""
|
||
uid = request.cookies.get("user_id", "").strip()
|
||
return uid if _USERNAME_RE.match(uid) else 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]:
|
||
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(5)]
|
||
|
||
|
||
def heures_jour_min() -> int:
|
||
return int(float(load_config().get("heures_jour", 7.8)) * 60)
|
||
|
||
|
||
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 _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
|
||
]
|
||
return {"result": compute_week(pointages, conges, h_jour)}
|
||
|
||
|
||
def _build_jour_ctx(date, conges, h_jour, user_id):
|
||
from calcul import heures_travaillees
|
||
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}
|
||
p = p_by_date.get(date, {"date": date, "matin_entree": None, "matin_sortie": None,
|
||
"aprem_entree": None, "aprem_sortie": None})
|
||
trav = heures_travaillees(p)
|
||
du = heures_dues(date, conges, h_jour)
|
||
delta = trav - du
|
||
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": {c["type"] for c in conges if c["date"] == date},
|
||
}
|
||
|
||
|
||
def _oob_calc(date, jour):
|
||
trav_inner = f'<span class="calc-num">{"—" if not jour["travaille_min"] else jour["travaille"]}</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}{jour["delta"]}</span>'
|
||
else:
|
||
delta_inner = '<span class="delta-zero">—</span>'
|
||
return (
|
||
f'<span id="calc-{date}-trav" hx-swap-oob="true">{trav_inner}</span>'
|
||
f'<span id="calc-{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-{date}-cumul" hx-swap-oob="true">{cumul_inner}</span>'
|
||
cible = None
|
||
if not jour.get("is_complete"):
|
||
if jour.get("entree_cible"):
|
||
suffix = "→12:00" if "aprem" in jour.get("conge", set()) else "→17:00"
|
||
cible = f'{jour["entree_cible"]}{suffix}'
|
||
elif jour.get("sortie_cible"):
|
||
cible = jour.get("sortie_cible")
|
||
cible_inner = f'<span class="cible-num">{cible}</span>' if cible else '<span class="delta-zero">—</span>'
|
||
html += f'<span id="calc-{date}-cible" hx-swap-oob="true">{cible_inner}</span>'
|
||
return html
|
||
|
||
|
||
def _oob_time_cells(date, jour):
|
||
cg = jour["conge"]
|
||
matin_cg = "matin" in cg or "jour" in cg
|
||
aprem_cg = "aprem" in cg or "jour" in cg
|
||
matin_warn = not matin_cg and jour["du_min"] > 0 and (
|
||
(jour.get("matin_entree") and jour["matin_entree"] > "09:00") or
|
||
(jour.get("matin_sortie") and jour["matin_sortie"] < "12:00")
|
||
)
|
||
aprem_warn = not aprem_cg and jour["du_min"] > 0 and (
|
||
(jour.get("aprem_entree") and jour["aprem_entree"] > "14:00") or
|
||
(jour.get("aprem_sortie") and jour["aprem_sortie"] < "17:00")
|
||
)
|
||
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-{date}" hx-swap-oob="true" class="{matin_cls}">'
|
||
f'<div class="time-pair">'
|
||
f'<input class="time-input" type="time" name="matin_entree" data-date="{date}" value="{jour["matin_entree"] or ""}">'
|
||
f'<span class="time-sep">→</span>'
|
||
f'<input class="time-input" type="time" name="matin_sortie" data-date="{date}" value="{jour["matin_sortie"] or ""}">'
|
||
f'</div></div>'
|
||
)
|
||
aprem_html = (
|
||
f'<div id="cg-aprem-{date}" hx-swap-oob="true" class="{aprem_cls}">'
|
||
f'<div class="time-pair">'
|
||
f'<input class="time-input" type="time" name="aprem_entree" data-date="{date}" value="{jour["aprem_entree"] or ""}">'
|
||
f'<span class="time-sep">→</span>'
|
||
f'<input class="time-input" type="time" name="aprem_sortie" data-date="{date}" value="{jour["aprem_sortie"] or ""}">'
|
||
f'</div></div>'
|
||
)
|
||
return matin_html + aprem_html
|
||
|
||
|
||
def _htmx_conge_and_soldes(request, date, d, conges, h_jour, user_id):
|
||
iso = d.isocalendar()
|
||
jour = _build_jour_ctx(date, conges, h_jour, user_id)
|
||
cg = jour["conge"]
|
||
btn_ma_on = " on" if "matin" in cg else ""
|
||
btn_am_on = " on" if "aprem" in cg else ""
|
||
btn_j_on = " jour-on" if "jour" in cg else ""
|
||
btns_html = (
|
||
f'<div id="cg-btn-{date}" hx-swap-oob="true">'
|
||
f'<div class="conge-btns">'
|
||
f'<button hx-post="/conge/{date}/matin" hx-swap="none" class="btn-cg{btn_ma_on}" title="Congé matin">MA</button>'
|
||
f'<button hx-post="/conge/{date}/aprem" hx-swap="none" class="btn-cg{btn_am_on}" title="Congé après-midi">AM</button>'
|
||
f'<button hx-post="/conge/{date}/jour" hx-swap="none" class="btn-cg btn-jour{btn_j_on}" title="Journée entière">J</button>'
|
||
f'</div></div>'
|
||
)
|
||
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)
|
||
+ 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()
|
||
jour = _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(
|
||
_oob_calc(date, jour)
|
||
+ _oob_time_cells(date, jour)
|
||
+ _oob_week_extras(soldes_ctx["result"])
|
||
+ soldes_html
|
||
)
|
||
|
||
|
||
# ── Login / logout ────────────────────────────────────────────────────────────
|
||
|
||
@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", {"request": request, "error": None})
|
||
|
||
|
||
@app.post("/login", response_class=HTMLResponse)
|
||
def login_submit(request: Request, response: Response, username: str = Form("")):
|
||
username = username.strip().lower()
|
||
if not _USERNAME_RE.match(username):
|
||
return templates.TemplateResponse("login.html", {
|
||
"request": request,
|
||
"error": "Nom d'utilisateur invalide (lettres, chiffres, - et _ uniquement, 1–32 caractères).",
|
||
})
|
||
# First-ever login: migrate legacy flat data to this user's directory
|
||
if needs_migration():
|
||
migrate_legacy_to_user(username)
|
||
|
||
resp = RedirectResponse("/", status_code=302)
|
||
resp.set_cookie("user_id", username, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30)
|
||
return resp
|
||
|
||
|
||
@app.get("/logout")
|
||
def logout():
|
||
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()
|
||
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)
|
||
|
||
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,
|
||
"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(""),
|
||
):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
def clean(v):
|
||
v = v.strip(); return v if 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}
|
||
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),
|
||
}
|
||
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.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,
|
||
})
|
||
|
||
|
||
@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)
|
||
base_url = str(request.base_url).rstrip("/")
|
||
return templates.TemplateResponse("settings.html", {
|
||
"request": request,
|
||
"current_user": user_id,
|
||
"ntfy_topic": notif.get("ntfy_topic") or "",
|
||
"arrivee_url": f"{base_url}/presence/{notif['token']}/arrivee",
|
||
"depart_url": f"{base_url}/presence/{notif['token']}/depart",
|
||
"saved": request.query_params.get("saved") == "1",
|
||
})
|
||
|
||
|
||
@app.post("/settings", response_class=HTMLResponse)
|
||
def settings_save(request: Request, ntfy_topic: str = Form("")):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
save_notif_config(user_id, ntfy_topic)
|
||
return RedirectResponse("/settings?saved=1", status_code=303)
|
||
|
||
|
||
@app.get("/presence/{token}/arrivee")
|
||
def presence_arrivee(token: str):
|
||
user_id = find_user_by_token(token)
|
||
if not user_id:
|
||
return JSONResponse({"error": "invalid token"}, status_code=404)
|
||
save_presence(user_id, present=True, since=datetime.now().isoformat())
|
||
return {"ok": True}
|
||
|
||
|
||
@app.get("/presence/{token}/depart")
|
||
def presence_depart(token: str):
|
||
user_id = find_user_by_token(token)
|
||
if not user_id:
|
||
return JSONResponse({"error": "invalid token"}, status_code=404)
|
||
save_presence(user_id, present=False, since=datetime.now().isoformat())
|
||
return {"ok": True}
|
||
|
||
|
||
@app.get("/classement", response_class=HTMLResponse)
|
||
def classement_page(request: Request):
|
||
user_id = _get_user(request)
|
||
if not user_id:
|
||
return RedirectResponse("/login", status_code=302)
|
||
|
||
ranking = compute_classement()
|
||
return templates.TemplateResponse("classement.html", {
|
||
"request": request,
|
||
"ranking": ranking,
|
||
"current_user": user_id,
|
||
})
|