646 lines
26 KiB
Python
646 lines
26 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
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,
|
|
load_plages, save_plages, heures_jour_min,
|
|
)
|
|
from calcul import compute_week, minutes_to_hhmm, hhmm_to_minutes, heures_dues
|
|
from stats import compute_all_stats
|
|
from export import export_semaine_ods, export_annee_ods
|
|
from notifications import reminder_loop
|
|
|
|
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.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"]
|
|
_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$")
|
|
|
|
|
|
# ── 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 _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, load_plages(user_id))}
|
|
|
|
|
|
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>'
|
|
parts = []
|
|
if jour.get("cible_matin"):
|
|
parts.append(f'<span class="cible-num">M {jour["cible_matin"]}</span>')
|
|
if jour.get("cible_aprem"):
|
|
parts.append(f'<span class="cible-num">A {jour["cible_aprem"]}</span>')
|
|
cible_inner = " ".join(parts) if parts 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, plages):
|
|
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"] > plages["matin_debut"]) or
|
|
(jour.get("matin_sortie") and jour["matin_sortie"] < plages["matin_fin"])
|
|
)
|
|
aprem_warn = not aprem_cg and jour["du_min"] > 0 and (
|
|
(jour.get("aprem_entree") and jour["aprem_entree"] > plages["aprem_debut"]) or
|
|
(jour.get("aprem_sortie") and jour["aprem_sortie"] < 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-{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()
|
|
plages = load_plages(user_id)
|
|
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, 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)
|
|
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)
|
|
+ _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()
|
|
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
|
|
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)
|
|
|
|
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:
|
|
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("")):
|
|
email = email.strip().lower()
|
|
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)
|
|
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:
|
|
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)
|
|
resp.set_cookie("user_id", user_id, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30)
|
|
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)
|
|
|
|
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(""),
|
|
):
|
|
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,
|
|
})
|
|
|
|
|
|
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)
|
|
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 "",
|
|
"ntfy_server": notif.get("ntfy_server") or "https://ntfy.sh",
|
|
"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,
|
|
"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",
|
|
})
|
|
|
|
|
|
@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),
|
|
):
|
|
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))
|
|
|
|
logger.info(
|
|
"réglages notif mis à jour: user_id=%r ntfy_topic=%r ntfy_server=%r plage=%sh-%sh",
|
|
user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h,
|
|
)
|
|
save_notif_config(user_id, ntfy_topic, ntfy_server, rappel_debut_h, rappel_fin_h)
|
|
return RedirectResponse("/settings?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(""),
|
|
):
|
|
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,
|
|
}
|
|
if not all(_HHMM_RE.match(v) for v in fields.values()):
|
|
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)
|
|
|
|
logger.info("réglages plages mis à jour: user_id=%r plages=%r", user_id, fields)
|
|
save_plages(user_id, fields)
|
|
return RedirectResponse("/settings?plages_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:
|
|
logger.warning("GET /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())
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/presence/{token}/depart")
|
|
def presence_depart(token: str):
|
|
user_id = find_user_by_token(token)
|
|
if not user_id:
|
|
logger.warning("GET /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())
|
|
return {"ok": True}
|
|
|
|
|