import asyncio import hashlib import hmac import html import json import logging import re import secrets 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, 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, ) 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, 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.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$") # É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: `.`. 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) -> None: """Set the signed, HttpOnly, SameSite session cookie.""" resp.set_cookie( "user_id", _sign_user_id(user_id), httponly=True, samesite="lax", max_age=60 * 60 * 24 * 30, ) 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 _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/` 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]: 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'{"—" if not jour["travaille_min"] else _e(jour["travaille"])}' 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'{sign}{_e(jour["delta"])}' else: delta_inner = '' return ( f'{trav_inner}' f'{delta_inner}' ) 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 = '0h00' else: cls = "delta-pos" if cm > 0 else "delta-neg" sign = "+" if cm > 0 else "-" cumul_inner = f'{sign}{jour["delta_cumul"]}' else: cumul_inner = '' html += f'{cumul_inner}' parts = [] if jour.get("cible_matin"): parts.append(f'M {_e(jour["cible_matin_entree"])} → {_e(jour["cible_matin"])}') if jour.get("cible_aprem"): parts.append(f'A {_e(jour["cible_aprem_entree"])} → {_e(jour["cible_aprem"])}') cible_inner = " ".join(parts) if parts else '' html += f'{cible_inner}' 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'
' f'
' f'' f'' f'' f'
' ) aprem_html = ( f'
' f'
' f'' f'' f'' f'
' ) 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'
' f'
' f'' f'' f'' f'
' ) 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 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("")): 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) _set_auth_cookie(resp, user_id) 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) 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): # 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} 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) # 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, "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), 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( '
' 'Renseigne et enregistre un topic ntfy avant de tester.' '
' ) 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'
{msg}
' ) @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'
' f'
' f'Arrivée' f'{arrivee_url}' f'
' f'
' f'Départ' f'{depart_url}' f'
' f'
Token régénéré. Mets à jour les URLs dans Tasker.
' f'
' ) @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) save_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.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()) append_log(user_id, "presence", "arrivée", "Présence détectée par Tasker") 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()) append_log(user_id, "presence", "départ", "Départ détecté par Tasker") 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, })