pointeuse-optimisator/app/stats.py

"""Statistics computation across all recorded weeks."""
import statistics as _st
from datetime import datetime, timedelta, date as _date

from models import (
    load_week_pointages, load_conges, list_all_weeks, load_plages, heures_jour_min,
)
from calcul import hhmm_to_minutes, minutes_to_hhmm, heures_travaillees, heures_dues

MOIS_FR = [
    "janvier", "février", "mars", "avril", "mai", "juin",
    "juillet", "août", "septembre", "octobre", "novembre", "décembre",
]


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 _to_hhmm(minutes_float: float) -> str:
    m = round(minutes_float)
    return f"{m // 60:02d}:{m % 60:02d}"


def _day_compliant(p: dict, mc: bool, ac: bool, plages: dict) -> bool | None:
    if mc and ac:
        return None
    matin_ok = mc or (bool(p.get("matin_entree")) and bool(p.get("matin_sortie")))
    aprem_ok = ac or (bool(p.get("aprem_entree")) and bool(p.get("aprem_sortie")))
    if not (matin_ok and aprem_ok):
        return None
    if not mc:
        if p.get("matin_entree") and hhmm_to_minutes(p["matin_entree"]) > hhmm_to_minutes(plages["matin_debut"]):
            return False
        if p.get("matin_sortie") and hhmm_to_minutes(p["matin_sortie"]) < hhmm_to_minutes(plages["matin_fin"]):
            return False
    if not ac:
        if p.get("aprem_entree") and hhmm_to_minutes(p["aprem_entree"]) > hhmm_to_minutes(plages["aprem_debut"]):
            return False
        if p.get("aprem_sortie") and hhmm_to_minutes(p["aprem_sortie"]) < hhmm_to_minutes(plages["aprem_fin"]):
            return False
    return True


def _compliance_over(all_days: list[tuple], cg_set: set, plages: dict) -> dict:
    n = n_ok = 0
    for date, p in all_days:
        mc = (date, "matin") in cg_set or (date, "jour") in cg_set
        ac = (date, "aprem") in cg_set or (date, "jour") in cg_set
        result = _day_compliant(p, mc, ac, plages)
        if result is None:
            continue
        n += 1
        if result:
            n_ok += 1
    return {"pct": round(100 * n_ok / n) if n else None, "n": n, "n_ok": n_ok}


def compute_all_stats(user_id: str = "") -> dict:
    all_weeks = sorted(list_all_weeks(user_id))
    conges = load_conges(user_id)
    h_jour = heures_jour_min()
    plages = load_plages(user_id)
    cg_set = {(c["date"], c["type"]) for c in conges}

    today = _date.today()
    prev_week_date = today - timedelta(weeks=1)
    prev_iso = prev_week_date.isocalendar()
    prev_week_dates = set(_week_dates(prev_iso.year, prev_iso.week))
    current_ym = (today.year, today.month)

    SLOTS = ["matin_entree", "matin_sortie", "aprem_entree", "aprem_sortie"]
    slot_pairs: dict[str, list[tuple[int, str]]] = {s: [] for s in SLOTS}
    comp_n = {s: 0 for s in SLOTS}
    comp_ok = {s: 0 for s in SLOTS}

    TARGETS = {
        "matin_entree": ("le", hhmm_to_minutes(plages["matin_debut"])),
        "matin_sortie": ("ge", hhmm_to_minutes(plages["matin_fin"])),
        "aprem_entree": ("le", hhmm_to_minutes(plages["aprem_debut"])),
        "aprem_sortie": ("ge", hhmm_to_minutes(plages["aprem_fin"])),
    }

    weekly_balances = []
    all_days: list[tuple] = []
    month_days: list[tuple] = []
    prev_week_day_list: list[tuple] = []

    for year, week in all_weeks:
        raw = load_week_pointages(year, week, user_id)
        if not raw:
            continue

        dates = _week_dates(year, week)
        p_by_date = {p["date"]: p for p in raw}
        week_delta = 0
        week_has_incomplete = False

        for date in dates:
            p = p_by_date.get(date, {
                "date": date, "matin_entree": None, "matin_sortie": None,
                "aprem_entree": None, "aprem_sortie": None,
            })
            mc = (date, "matin") in cg_set or (date, "jour") in cg_set
            ac = (date, "aprem") in cg_set or (date, "jour") in cg_set

            matin_filled = mc or (bool(p.get("matin_entree")) and bool(p.get("matin_sortie")))
            aprem_filled = ac or (bool(p.get("aprem_entree")) and bool(p.get("aprem_sortie")))

            all_days.append((date, p))
            d = datetime.strptime(date, "%Y-%m-%d").date()
            if (d.year, d.month) == current_ym:
                month_days.append((date, p))
            if date in prev_week_dates:
                prev_week_day_list.append((date, p))

            if not (matin_filled and aprem_filled):
                week_has_incomplete = True
                continue

            trav = heures_travaillees(p)
            du = heures_dues(date, conges, h_jour)
            week_delta += trav - du

            if not mc:
                for slot in ("matin_entree", "matin_sortie"):
                    if p.get(slot):
                        v = hhmm_to_minutes(p[slot])
                        slot_pairs[slot].append((v, date))
                        op, thr = TARGETS[slot]
                        comp_n[slot] += 1
                        comp_ok[slot] += 1 if (v <= thr if op == "le" else v >= thr) else 0
            if not ac:
                for slot in ("aprem_entree", "aprem_sortie"):
                    if p.get(slot):
                        v = hhmm_to_minutes(p[slot])
                        slot_pairs[slot].append((v, date))
                        op, thr = TARGETS[slot]
                        comp_n[slot] += 1
                        comp_ok[slot] += 1 if (v <= thr if op == "le" else v >= thr) else 0

        weekly_balances.append({
            "year": year, "week": week,
            "date": dates[0],
            "delta_min": week_delta,
            "delta": minutes_to_hhmm(week_delta),
            "partial": week_has_incomplete,
        })

    # Suffix the label with a 2-digit year when weeks span more than one year,
    # otherwise "S24" is ambiguous between e.g. 2024 and 2026.
    spans_multiple_years = len({b["year"] for b in weekly_balances}) > 1
    for b in weekly_balances:
        b["label"] = f"S{b['week']:02d}'{b['year'] % 100:02d}" if spans_multiple_years else f"S{b['week']:02d}"

    time_stats = {}
    for slot, pairs in slot_pairs.items():
        if len(pairs) < 2:
            time_stats[slot] = None
            continue
        vals = [v for v, _ in pairs]
        mean = _st.mean(vals)
        sigma = _st.stdev(vals)
        time_stats[slot] = {
            "mean": mean, "sigma": sigma,
            "mean_hhmm": _to_hhmm(mean),
            "sigma_min": round(sigma),
            "n": len(vals),
            "dated_values": [{"v": v, "d": d} for v, d in pairs],
        }

    compliance = {
        slot: (round(100 * comp_ok[slot] / comp_n[slot]) if comp_n[slot] else None)
        for slot in comp_n
    }
    rates = [v for v in compliance.values() if v is not None]

    n_jours = max(len(slot_pairs["matin_entree"]), len(slot_pairs["aprem_entree"]))
    deltas = [b["delta_min"] for b in weekly_balances]
    avg_delta_min = round(_st.mean(deltas)) if deltas else 0

    comp_globale = _compliance_over(all_days, cg_set, plages)
    comp_mois    = _compliance_over(month_days, cg_set, plages)
    comp_sem_n1  = _compliance_over(prev_week_day_list, cg_set, plages)

    return {
        "time_stats": time_stats,
        "compliance": compliance,
        "plages": plages,
        "weekly_balances": weekly_balances,
        "total_weeks": len(weekly_balances),
        "n_jours": n_jours,
        "avg_delta": minutes_to_hhmm(avg_delta_min),
        "avg_delta_min": avg_delta_min,
        "comp_globale":  comp_globale,
        "comp_mois":     comp_mois,
        "comp_sem_n1":   comp_sem_n1,
        "mois_label":    f"{MOIS_FR[today.month - 1].capitalize()} {today.year}",
        "prev_week_label": f"S{prev_iso.week:02d} / {prev_iso.year}",
        "today": today.strftime("%Y-%m-%d"),
    }