flowdeck/app/services/view_aggregate.py

"""FlowDeck — Aggregation Service (vues Notion §8.3).

Un seul moteur d'agregation cote serveur pour Chart, KPI, widgets de
Dashboard et drilldown : grouper / sous-grouper / mesurer / cumuler /
limiter, avec contrat de sortie normalise.

Mesures : count | sum | average | median | min | max (definitions alignees
sur le Rollup Engine). Groupes : ordre du schema pour select/status/
multi_select (couleurs des options incluses), chronologique pour les dates.
"""
from __future__ import annotations

import json
from typing import Any

from .view_query import (
    CHART_MAX_GROUPS,
    CHART_MAX_SUBGROUPS,
    apply_filters,
    apply_sorts,
    prop_value,
    text_of,
)

NON_GROUPABLE = {"rollup", "button", "unique_id", "files"}
NON_MEASURABLE_AXES = {"rollup", "button", "unique_id", "files"}


def _schema_options(prop: dict) -> list[dict]:
    try:
        opts = json.loads(prop.get("options_json") or "[]")
    except (json.JSONDecodeError, TypeError):
        opts = []
    out = []
    for o in opts or []:
        if isinstance(o, str):
            out.append({"name": o, "color": "gray"})
        elif isinstance(o, dict):
            out.append({"name": o.get("name", ""), "color": o.get("color", "gray")})
    return out


def _measure_values(rows: list[dict], measure_prop: dict | None) -> list[float]:
    vals: list[float] = []
    for p in rows:
        if measure_prop is None:
            vals.append(1.0)
            continue
        v = prop_value(p, measure_prop)
        if v is None or v == "":
            continue
        try:
            vals.append(float(v) if not isinstance(v, bool) else float(v))
        except (ValueError, TypeError):
            continue
    return vals


def _reduce(kind: str, vals: list[float], count: int) -> float:
    if kind == "count":
        return float(count)
    if not vals:
        return 0.0
    if kind == "sum":
        return float(sum(vals))
    if kind in ("average", "avg"):
        return float(sum(vals) / len(vals))
    if kind == "median":
        s = sorted(vals)
        mid = len(s) // 2
        return float(s[mid] if len(s) % 2 else (s[mid - 1] + s[mid]) / 2)
    if kind == "min":
        return float(min(vals))
    if kind == "max":
        return float(max(vals))
    return float(count)


def _date_bucket(value: Any, span_days: int) -> str:
    s = str(value or "")[:10]
    if len(s) < 10:
        return s or "(empty)"
    if span_days <= 62:
        return s
    if span_days <= 371:
        # week bucket: keep ISO week label
        try:
            from datetime import date as _d
            d = _d.fromisoformat(s)
            iso = d.isocalendar()
            return f"{iso.year}-W{iso.week:02d}"
        except ValueError:
            return s[:7]
    return s[:7]


def aggregate(pages: list[dict], properties: list[dict], spec: dict) -> dict:
    """Calcule l'agregat normalise ``{groups, total, truncated, scanned}``."""
    by_name = {p["name"]: p for p in properties}
    by_id = {str(p["id"]): p for p in properties}
    group_prop = by_name.get(spec.get("group_by")) or by_id.get(str(spec.get("group_by", "")))
    sub_prop = None
    if spec.get("sub_group_by"):
        sub_prop = by_name.get(spec["sub_group_by"]) or by_id.get(str(spec["sub_group_by"]))
    measure = spec.get("measure") or {"kind": "count"}
    kind = measure.get("kind", "count")
    measure_prop = None
    if kind != "count" and measure.get("property"):
        measure_prop = by_name.get(measure["property"]) or by_id.get(str(measure["property"]))

    filters = spec.get("filters") or []
    rows = apply_filters(pages, properties, filters, spec.get("filter_conjunction", "and"))
    rows = apply_sorts(rows, properties, spec.get("sorts"))
    scanned = len(rows)

    hidden = set(spec.get("hidden_groups") or [])
    omit_zero = spec.get("omit_zero", True)
    cumulative = bool(spec.get("cumulative"))
    order_cfg = spec.get("order") or {"by": "group_order", "direction": "asc"}
    if isinstance(order_cfg, str):
        order_cfg = {"by": order_cfg, "direction": "asc"}

    # Date grouping needs the span first.
    is_date = group_prop is not None and group_prop.get("prop_type") == "date"
    span_days = 0
    if is_date:
        dates = sorted(text_of(group_prop, prop_value(p, group_prop))[:10]
                       for p in rows if text_of(group_prop, prop_value(p, group_prop)))
        if len(dates) >= 2:
            try:
                from datetime import date as _d
                span_days = (_d.fromisoformat(dates[-1][:10]) - _d.fromisoformat(dates[0][:10])).days
            except ValueError:
                span_days = 0

    buckets: dict[str, list[dict]] = {}
    for p in rows:
        if group_prop is None:
            key = "All"
        elif group_prop.get("prop_type") == "multi_select":
            v = prop_value(p, group_prop)
            keys = v if isinstance(v, list) and v else ["(empty)"]
            for k in keys:
                buckets.setdefault(str(k or "(empty)"), []).append(p)
            continue
        elif is_date:
            key = _date_bucket(prop_value(p, group_prop), span_days)
        else:
            t = text_of(group_prop, prop_value(p, group_prop))
            key = t if t else "(empty)"
        buckets.setdefault(key, []).append(p)

    # Empty schema options (omit_zero=False) for select/status.
    if group_prop is not None and not omit_zero and group_prop.get("prop_type") in ("select", "status"):
        for o in _schema_options(group_prop):
            buckets.setdefault(o["name"], [])

    schema_order = [o["name"] for o in _schema_options(group_prop)] if group_prop else []
    opt_colors = {o["name"]: o.get("color", "gray") for o in _schema_options(group_prop)} if group_prop else {}

    def sort_key(item: tuple[str, list]) -> Any:
        k, members = item
        by = order_cfg.get("by", "group_order")
        if by == "value":
            mvals = _measure_values(members, measure_prop)
            return _reduce(kind, mvals, len(members))
        if by == "label":
            return k.lower()
        if group_prop is not None and group_prop.get("prop_type") == "date":
            return k
        if k in schema_order:
            return (0, schema_order.index(k))
        return (1, k.lower())

    items = sorted(buckets.items(), key=sort_key)
    if str(order_cfg.get("direction", "asc")).lower() in ("desc", "descending"):
        items = list(reversed(items))

    truncated = False
    if len(items) > CHART_MAX_GROUPS:
        items = items[:CHART_MAX_GROUPS]
        truncated = True

    groups = []
    running = 0.0
    for key, members in items:
        if key in hidden:
            continue
        mvals = _measure_values(members, measure_prop)
        value = _reduce(kind, mvals, len(members))
        if cumulative and kind in ("count", "sum"):
            running += value
            value = running
        sub = []
        if sub_prop is not None:
            sub_buckets: dict[str, list] = {}
            for p in members:
                sk = text_of(sub_prop, prop_value(p, sub_prop)) or "(empty)"
                sub_buckets.setdefault(sk, []).append(p)
            for sk, sp in list(sub_buckets.items())[:CHART_MAX_SUBGROUPS]:
                svals = _measure_values(sp, measure_prop)
                sub.append({"key": sk, "label": sk, "value": _reduce(kind, svals, len(sp))})
            sub.sort(key=lambda g: -g["value"])
        groups.append({
            "key": key,
            "label": key,
            "color": opt_colors.get(key, "gray"),
            "value": value,
            "count": len(members),
            "row_ids": [p.get("id") for p in members[:500]],
            "sub": sub,
        })

    total_vals = _measure_values(rows, measure_prop)
    total = _reduce(kind, total_vals, len(rows))
    grand = sum(g["value"] for g in groups) or 1
    for g in groups:
        g["percent"] = g["value"] / grand if grand else 0

    return {"groups": groups, "total": total, "truncated": truncated,
            "scanned": scanned, "measure": kind}