flowdeck/app/services/formula_engine.py

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

from datetime import date, datetime, timedelta
from typing import Any, Callable


class FormulaEngine:
    """Moteur d'expressions JavaScript-like pour propriétés calculées.

    Usage:
        engine = FormulaEngine()
        result = engine.evaluate("prop('Price') * prop('Qty')", {"Price": 10, "Qty": 5})
        # → 50
    """

    def __init__(self):
        self._functions: dict[str, Callable] = {
            "prop": self._prop,
            "now": self._now,
            "today": self._today,
            "if": self._if,
            "concat": self._concat,
            "round": self._round,
            "contains": self._contains,
            "length": self._length,
            "toNumber": self._to_number,
            "formatDate": self._format_date,
            "dateAdd": self._date_add,
            "dateSubtract": self._date_subtract,
            "replace": self._replace,
            "replaceAll": self._replace_all,
            "join": self._join,
            "empty": self._empty,
            "and": self._and,
            "or": self._or,
            "not": self._not,
            "start": self._start,
            "end": self._end,
        }

    def evaluate(self, expression: str, context: dict[str, Any]) -> Any:
        """Évalue une expression de formule dans un contexte donné.

        Le contexte contient les valeurs des propriétés accessibles via prop('name').
        """
        if not expression or not expression.strip():
            return None
        try:
            return self._eval_expr(expression.strip(), context)
        except Exception:
            return None

    def _eval_expr(self, expr: str, ctx: dict[str, Any]) -> Any:
        """Parse et évalue l'expression."""
        expr = expr.strip()

        # Function call
        if "(" in expr and expr.endswith(")"):
            paren = expr.index("(")
            func_name = expr[:paren].strip()
            args_str = expr[paren + 1:-1]
            args = self._parse_args(args_str)
            evaluated = [self._eval_expr(a.strip(), ctx) for a in args]
            if func_name in self._functions:
                return self._functions[func_name](ctx, *evaluated)
            return None

        # String literal
        if (expr.startswith("'") and expr.endswith("'")) or \
           (expr.startswith('"') and expr.endswith('"')):
            return expr[1:-1]

        # Number
        try:
            return float(expr) if "." in expr else int(expr)
        except ValueError:
            pass

        # Boolean
        if expr.lower() == "true":
            return True
        if expr.lower() == "false":
            return False
        if expr.lower() == "null":
            return None

        # Context lookup
        if expr in ctx:
            return ctx[expr]

        return expr  # raw string

    def _parse_args(self, s: str) -> list[str]:
        """Parse une liste d'arguments en respectant les guillemets."""
        args = []
        current = ""
        depth = 0
        in_quote = False
        quote_char = ""
        for ch in s:
            if ch in ("'", '"') and not in_quote:
                in_quote = True
                quote_char = ch
                current += ch
            elif ch == quote_char and in_quote:
                in_quote = False
                current += ch
            elif ch == "(":
                depth += 1
                current += ch
            elif ch == ")":
                depth -= 1
                current += ch
            elif ch == "," and depth == 0 and not in_quote:
                args.append(current.strip())
                current = ""
            else:
                current += ch
        if current.strip():
            args.append(current.strip())
        return args

    # ── Built-in functions ──

    def _prop(self, ctx: dict, name: str) -> Any:
        return ctx.get(name)

    def _now(self, ctx: dict) -> str:
        return datetime.now().isoformat()

    def _today(self, ctx: dict) -> str:
        return date.today().isoformat()

    def _if(self, ctx: dict, cond: Any, a: Any, b: Any) -> Any:
        return a if cond else b

    def _concat(self, ctx: dict, *args) -> str:
        return "".join(str(a) for a in args)

    def _round(self, ctx: dict, n: Any, d: int = 0) -> float:
        return round(float(n), int(d))

    def _contains(self, ctx: dict, s: Any, sub: str) -> bool:
        return str(sub) in str(s)

    def _length(self, ctx: dict, s: Any) -> int:
        return len(str(s))

    def _to_number(self, ctx: dict, s: Any) -> float:
        try:
            return float(s)
        except (ValueError, TypeError):
            return 0.0

    def _format_date(self, ctx: dict, d: str, fmt: str = "%Y-%m-%d") -> str:
        try:
            dt = datetime.fromisoformat(d.replace("Z", "+00:00"))
            return dt.strftime(fmt)
        except Exception:
            return str(d)

    def _date_add(self, ctx: dict, d: str, n: int, unit: str = "days") -> str:
        try:
            dt = datetime.fromisoformat(d.replace("Z", "+00:00"))
            if unit == "days":
                dt += timedelta(days=int(n))
            elif unit == "hours":
                dt += timedelta(hours=int(n))
            elif unit == "minutes":
                dt += timedelta(minutes=int(n))
            return dt.isoformat()
        except Exception:
            return d

    def _date_subtract(self, ctx: dict, d: str, n: int, unit: str = "days") -> str:
        try:
            dt = datetime.fromisoformat(d.replace("Z", "+00:00"))
            if unit == "days":
                dt -= timedelta(days=int(n))
            elif unit == "hours":
                dt -= timedelta(hours=int(n))
            elif unit == "minutes":
                dt -= timedelta(minutes=int(n))
            return dt.isoformat()
        except Exception:
            return d

    def _replace(self, ctx: dict, s: Any, old: str, new: str) -> str:
        return str(s).replace(old, new)

    def _replace_all(self, ctx: dict, s: Any, old: str, new: str) -> str:
        return str(s).replace(old, new)

    def _join(self, ctx: dict, sep: str, *args) -> str:
        return sep.join(str(a) for a in args)

    def _empty(self, ctx: dict, v: Any) -> bool:
        return v is None or v == "" or v == [] or v == 0

    def _and(self, ctx: dict, *args) -> bool:
        return all(args)

    def _or(self, ctx: dict, *args) -> bool:
        return any(args)

    def _not(self, ctx: dict, v: Any) -> bool:
        return not v

    def _start(self, ctx: dict, s: Any) -> str:
        """Extract start date from a date range (iso1...iso2 or just iso1)."""
        val = str(s)
        if "..." in val:
            return val.split("...")[0]
        return val

    def _end(self, ctx: dict, s: Any) -> str | None:
        """Extract end date from a date range."""
        val = str(s)
        if "..." in val:
            parts = val.split("...")
            return parts[1] if len(parts) > 1 else None
        return None