feat: generic webhook provider with JSONPath payload extraction

Add a new "webhook" provider type that accepts arbitrary HTTP POST payloads,
extracts template variables via user-defined JSONPath mappings, and dispatches
notifications through the existing pipeline. Supports three auth modes
(HMAC-SHA256, Bearer token, none), bounded JSONPath cache, and 1MB payload limit.

Full stack: core provider + event parser, API endpoint, DB migration,
capabilities, seeds, default templates (EN/RU), frontend descriptor, i18n.
This commit is contained in:
2026-03-27 23:51:14 +03:00
parent 307871cae5
commit 616b221c92
38 changed files with 603 additions and 0 deletions
+1
View File
@@ -11,6 +11,7 @@ dependencies = [
"aiohttp>=3.9",
"jinja2>=3.1",
"aiosmtplib>=3.0",
"jsonpath-ng>=1.6",
]
[project.optional-dependencies]
@@ -52,6 +52,9 @@ class EventType(str, Enum):
# Scheduler events
SCHEDULED_MESSAGE = "scheduled_message"
# Generic Webhook events
WEBHOOK_RECEIVED = "webhook_received"
# NUT (Network UPS Tools) events
UPS_ONLINE = "ups_online"
UPS_ON_BATTERY = "ups_on_battery"
@@ -20,6 +20,7 @@ class ServiceProviderType(str, Enum):
SCHEDULER = "scheduler"
NUT = "nut"
GOOGLE_PHOTOS = "google_photos"
WEBHOOK = "webhook"
class ServiceProvider(ABC):
@@ -407,6 +407,43 @@ GOOGLE_PHOTOS_CAPABILITIES = ProviderCapabilities(
],
)
# ---------------------------------------------------------------------------
# Generic Webhook provider capabilities
# ---------------------------------------------------------------------------
WEBHOOK_CAPABILITIES = ProviderCapabilities(
provider_type="webhook",
display_name="Generic Webhook",
webhook_based=True,
supported_filters=[
{
"key": "collections",
"label": "Collections",
"type": "tags",
"placeholder": "value from collection_path",
},
],
notification_slots=[
{"name": "message_webhook_received", "description": "Incoming webhook event"},
],
events=[
{"name": "webhook_received", "description": "Webhook payload received"},
],
command_slots=[
{"name": "start", "description": "/start greeting message"},
{"name": "help", "description": "/help command listing"},
{"name": "status", "description": "/status webhook summary"},
{"name": "rate_limited", "description": "Rate limit warning message"},
{"name": "no_results", "description": "Empty results fallback"},
{"name": "desc_help", "description": "Menu description for /help"},
{"name": "desc_status", "description": "Menu description for /status"},
],
commands=[
{"name": "status", "description": "Show webhook status"},
{"name": "help", "description": "Show commands"},
],
)
# ---------------------------------------------------------------------------
# Registry
# ---------------------------------------------------------------------------
@@ -418,6 +455,7 @@ _REGISTRY: dict[str, ProviderCapabilities] = {
"scheduler": SCHEDULER_CAPABILITIES,
"nut": NUT_CAPABILITIES,
"google_photos": GOOGLE_PHOTOS_CAPABILITIES,
"webhook": WEBHOOK_CAPABILITIES,
}
@@ -0,0 +1,16 @@
"""Generic Webhook service provider implementation."""
from notify_bridge_core.providers.base import ServiceProviderType
from notify_bridge_core.templates.variables import registry
from .event_parser import parse_webhook
from .provider import WEBHOOK_VARIABLES, WebhookServiceProvider
# Register Webhook variables in the global registry
registry.register_provider_variables(ServiceProviderType.WEBHOOK, WEBHOOK_VARIABLES)
__all__ = [
"WebhookServiceProvider",
"WEBHOOK_VARIABLES",
"parse_webhook",
]
@@ -0,0 +1,121 @@
"""Parse generic webhook payloads into ServiceEvent objects using JSONPath mappings."""
from __future__ import annotations
import logging
from collections import OrderedDict
from datetime import datetime, timezone
from typing import Any
from jsonpath_ng import parse as jsonpath_parse
from jsonpath_ng.exceptions import JsonPathParserError
from notify_bridge_core.models.events import EventType, ServiceEvent
from notify_bridge_core.providers.base import ServiceProviderType
_LOGGER = logging.getLogger(__name__)
# Bounded LRU cache for compiled JSONPath expressions (max 1024 entries).
# Expressions come from provider configs so the set is finite in practice,
# but a bound prevents unbounded growth in long-lived processes.
_JSONPATH_CACHE_MAX = 1024
_JSONPATH_CACHE: OrderedDict[str, Any] = OrderedDict()
def _compile_jsonpath(expression: str) -> Any | None:
"""Compile a JSONPath expression, returning cached result or None on error."""
if expression in _JSONPATH_CACHE:
_JSONPATH_CACHE.move_to_end(expression)
return _JSONPATH_CACHE[expression]
try:
compiled = jsonpath_parse(expression)
except (JsonPathParserError, Exception) as exc:
_LOGGER.warning("Invalid JSONPath expression '%s': %s", expression, exc)
compiled = None
_JSONPATH_CACHE[expression] = compiled
if len(_JSONPATH_CACHE) > _JSONPATH_CACHE_MAX:
_JSONPATH_CACHE.popitem(last=False)
return compiled
def _extract_value(compiled_path: Any, data: dict[str, Any], default: Any = "") -> Any:
"""Extract a single value from data using a compiled JSONPath expression."""
if compiled_path is None:
return default
matches = compiled_path.find(data)
if not matches:
return default
return matches[0].value
def parse_webhook(
payload: dict[str, Any],
provider_name: str,
config: dict[str, Any],
*,
headers: dict[str, str] | None = None,
) -> ServiceEvent | None:
"""Parse a generic webhook payload into a ServiceEvent.
Uses JSONPath mappings from the provider config to extract template variables.
The full payload is always available as ``raw_payload`` in the extra dict.
Args:
payload: Parsed JSON body of the webhook.
provider_name: Display name of the ServiceProvider instance.
config: Provider configuration containing payload_mappings, etc.
headers: Optional HTTP headers (made available for event_type_path extraction).
Returns:
A ServiceEvent, or None if parsing fails critically.
"""
# Build a combined data dict so JSONPath can reference headers too
data: dict[str, Any] = {**payload}
if headers:
data["headers"] = headers
# Resolve event type from config path or default
event_type_raw = "webhook_received"
event_type_path = config.get("event_type_path", "")
if event_type_path:
compiled = _compile_jsonpath(event_type_path)
extracted = _extract_value(compiled, data, "webhook_received")
if isinstance(extracted, str) and extracted:
event_type_raw = extracted
# Resolve collection from config path or default
collection_id = ""
collection_name = ""
collection_path = config.get("collection_path", "")
if collection_path:
compiled = _compile_jsonpath(collection_path)
extracted = _extract_value(compiled, data, "")
if isinstance(extracted, str):
collection_id = extracted
collection_name = extracted
# Apply payload_mappings to extract named variables
extra: dict[str, Any] = {}
mappings = config.get("payload_mappings", [])
for mapping in mappings:
variable = mapping.get("variable", "")
jsonpath_expr = mapping.get("jsonpath", "")
default = mapping.get("default", "")
if not variable or not jsonpath_expr:
continue
compiled = _compile_jsonpath(jsonpath_expr)
extra[variable] = _extract_value(compiled, data, default)
# Always include raw payload and event type raw
extra["raw_payload"] = payload
extra["event_type_raw"] = event_type_raw
return ServiceEvent(
event_type=EventType.WEBHOOK_RECEIVED,
provider_type=ServiceProviderType.WEBHOOK,
provider_name=provider_name,
collection_id=collection_id,
collection_name=collection_name,
timestamp=datetime.now(timezone.utc),
extra=extra,
)
@@ -0,0 +1,107 @@
"""Generic Webhook service provider — receives arbitrary HTTP payloads."""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from notify_bridge_core.providers.base import ServiceProvider, ServiceProviderType
from notify_bridge_core.templates.variables import TemplateVariableDefinition
if TYPE_CHECKING:
from notify_bridge_core.models.events import ServiceEvent
# Provider-specific variables (user-defined variables from payload_mappings
# are dynamic and documented separately in the template editor).
WEBHOOK_VARIABLES: list[TemplateVariableDefinition] = [
TemplateVariableDefinition(
name="raw_payload",
type="dict",
description="Full incoming JSON payload as a dict (use raw_payload.field or raw_payload | tojson)",
example='{"action": "opened", "issue": {"title": "Bug"}}',
provider_type=ServiceProviderType.WEBHOOK,
),
TemplateVariableDefinition(
name="event_type_raw",
type="string",
description="Raw event type extracted from payload via event_type_path (or 'webhook_received')",
example="webhook_received",
provider_type=ServiceProviderType.WEBHOOK,
),
TemplateVariableDefinition(
name="source_ip",
type="string",
description="IP address of the webhook sender",
example="192.168.1.100",
provider_type=ServiceProviderType.WEBHOOK,
),
]
class WebhookServiceProvider(ServiceProvider):
"""Generic webhook provider — no polling, receives events via HTTP POST."""
provider_type = ServiceProviderType.WEBHOOK
def __init__(self, name: str = "Webhook") -> None:
self._name = name
async def connect(self) -> bool:
return True
async def disconnect(self) -> None:
pass
async def poll(
self,
collection_ids: list[str],
tracker_state: dict[str, Any],
) -> tuple[list[ServiceEvent], dict[str, Any]]:
# Webhook-based provider — never polled
return [], tracker_state
def get_available_variables(self) -> list[TemplateVariableDefinition]:
return WEBHOOK_VARIABLES
def get_provider_config_schema(self) -> dict[str, Any]:
return {
"type": "object",
"properties": {
"auth_mode": {
"type": "string",
"enum": ["hmac_sha256", "bearer_token", "none"],
"default": "none",
"description": "Authentication mode for incoming webhooks",
},
"webhook_secret": {
"type": "string",
"description": "Secret for HMAC-SHA256 or Bearer token authentication",
},
"payload_mappings": {
"type": "array",
"items": {
"type": "object",
"properties": {
"variable": {"type": "string"},
"jsonpath": {"type": "string"},
"default": {"type": "string"},
},
"required": ["variable", "jsonpath"],
},
"description": "JSONPath rules to extract template variables from payload",
},
"event_type_path": {
"type": "string",
"description": "JSONPath to extract event type from payload",
},
"collection_path": {
"type": "string",
"description": "JSONPath to extract collection identifier from payload",
},
},
}
async def list_collections(self) -> list[dict[str, Any]]:
return []
async def test_connection(self) -> dict[str, Any]:
return {"ok": True, "message": "Webhook provider is ready to receive events"}
@@ -0,0 +1 @@
Show available commands
@@ -0,0 +1,4 @@
📋 <b>Available commands:</b>
{%- for cmd in commands %}
/{{ cmd.name }} — {{ cmd.description }}
{%- endfor %}
@@ -0,0 +1 @@
⏳ Please wait {{ wait }}s before using this command again.
@@ -0,0 +1,2 @@
👋 Hi! I'm your Notify Bridge bot for <b>Generic Webhook</b>.
Use /help to see available commands.
@@ -0,0 +1,3 @@
📊 <b>Webhook Status</b>
Trackers active: {{ trackers_active }}
Last event: {{ last_event }}
@@ -47,6 +47,13 @@ PROVIDER_COMMAND_SLOTS: dict[str, list[str]] = {
# Description slots
"desc_help", "desc_status", "desc_devices", "desc_battery",
],
"webhook": [
# Response templates
"start", "help", "status",
"rate_limited", "no_results",
# Description slots
"desc_help", "desc_status",
],
"google_photos": [
# Response templates
"start", "help", "status", "albums", "latest", "search", "random",
@@ -0,0 +1 @@
Показать доступные команды
@@ -0,0 +1 @@
Показать статус вебхука
@@ -0,0 +1,4 @@
📋 <b>Доступные команды:</b>
{%- for cmd in commands %}
/{{ cmd.name }} — {{ cmd.description }}
{%- endfor %}
@@ -0,0 +1 @@
Результатов не найдено.
@@ -0,0 +1 @@
⏳ Подождите {{ wait }} сек. перед повторным использованием команды.
@@ -0,0 +1,2 @@
👋 Привет! Я ваш бот Notify Bridge для <b>вебхуков</b>.
Используйте /help для списка команд.
@@ -0,0 +1,3 @@
📊 <b>Статус вебхука</b>
Активных трекеров: {{ trackers_active }}
Последнее событие: {{ last_event }}
@@ -149,6 +149,12 @@ def build_template_context(
if event.new_name:
ctx.setdefault("new_album_name", event.new_name)
# Provider-specific defaults for Generic Webhook
if event.provider_type.value == "webhook":
ctx.setdefault("raw_payload", event.extra.get("raw_payload", {}))
ctx.setdefault("event_type_raw", event.extra.get("event_type_raw", "webhook_received"))
ctx.setdefault("source_ip", event.extra.get("source_ip", ""))
# Provider-specific defaults for Scheduler
if event.provider_type.value == "scheduler":
ctx.setdefault("schedule_name", event.collection_name)
@@ -0,0 +1,8 @@
📩 <b>Webhook received</b> — {{ service_name }}
{%- if event_type_raw and event_type_raw != 'webhook_received' %}
Event: <code>{{ event_type_raw }}</code>
{%- endif %}
{%- if collection_name %}
Source: {{ collection_name }}
{%- endif %}
<pre>{{ raw_payload | tojson(indent=2) | truncate(2000) }}</pre>
@@ -60,6 +60,9 @@ PROVIDER_SLOT_FILE_MAP: dict[str, dict[str, str]] = {
"message_collection_deleted": "gp_collection_deleted.jinja2",
"message_sharing_changed": "gp_sharing_changed.jinja2",
},
"webhook": {
"message_webhook_received": "webhook_received.jinja2",
},
"nut": {
"message_ups_online": "nut_ups_online.jinja2",
"message_ups_on_battery": "nut_ups_on_battery.jinja2",
@@ -0,0 +1,8 @@
📩 <b>Вебхук получен</b> — {{ service_name }}
{%- if event_type_raw and event_type_raw != 'webhook_received' %}
Событие: <code>{{ event_type_raw }}</code>
{%- endif %}
{%- if collection_name %}
Источник: {{ collection_name }}
{%- endif %}
<pre>{{ raw_payload | tojson(indent=2) | truncate(2000) }}</pre>
@@ -26,6 +26,7 @@ def validate_template(
"target_type", "has_videos", "has_photos",
"has_oversized_videos", "max_video_size", "max_video_size_mb",
"added_assets", "assets", "albums",
"raw_payload", "event_type_raw", "source_ip",
}
allowed = available | runtime_vars