pointeuse-optimisator/app/models.py

"""Flat-file storage: JSON files per week for pointages, one file for congés."""
import json
import secrets
from pathlib import Path

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,
    },
}

# ── 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))


# ── 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"


def load_notif_config(user_id: str) -> dict:
    """Return {"ntfy_topic": str|None, "token": str}, generating a token on first use."""
    f = _notif_file(user_id)
    if f.exists():
        cfg = json.loads(f.read_text())
        if cfg.get("token"):
            return cfg
    else:
        cfg = {"ntfy_topic": None}
    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):
    cfg = load_notif_config(user_id)
    cfg["ntfy_topic"] = ntfy_topic.strip() or None
    _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 _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)
    )


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)