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