c208a17eec
La saisie astreinte devient une liste d'événements (début, fin, description,
nombre illimité, fin antérieure au début = intervention à cheval sur minuit).
Ces heures ne comptent jamais dans la semaine : elles sont décomptées à part
par pas de 15 minutes, toute tranche entamée comptant en totalité, et rangées
dans des compteurs selon le jour et l'heure (Nuit ×1,5, Journée semaine ×1,
Samedi ×1,5, Dimanche/férié ×2 ; jours fériés français calculés
automatiquement, règles ajustables dans config.json via astreinte_facteurs).
Le détail par événement et les totaux pondérés s'affichent sous le bandeau de
la semaine, et les exports .ods gagnent les colonnes Astreinte brute et
pondérée. Le retour au bureau (fin du dernier événement + 11h de repos) est
désormais dérivé des événements.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
388 lines
14 KiB
Python
388 lines
14 KiB
Python
"""Flat-file storage: JSON files per week for pointages, one file for congés."""
|
||
import json
|
||
import secrets
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from calcul import DEFAULT_PLAGES, DEFAULT_FACTEURS_ASTREINTE
|
||
|
||
DATA_DIR = Path("/data")
|
||
CONFIG_FILE = DATA_DIR / "config.json"
|
||
|
||
DEFAULT_CONFIG = {
|
||
"heures_semaine": 39.0, # heures hebdomadaires visées
|
||
"jours_par_semaine": 5, # jours travaillés dans la semaine — heures_jour en est dérivé
|
||
"log_level": "DEBUG", # DEBUG, INFO, WARNING ou ERROR
|
||
"allowed_email_domain": "", # ex: "sdv.fr" — seules ces adresses peuvent se connecter
|
||
"base_url": "", # ex: "https://pointeuse.sdv.fr" — utilisée pour les liens emails (anti open-redirect via Host header)
|
||
"mail_from": "", # ex: "no-reply@sdv.fr" — expéditeur des emails de connexion
|
||
"smtp": {
|
||
"host": "", # ex: "smtp.sdv.fr"
|
||
"port": 587,
|
||
"user": "",
|
||
"password": "",
|
||
"use_tls": True,
|
||
},
|
||
"plages": dict(DEFAULT_PLAGES),
|
||
"astreinte_facteurs": list(DEFAULT_FACTEURS_ASTREINTE),
|
||
}
|
||
|
||
# ── User path helpers ──────────────────────────────────────────────────────────
|
||
|
||
def _user_root(user_id: str) -> Path:
|
||
return DATA_DIR / "users" / user_id
|
||
|
||
|
||
def _pointages_dir(user_id: str) -> Path:
|
||
return _user_root(user_id) / "pointages"
|
||
|
||
|
||
def _conges_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "conges.json"
|
||
|
||
|
||
def _ensure_user_dirs(user_id: str):
|
||
_pointages_dir(user_id).mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
# ── User list ─────────────────────────────────────────────────────────────────
|
||
|
||
def list_users() -> list[str]:
|
||
users_dir = DATA_DIR / "users"
|
||
if not users_dir.exists():
|
||
return []
|
||
return sorted(p.name for p in users_dir.iterdir() if p.is_dir())
|
||
|
||
|
||
# ── Config (global) ───────────────────────────────────────────────────────────
|
||
|
||
def load_config() -> dict:
|
||
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||
if CONFIG_FILE.exists():
|
||
try:
|
||
return json.loads(CONFIG_FILE.read_text())
|
||
except json.JSONDecodeError as e:
|
||
raise ValueError(f"{CONFIG_FILE} invalide : {e}") from e
|
||
CONFIG_FILE.write_text(json.dumps(DEFAULT_CONFIG, indent=2))
|
||
return DEFAULT_CONFIG.copy()
|
||
|
||
|
||
def save_config(cfg: dict):
|
||
CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||
CONFIG_FILE.write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def load_or_create_secret() -> bytes:
|
||
"""Charge ou crée la clé secrète utilisée pour signer les cookies de session.
|
||
|
||
Persistée hors config pour ne pas être réinitialisée à chaque modification des
|
||
réglages et rester stable entre les redémarrages.
|
||
"""
|
||
secret_file = DATA_DIR / "secret.key"
|
||
secret_file.parent.mkdir(parents=True, exist_ok=True)
|
||
if secret_file.exists():
|
||
try:
|
||
return bytes.fromhex(secret_file.read_text().strip())
|
||
except ValueError:
|
||
pass
|
||
secret = secrets.token_bytes(32)
|
||
secret_file.write_text(secret.hex())
|
||
return secret
|
||
|
||
|
||
def heures_jour_min() -> int:
|
||
"""Minutes attendues par jour, dérivées d'`heures_semaine` et `jours_par_semaine`.
|
||
|
||
Rétrocompat : si `heures_semaine` est absent mais `heures_jour` présent,
|
||
on utilise l'ancienne clé.
|
||
"""
|
||
cfg = load_config()
|
||
jours = int(cfg.get("jours_par_semaine", 5)) or 5
|
||
if "heures_semaine" in cfg:
|
||
return int(float(cfg["heures_semaine"]) * 60 / jours)
|
||
return int(float(cfg.get("heures_jour", 7.8)) * 60)
|
||
|
||
|
||
# ── Pointages ─────────────────────────────────────────────────────────────────
|
||
|
||
def _week_file(user_id: str, year: int, week: int) -> Path:
|
||
return _pointages_dir(user_id) / f"{year}-W{week:02d}.json"
|
||
|
||
|
||
def load_week_pointages(year: int, week: int, user_id: str = "") -> list[dict]:
|
||
f = _week_file(user_id, year, week)
|
||
return json.loads(f.read_text()) if f.exists() else []
|
||
|
||
|
||
def save_week_pointages(year: int, week: int, pointages: list[dict], user_id: str = ""):
|
||
_ensure_user_dirs(user_id)
|
||
_week_file(user_id, year, week).write_text(
|
||
json.dumps(pointages, indent=2, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
def list_all_weeks(user_id: str = "") -> list[tuple[int, int]]:
|
||
pt_dir = _pointages_dir(user_id)
|
||
if not pt_dir.exists():
|
||
return []
|
||
weeks = []
|
||
for f in pt_dir.glob("????-W??.json"):
|
||
parts = f.stem.split("-W")
|
||
if len(parts) == 2:
|
||
weeks.append((int(parts[0]), int(parts[1])))
|
||
return sorted(weeks)
|
||
|
||
|
||
# ── Congés ────────────────────────────────────────────────────────────────────
|
||
|
||
def load_conges(user_id: str = "") -> list[dict]:
|
||
f = _conges_file(user_id)
|
||
return json.loads(f.read_text()) if f.exists() else []
|
||
|
||
|
||
def save_conges(conges: list[dict], user_id: str = ""):
|
||
_ensure_user_dirs(user_id)
|
||
_conges_file(user_id).write_text(
|
||
json.dumps(conges, indent=2, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
# ── Astreintes ────────────────────────────────────────────────────────────────
|
||
|
||
def _astreintes_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "astreintes.json"
|
||
|
||
|
||
def load_astreintes(user_id: str = "") -> list[str]:
|
||
"""Dates marquées comme jours d'astreinte (triées, dédupliquées)."""
|
||
f = _astreintes_file(user_id)
|
||
if not f.exists():
|
||
return []
|
||
try:
|
||
dates = json.loads(f.read_text())
|
||
except json.JSONDecodeError:
|
||
return []
|
||
return sorted(set(d for d in dates if isinstance(d, str)))
|
||
|
||
|
||
def save_astreintes(dates: list[str], user_id: str = ""):
|
||
_ensure_user_dirs(user_id)
|
||
_astreintes_file(user_id).write_text(
|
||
json.dumps(sorted(set(dates)), indent=2, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
def toggle_astreinte(date: str, user_id: str = ""):
|
||
dates = load_astreintes(user_id)
|
||
if date in dates:
|
||
dates.remove(date)
|
||
else:
|
||
dates.append(date)
|
||
save_astreintes(dates, user_id)
|
||
|
||
|
||
def load_facteurs_astreinte() -> list[dict]:
|
||
"""Règles globales de comptage des heures d'astreinte (type de jour × fenêtre
|
||
horaire × facteur) ; défauts cohérents si absentes ou invalides dans la config."""
|
||
facteurs = load_config().get("astreinte_facteurs")
|
||
if facteurs and isinstance(facteurs, list) and all(isinstance(f, dict) for f in facteurs):
|
||
return facteurs
|
||
return DEFAULT_FACTEURS_ASTREINTE
|
||
|
||
|
||
# ── Auth (email + password) ────────────────────────────────────────────────────
|
||
|
||
def _auth_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "auth.json"
|
||
|
||
|
||
def load_auth(user_id: str) -> dict:
|
||
f = _auth_file(user_id)
|
||
if f.exists():
|
||
return json.loads(f.read_text())
|
||
return {"email": None, "password_hash": None, "token": None, "token_expires": None}
|
||
|
||
|
||
def save_auth(user_id: str, **fields):
|
||
auth = load_auth(user_id)
|
||
auth.update(fields)
|
||
_ensure_user_dirs(user_id)
|
||
_auth_file(user_id).write_text(json.dumps(auth, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def find_user_by_reset_token(token: str) -> str | None:
|
||
for user_id in list_users():
|
||
if load_auth(user_id).get("token") == token:
|
||
return user_id
|
||
return None
|
||
|
||
|
||
# ── Notifications (ntfy topic + presence tracking) ────────────────────────────
|
||
|
||
def _notif_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "notif.json"
|
||
|
||
|
||
DEFAULT_NTFY_SERVER = "https://ntfy.sh"
|
||
DEFAULT_RAPPEL_DEBUT_H = 7
|
||
DEFAULT_RAPPEL_FIN_H = 20
|
||
|
||
|
||
def load_notif_config(user_id: str) -> dict:
|
||
"""Return {ntfy_topic, ntfy_server, ntfy_token, rappel_debut_h, rappel_fin_h, token}."""
|
||
f = _notif_file(user_id)
|
||
if f.exists():
|
||
cfg = json.loads(f.read_text())
|
||
cfg.setdefault("ntfy_server", DEFAULT_NTFY_SERVER)
|
||
cfg.setdefault("ntfy_token", "")
|
||
cfg.setdefault("rappel_debut_h", DEFAULT_RAPPEL_DEBUT_H)
|
||
cfg.setdefault("rappel_fin_h", DEFAULT_RAPPEL_FIN_H)
|
||
if cfg.get("token"):
|
||
return cfg
|
||
else:
|
||
cfg = {
|
||
"ntfy_topic": None, "ntfy_server": DEFAULT_NTFY_SERVER,
|
||
"ntfy_token": "",
|
||
"rappel_debut_h": DEFAULT_RAPPEL_DEBUT_H, "rappel_fin_h": DEFAULT_RAPPEL_FIN_H,
|
||
}
|
||
cfg["token"] = secrets.token_hex(16)
|
||
_ensure_user_dirs(user_id)
|
||
f.write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
|
||
return cfg
|
||
|
||
|
||
def save_notif_config(
|
||
user_id: str, ntfy_topic: str, ntfy_server: str = "",
|
||
rappel_debut_h: int = DEFAULT_RAPPEL_DEBUT_H, rappel_fin_h: int = DEFAULT_RAPPEL_FIN_H,
|
||
ntfy_token: str = "",
|
||
):
|
||
cfg = load_notif_config(user_id)
|
||
cfg["ntfy_topic"] = ntfy_topic.strip() or None
|
||
cfg["ntfy_server"] = ntfy_server.strip().rstrip("/") or DEFAULT_NTFY_SERVER
|
||
cfg["ntfy_token"] = ntfy_token.strip()
|
||
cfg["rappel_debut_h"] = rappel_debut_h
|
||
cfg["rappel_fin_h"] = rappel_fin_h
|
||
_ensure_user_dirs(user_id)
|
||
_notif_file(user_id).write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def find_user_by_token(token: str) -> str | None:
|
||
for user_id in list_users():
|
||
f = _notif_file(user_id)
|
||
if f.exists() and json.loads(f.read_text()).get("token") == token:
|
||
return user_id
|
||
return None
|
||
|
||
|
||
def regenerate_presence_token(user_id: str) -> str:
|
||
"""Génère un nouveau token de présence (invalide les URLs Tasker précédentes)."""
|
||
cfg = load_notif_config(user_id)
|
||
cfg["token"] = secrets.token_hex(16)
|
||
_ensure_user_dirs(user_id)
|
||
_notif_file(user_id).write_text(json.dumps(cfg, indent=2, ensure_ascii=False))
|
||
return cfg["token"]
|
||
|
||
|
||
def _presence_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "presence.json"
|
||
|
||
|
||
def load_presence(user_id: str) -> dict:
|
||
f = _presence_file(user_id)
|
||
if f.exists():
|
||
return json.loads(f.read_text())
|
||
return {"present": False, "since": None}
|
||
|
||
|
||
def save_presence(user_id: str, present: bool, since: str | None):
|
||
_ensure_user_dirs(user_id)
|
||
_presence_file(user_id).write_text(
|
||
json.dumps({"present": present, "since": since}, indent=2, ensure_ascii=False)
|
||
)
|
||
|
||
|
||
# ── Plages horaires (par utilisateur) ─────────────────────────────────────────
|
||
|
||
def _plages_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "plages.json"
|
||
|
||
|
||
def load_plages(user_id: str) -> dict:
|
||
"""Return this user's plage config, falling back to the (legacy) global config
|
||
default, then to DEFAULT_PLAGES, for any key not yet customized."""
|
||
base = {**DEFAULT_PLAGES, **load_config().get("plages", {})}
|
||
f = _plages_file(user_id)
|
||
if f.exists():
|
||
try:
|
||
return {**base, **json.loads(f.read_text())}
|
||
except json.JSONDecodeError:
|
||
pass
|
||
return base
|
||
|
||
|
||
def save_plages(user_id: str, plages: dict):
|
||
_ensure_user_dirs(user_id)
|
||
_plages_file(user_id).write_text(json.dumps(plages, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def toggle_conge(date: str, type_conge: str, user_id: str = ""):
|
||
assert type_conge in ("jour", "matin", "aprem")
|
||
conges = load_conges(user_id)
|
||
already = any(c["date"] == date and c["type"] == type_conge for c in conges)
|
||
|
||
if already:
|
||
conges = [c for c in conges if not (c["date"] == date and c["type"] == type_conge)]
|
||
else:
|
||
if type_conge == "jour":
|
||
conges = [c for c in conges if c["date"] != date]
|
||
else:
|
||
conges = [c for c in conges if not (c["date"] == date and c["type"] == "jour")]
|
||
conges.append({"date": date, "type": type_conge})
|
||
|
||
conges.sort(key=lambda c: (c["date"], c["type"]))
|
||
save_conges(conges, user_id)
|
||
|
||
|
||
# ── Journaux (notifications + présence) ───────────────────────────────────────
|
||
|
||
LOG_TYPES = ("notif", "presence", "settings")
|
||
MAX_LOGS = 500
|
||
|
||
|
||
def _logs_file(user_id: str) -> Path:
|
||
return _user_root(user_id) / "logs.json"
|
||
|
||
|
||
def append_log(user_id: str, entry_type: str, event: str, message: str = "", ts: str | None = None):
|
||
"""Ajoute une entrée au journal de l'utilisateur (rotation sur MAX_LOGS)."""
|
||
assert entry_type in LOG_TYPES, f"type de journal invalide: {entry_type!r}"
|
||
f = _logs_file(user_id)
|
||
logs = []
|
||
if f.exists():
|
||
try:
|
||
logs = json.loads(f.read_text())
|
||
except json.JSONDecodeError:
|
||
logs = []
|
||
logs.append({
|
||
"ts": ts or datetime.now().isoformat(timespec="seconds"),
|
||
"type": entry_type,
|
||
"event": event,
|
||
"message": message,
|
||
})
|
||
if len(logs) > MAX_LOGS:
|
||
logs = logs[-MAX_LOGS:]
|
||
_ensure_user_dirs(user_id)
|
||
f.write_text(json.dumps(logs, indent=2, ensure_ascii=False))
|
||
|
||
|
||
def load_logs(user_id: str) -> list[dict]:
|
||
"""Retourne les entrées du journal, des plus récentes aux plus anciennes."""
|
||
f = _logs_file(user_id)
|
||
if not f.exists():
|
||
return []
|
||
try:
|
||
logs = json.loads(f.read_text())
|
||
except json.JSONDecodeError:
|
||
return []
|
||
return list(reversed(logs))
|