280 lines
9.1 KiB
Python
280 lines
9.1 KiB
Python
|
"""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
|