Files
pointeuse-optimisator/app/main.py
Antoine e6a523f76f Envoie les emails via un relais SMTP configurable plutôt qu'en direct-to-MX
Ajoute le bloc "smtp" (host, port, identifiants, TLS) dans config.json,
plus fiable et plus simple à opérer qu'une livraison directe au MX du
destinataire. data/config.json n'est plus versionné (secrets), remplacé
par data/config.example.json.
2026-07-17 21:46:52 +02:00

543 lines
21 KiB
Python

import asyncio
import json
import re
import secrets
from datetime import datetime, timedelta
from itertools import groupby
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,
load_notif_config, save_notif_config, find_user_by_token, save_presence,
load_auth, save_auth, find_user_by_reset_token,
)
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,64}$")
_EMAIL_RE = re.compile(r"^([a-zA-Z0-9_.\-]{1,64})@([a-zA-Z0-9.\-]{1,255})$")
# ── Auth helpers ──────────────────────────────────────────────────────────────
def _get_user(request: Request) -> str | None:
"""Extract current user from cookie."""
uid = request.cookies.get("user_id", "").strip()
return uid if _USERNAME_RE.match(uid) else None
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", {})
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 ────────────────────────────────────────────────────────────
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("")):
email = email.strip().lower()
m = _EMAIL_RE.match(email)
domain = allowed_email_domain()
if not m or not domain or m.group(2) != domain:
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)
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
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)
base_url = str(request.base_url).rstrip("/")
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:
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("")):
email = email.strip().lower()
m = _EMAIL_RE.match(email)
domain = allowed_email_domain()
if not m or not domain or m.group(2) != domain:
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()):
return templates.TemplateResponse("login.html", _login_ctx(
request, "password", email, "Mot de passe incorrect.",
))
resp = RedirectResponse("/", status_code=302)
resp.set_cookie("user_id", user_id, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30)
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:
return templates.TemplateResponse("set_password.html", {
"request": request, "token": token, "valid": False, "error": None,
})
if len(password) < 8 or password != password2:
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)
resp = RedirectResponse("/", status_code=302)
resp.set_cookie("user_id", user_id, 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,
})