"""Webhook endpoint for automation triggers. External services call POST /api/v1/webhooks/{token} with a JSON body containing {"action": "activate"} or {"action": "deactivate"} to control automations that have a webhook condition. No API-key auth is required — the secret token itself authenticates the caller. """ from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from wled_controller.api.dependencies import get_automation_engine, get_automation_store from wled_controller.core.automations.automation_engine import AutomationEngine from wled_controller.storage.automation import WebhookCondition from wled_controller.storage.automation_store import AutomationStore from wled_controller.utils import get_logger logger = get_logger(__name__) router = APIRouter() class WebhookPayload(BaseModel): action: str = Field(description="'activate' or 'deactivate'") @router.post( "/api/v1/webhooks/{token}", tags=["Webhooks"], ) async def handle_webhook( token: str, body: WebhookPayload, store: AutomationStore = Depends(get_automation_store), engine: AutomationEngine = Depends(get_automation_engine), ): """Receive a webhook call and set the corresponding condition state.""" if body.action not in ("activate", "deactivate"): raise HTTPException(status_code=400, detail="action must be 'activate' or 'deactivate'") # Find the automation that owns this token for automation in store.get_all_automations(): for condition in automation.conditions: if isinstance(condition, WebhookCondition) and condition.token == token: active = body.action == "activate" await engine.set_webhook_state(token, active) logger.info( "Webhook %s: automation '%s' (%s) → %s", token[:8], automation.name, automation.id, body.action, ) return { "ok": True, "automation_id": automation.id, "automation_name": automation.name, "action": body.action, } raise HTTPException(status_code=404, detail="Webhook token not found")