flowdeck/app/services/rollup_engine.py

"""FlowDeck — Rollup Engine (v1.5.0)."""
from __future__ import annotations

import json
import statistics
from typing import Any, Optional

from app.db import get_conn

ROLLUP_FUNCTIONS = {
    "count": lambda values: len(values),
    "count_values": lambda values: len([v for v in values if v is not None and v != ""]),
    "empty": lambda values: len([v for v in values if v is None or v == ""]),
    "not_empty": lambda values: len([v for v in values if v is not None and v != ""]),
    "sum": lambda values: sum(float(v) for v in values if v is not None),
    "average": lambda values: _safe_avg(values),
    "median": lambda values: _safe_stat(values, statistics.median),
    "min": lambda values: min(_numeric(v for v in values if v is not None), default=None),
    "max": lambda values: max(_numeric(v for v in values if v is not None), default=None),
    "range": lambda values: _safe_range(values),
    "unique": lambda values: list(set(str(v) for v in values if v is not None)),
    "percent_checked": lambda values: _percent_checked(values),
    "percent_per_group": lambda values: _percent_per_group(values),
}


class RollupEngine:
    """Moteur d'agrégation pour les propriétés rollup."""

    def compute(
        self,
        collection_id: int,
        relation_property_id: int,
        target_property_id: int,
        page_id: int,
        rollup_function: str,
    ) -> Any:
        """Calcule l'agrégation d'une propriété via une relation.

        Args:
            collection_id: la collection de la page source
            relation_property_id: la propriété relation (dans cette collection)
            target_property_id: la propriété à agréger (dans la collection liée)
            page_id: la page qui porte la relation
            rollup_function: nom de la fonction (count, sum, avg, etc.)
        """
        if rollup_function not in ROLLUP_FUNCTIONS:
            return None

        with get_conn() as conn:
            # Get the relation property to find the related collection
            rel_prop = conn.execute(
                "SELECT * FROM collection_properties WHERE id=?",
                (relation_property_id,),
            ).fetchone()
            if not rel_prop:
                return None

            related_collection_id = rel_prop["related_collection_id"]
            if not related_collection_id:
                return None

            # Get the target property name
            target_prop = conn.execute(
                "SELECT * FROM collection_properties WHERE id=?",
                (target_property_id,),
            ).fetchone()
            if not target_prop:
                return None

            target_name = target_prop["name"]

            # Get the related page IDs from the source page's property_values_json
            source_page = conn.execute(
                "SELECT property_values_json FROM collection_pages WHERE id=?",
                (page_id,),
            ).fetchone()
            if not source_page:
                return None

            props = json.loads(source_page["property_values_json"])
            related_ids = props.get(str(relation_property_id), [])
            if not isinstance(related_ids, list):
                related_ids = [related_ids] if related_ids else []

            if not related_ids:
                return ROLLUP_FUNCTIONS[rollup_function]([])

            # Get the target values from related pages
            placeholders = ",".join("?" for _ in related_ids)
            rows = conn.execute(
                f"SELECT property_values_json FROM collection_pages WHERE id IN ({placeholders})",
                related_ids,
            ).fetchall()

            values = []
            for row in rows:
                page_props = json.loads(row["property_values_json"])
                val = page_props.get(str(target_property_id))
                if val is None:
                    # Try by name
                    val = page_props.get(target_name)
                values.append(val)

            return ROLLUP_FUNCTIONS[rollup_function](values)


def _safe_avg(values: list) -> Optional[float]:
    nums = [float(v) for v in values if v is not None]
    return sum(nums) / len(nums) if nums else None


def _safe_stat(values: list, fn) -> Optional[float]:
    nums = [float(v) for v in values if v is not None]
    return fn(nums) if nums else None


def _numeric(gen):
    for v in gen:
        try:
            yield float(v)
        except (ValueError, TypeError):
            pass


def _safe_range(values: list) -> Optional[float]:
    nums = list(_numeric(v for v in values if v is not None))
    return max(nums) - min(nums) if len(nums) >= 2 else None


def _percent_checked(values: list) -> Optional[float]:
    """Percentage of true values (for checkbox properties)."""
    if not values:
        return 0.0
    booleans = [v for v in values if isinstance(v, bool)]
    if not booleans:
        return 0.0
    return (sum(1 for v in booleans if v) / len(booleans)) * 100


def _percent_per_group(values: list) -> dict:
    """Percentage distribution across groups."""
    if not values:
        return {}
    counts = {}
    for v in values:
        key = str(v) if v is not None else "Empty"
        counts[key] = counts.get(key, 0) + 1
    total = sum(counts.values())
    return {k: round(v / total * 100, 1) for k, v in counts.items()}