Add webhook trigger condition for automations

Per-automation webhook URL with auto-generated 128-bit hex token.
External services (Home Assistant, IFTTT, curl) can POST to
/api/v1/webhooks/{token} with {"action": "activate"|"deactivate"}
to control automation state — no API key required (token is auth).

Backend: WebhookCondition model, engine state tracking with
immediate evaluation, webhook endpoint, schema/route updates.
Frontend: webhook option in condition editor, URL display with
copy button, card badge, i18n for en/ru/zh.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-01 18:28:31 +03:00
parent 01104acad1
commit aafcf83896
12 changed files with 221 additions and 9 deletions

View File

@@ -15,6 +15,7 @@ from wled_controller.storage.automation import (
MQTTCondition,
SystemIdleCondition,
TimeOfDayCondition,
WebhookCondition,
)
from wled_controller.storage.automation_store import AutomationStore
from wled_controller.storage.scene_preset import ScenePreset
@@ -55,6 +56,8 @@ class AutomationEngine:
# automation_id → datetime of last activation / deactivation
self._last_activated: Dict[str, datetime] = {}
self._last_deactivated: Dict[str, datetime] = {}
# webhook_token → bool (volatile state set by webhook calls)
self._webhook_states: Dict[str, bool] = {}
async def start(self) -> None:
if self._task is not None:
@@ -213,6 +216,8 @@ class AutomationEngine:
return self._evaluate_display_state(condition, display_state)
if isinstance(condition, MQTTCondition):
return self._evaluate_mqtt(condition)
if isinstance(condition, WebhookCondition):
return self._webhook_states.get(condition.token, False)
return False
@staticmethod
@@ -420,6 +425,15 @@ class AutomationEngine:
except Exception as e:
logger.error(f"Immediate automation evaluation error: {e}", exc_info=True)
async def set_webhook_state(self, token: str, active: bool) -> None:
"""Set webhook condition state and trigger immediate evaluation."""
self._webhook_states[token] = active
await self.trigger_evaluate()
def get_webhook_state(self, token: str) -> bool:
"""Read current webhook state (False if never called)."""
return self._webhook_states.get(token, False)
async def deactivate_if_active(self, automation_id: str) -> None:
"""Deactivate an automation immediately (used when disabling/deleting)."""
if automation_id in self._active_automations: