core/execution/polars_builder.py

"""
Polars Expression Builder.
Builds programmatic expressions securely without raw eval/exec.
"""
from __future__ import annotations

import polars as pl

def build_polars_expression(operation: str, column: str, params: dict) -> pl.Expr:
    """Build programmatic expressions for increase, decrease, cast_type, and find_replace."""
    if operation == "increase":
        value = params.get("value", 10.0)
        is_percent = params.get("is_percent", True)
        if is_percent:
            return (pl.col(column) * (1 + value / 100)).alias(column)
        else:
            return (pl.col(column) + value).alias(column)
            
    elif operation == "decrease":
        value = params.get("value", 10.0)
        is_percent = params.get("is_percent", True)
        if is_percent:
            return (pl.col(column) * (1 - value / 100)).alias(column)
        else:
            return (pl.col(column) - value).alias(column)
            
    elif operation == "cast_type":
        target = params.get("target_dtype")
        dtype_map = {
            "Int64": pl.Int64,
            "Int32": pl.Int32,
            "Float64": pl.Float64,
            "Float32": pl.Float32,
            "String": pl.String,
            "Boolean": pl.Boolean,
            "Date": pl.Date,
        }
        pl_dtype = dtype_map.get(target)
        if pl_dtype is None:
            raise ValueError(f"Unsupported target dtype: {target}")
        return pl.col(column).cast(pl_dtype).alias(column)
        
    elif operation == "find_replace":
        old_val = params.get("old_value")
        new_val = params.get("new_value")
        is_numeric = params.get("is_numeric", False)
        
        if is_numeric:
            try:
                old_num = float(old_val)
                new_num = float(new_val)
                return pl.when(pl.col(column) == old_num).then(pl.lit(new_num)).otherwise(pl.col(column)).alias(column)
            except (ValueError, TypeError):
                raise ValueError("Values must be numeric for a numeric column replacement.")
        else:
            return (
                pl.when(pl.col(column).cast(pl.String) == str(old_val))
                .then(pl.lit(str(new_val)))
                .otherwise(pl.col(column))
                .alias(column)
            )
            
    else:
        raise ValueError(f"Unsupported operation for programmatic expression building: {operation}")