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