1d8a273ffb
Le bouton "Régénérer le token" de la carte Tasker crée un nouveau token
secret et invalide immédiatement les anciennes URLs, au cas où elles
auraient fuité. La fonction regenerate_presence_token préserve les
autres réglages notif (topic, serveur, jeton d'accès, plage). Au passage
le message de retour du bouton de test ntfy occupe désormais toute la
largeur de la carte au lieu d'être limité à 420px.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
315 lines
11 KiB
Python
315 lines
11 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
|
|
|
|
DATA_DIR = Path("/data")
|
|
CONFIG_FILE = DATA_DIR / "config.json"
|
|
|
|
DEFAULT_CONFIG = {
|
|
"heures_jour": 7.8, # 7h48 = 39h/semaine
|
|
"log_level": "DEBUG", # DEBUG, INFO, WARNING ou ERROR
|
|
"allowed_email_domain": "", # ex: "sdv.fr" — seules ces adresses peuvent se connecter
|
|
"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),
|
|
}
|
|
|
|
# ── 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 heures_jour_min() -> int:
|
|
return int(float(load_config().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)
|
|
)
|
|
|
|
|
|
# ── 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")
|
|
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))
|