leocrm/app/plugins/builtins/automation/execution_engine.py

"""Automation execution engine — evaluates conditions and executes actions."""

from __future__ import annotations

import json
import logging
from datetime import UTC, datetime
from typing import Any

import httpx
from sqlalchemy import select

from app.core.db import get_session_factory
from app.models.notification import Notification

logger = logging.getLogger(__name__)


def _evaluate_condition(condition: dict[str, Any], trigger_data: dict[str, Any]) -> bool:
    """Evaluate a single condition against trigger_data.

    Condition format: {field, operator, value}
    Operators: eq, ne, gt, lt, contains, exists
    """
    field = condition.get("field", "")
    operator = condition.get("operator", "eq")
    expected = condition.get("value")

    actual = trigger_data.get(field)

    if operator == "exists":
        return actual is not None

    if actual is None:
        return False

    if operator == "eq":
        return actual == expected
    elif operator == "ne":
        return actual != expected
    elif operator == "gt":
        try:
            return float(actual) > float(expected)
        except (TypeError, ValueError):
            return str(actual) > str(expected)
    elif operator == "lt":
        try:
            return float(actual) < float(expected)
        except (TypeError, ValueError):
            return str(actual) < str(expected)
    elif operator == "contains":
        return str(expected) in str(actual)
    else:
        logger.warning("Unknown operator: %s", operator)
        return False


def _evaluate_conditions(
    conditions: list[dict[str, Any]],
    trigger_data: dict[str, Any],
    logic: str = "and",
) -> bool:
    """Evaluate a list of conditions with AND/OR logic."""
    if not conditions:
        return True

    results = [_evaluate_condition(c, trigger_data) for c in conditions]

    if logic == "or":
        return any(results)
    else:
        return all(results)


async def _execute_action(
    action: dict[str, Any],
    trigger_data: dict[str, Any],
    tenant_id: str,
    dry_run: bool = False,
) -> dict[str, Any]:
    """Execute a single action. Returns result dict."""
    action_type = action.get("type", "")
    config = action.get("config", {})

    result: dict[str, Any] = {
        "action_type": action_type,
        "status": "skipped" if dry_run else "pending",
        "details": {},
    }

    if dry_run:
        result["status"] = "dry_run"
        result["details"] = {"would_execute": True, "config": config}
        return result

    if action_type == "api_call":
        url = config.get("url", "")
        method = config.get("method", "GET").upper()
        headers = config.get("headers", {})
        body = config.get("body")

        try:
            async with httpx.AsyncClient(timeout=30) as client:
                if method == "GET":
                    resp = await client.get(url, headers=headers)
                elif method == "POST":
                    resp = await client.post(url, headers=headers, json=body)
                elif method == "PUT":
                    resp = await client.put(url, headers=headers, json=body)
                elif method == "PATCH":
                    resp = await client.patch(url, headers=headers, json=body)
                elif method == "DELETE":
                    resp = await client.delete(url, headers=headers)
                else:
                    result["status"] = "failed"
                    result["error"] = f"Unsupported HTTP method: {method}"
                    return result

            result["status"] = "completed" if resp.is_success else "failed"
            result["details"] = {
                "status_code": resp.status_code,
                "response": resp.text[:1000],
            }
        except Exception as e:
            result["status"] = "failed"
            result["error"] = str(e)

    elif action_type == "notification":
        user_id = config.get("user_id")
        title = config.get("title", "Automation notification")
        body = config.get("body", "")
        notif_type = config.get("type", "automation")

        if not user_id:
            result["status"] = "failed"
            result["error"] = "user_id is required for notification action"
            return result

        try:
            from uuid import UUID
            factory = get_session_factory()
            async with factory() as db:
                notification = Notification(
                    tenant_id=UUID(tenant_id),
                    user_id=UUID(user_id),
                    type=notif_type,
                    title=title,
                    body=body,
                )
                db.add(notification)
                await db.flush()

            result["status"] = "completed"
            result["details"] = {"user_id": user_id, "title": title}
        except Exception as e:
            result["status"] = "failed"
            result["error"] = str(e)

    elif action_type == "workflow_start":
        workflow_id = config.get("workflow_id")
        context = config.get("context", {})

        if not workflow_id:
            result["status"] = "failed"
            result["error"] = "workflow_id is required for workflow_start action"
            return result

        try:
            from app.services.workflow_service import create_instance
            from uuid import UUID

            factory = get_session_factory()
            async with factory() as db:
                instance = await create_instance(
                    db=db,
                    tenant_id=UUID(tenant_id),
                    user_id=UUID(config.get("user_id", "00000000-0000-0000-0000-000000000000")),
                    workflow_id=workflow_id,
                    context={**trigger_data, **context},
                )

            result["status"] = "completed"
            result["details"] = {"instance": instance}
        except Exception as e:
            result["status"] = "failed"
            result["error"] = str(e)

    else:
        result["status"] = "failed"
        result["error"] = f"Unknown action type: {action_type}"

    return result


async def run_automation(
    ctx: dict[str, Any],
    automation_id: str,
    trigger_type: str = "manual",
    trigger_data: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """ARQ job function. Loads AutomationDefinition, evaluates conditions,
    executes actions if conditions pass. Saves result to AutomationRun."""
    from app.plugins.builtins.automation.models import AutomationDefinition, AutomationRun

    trigger_data = trigger_data or {}
    factory = get_session_factory()

    # Load automation definition
    async with factory() as db:
        result = await db.execute(
            select(AutomationDefinition).where(AutomationDefinition.id == automation_id)
        )
        automation = result.scalar_one_or_none()

    if automation is None:
        logger.error("AutomationDefinition %s not found", automation_id)
        return {"error": f"AutomationDefinition {automation_id} not found", "status": "failed"}

    if not automation.is_active:
        logger.warning("AutomationDefinition %s is inactive", automation_id)
        return {"error": "Automation is inactive", "status": "skipped"}

    # Evaluate conditions
    conditions = automation.conditions or []
    logic = automation.condition_logic or "and"
    conditions_pass = _evaluate_conditions(conditions, trigger_data, logic)

    result_data: dict[str, Any] = {
        "automation_id": str(automation.id),
        "automation_name": automation.name,
        "trigger_type": trigger_type,
        "status": "running",
        "conditions_pass": conditions_pass,
        "actions": [],
        "error": None,
    }

    if not conditions_pass:
        result_data["status"] = "conditions_not_met"
        logger.info("Automation %s: conditions not met, skipping", automation.id)
    else:
        # Execute actions
        actions = automation.actions or []
        is_dry_run = trigger_data.get("dry_run", False)

        for action in actions:
            action_result = await _execute_action(
                action, trigger_data, str(automation.tenant_id), dry_run=is_dry_run
            )
            result_data["actions"].append(action_result)

        # Determine overall status
        if is_dry_run:
            result_data["status"] = "dry_run"
        elif any(a.get("status") == "failed" for a in result_data["actions"]):
            result_data["status"] = "partial"
        else:
            result_data["status"] = "completed"

    # Save result to AutomationRun
    try:
        async with factory() as db:
            run = AutomationRun(
                tenant_id=automation.tenant_id,
                automation_id=automation.id,
                trigger_type=trigger_type,
                status=result_data["status"],
                input_data=trigger_data,
                output_data=result_data,
                error_message=result_data.get("error"),
            )
            db.add(run)
            await db.flush()
            result_data["run_id"] = str(run.id)
    except Exception as e:
        logger.exception("Failed to save AutomationRun for %s", automation.id)
        result_data["save_error"] = str(e)

    return result_data