feat: Receiver OOP hierarchy with per-receiver locale resolution
- Introduce Receiver base class + typed subclasses (TelegramReceiver, WebhookReceiver, EmailReceiver, etc.) in core/notifications/receiver.py - Dispatcher uses typed Receiver objects instead of raw dicts, with per-receiver locale-aware template rendering - load_link_data resolves locale from TelegramChat.language_override at load time: TargetReceiver.locale || chat.language_override || chat.language_code - Add language_override field to TelegramChat (separate from auto-detected language_code), with per-chat commands toggle and command dispatch using override language - Add locale field to TargetReceiver for explicit per-receiver overrides
This commit is contained in:
@@ -12,6 +12,16 @@ from notify_bridge_core.models.events import ServiceEvent
|
||||
from notify_bridge_core.templates.context import build_template_context
|
||||
from notify_bridge_core.templates.renderer import render_template
|
||||
|
||||
from .receiver import (
|
||||
Receiver,
|
||||
TelegramReceiver,
|
||||
WebhookReceiver,
|
||||
EmailReceiver,
|
||||
DiscordReceiver,
|
||||
SlackReceiver,
|
||||
NtfyReceiver,
|
||||
MatrixReceiver,
|
||||
)
|
||||
from .telegram.cache import TelegramFileCache
|
||||
from .telegram.client import TelegramClient
|
||||
from .webhook.client import WebhookClient
|
||||
@@ -28,14 +38,13 @@ class TargetConfig:
|
||||
type: str # "telegram", "webhook", "email", "discord", "slack", "ntfy", "matrix"
|
||||
config: dict[str, Any] # target-level config (bot_token, settings, etc.)
|
||||
template_slots: dict[str, dict[str, str]] | None = None # event_type -> {locale -> template}
|
||||
locale: str = "en" # preferred locale for template resolution
|
||||
locale: str = "en" # default locale for template resolution
|
||||
date_format: str = "%d.%m.%Y, %H:%M UTC"
|
||||
date_only_format: str = "%d.%m.%Y"
|
||||
provider_api_key: str | None = None # API key for downloading assets from provider
|
||||
provider_internal_url: str | None = None # Internal provider URL for API key scoping
|
||||
provider_external_url: str | None = None # External domain for API key scoping
|
||||
# Broadcast receivers — if non-empty, sends to each receiver instead of config
|
||||
receivers: list[dict[str, Any]] = field(default_factory=list)
|
||||
receivers: list[Receiver] = field(default_factory=list)
|
||||
|
||||
|
||||
class NotificationDispatcher:
|
||||
@@ -69,24 +78,43 @@ class NotificationDispatcher:
|
||||
results.append({"success": False, "error": str(e)})
|
||||
return results
|
||||
|
||||
async def _send_to_target(
|
||||
self, event: ServiceEvent, target: TargetConfig
|
||||
) -> dict[str, Any]:
|
||||
"""Send event to a single target (potentially multiple receivers)."""
|
||||
# Select template with locale fallback
|
||||
def _resolve_template(
|
||||
self, event: ServiceEvent, target: TargetConfig, locale: str,
|
||||
) -> str:
|
||||
"""Resolve template string for an event, with locale fallback."""
|
||||
template_str = DEFAULT_TEMPLATE
|
||||
if target.template_slots:
|
||||
locale_map = target.template_slots.get(event.event_type.value)
|
||||
if locale_map:
|
||||
template_str = locale_map.get(target.locale) or locale_map.get("en") or template_str
|
||||
template_str = locale_map.get(locale) or locale_map.get("en") or template_str
|
||||
return template_str
|
||||
|
||||
# Build context and render ONCE
|
||||
def _render_message(
|
||||
self, event: ServiceEvent, target: TargetConfig, locale: str,
|
||||
) -> str:
|
||||
"""Resolve template and render message for a given locale."""
|
||||
template_str = self._resolve_template(event, target, locale)
|
||||
ctx = build_template_context(
|
||||
event, target_type=target.type,
|
||||
date_format=target.date_format,
|
||||
date_only_format=target.date_only_format,
|
||||
)
|
||||
message = render_template(template_str, ctx)
|
||||
return render_template(template_str, ctx)
|
||||
|
||||
def _message_for_receiver(
|
||||
self, receiver: Receiver, default_message: str,
|
||||
event: ServiceEvent, target: TargetConfig,
|
||||
) -> str:
|
||||
"""Return per-receiver message, re-rendering if receiver has a different locale."""
|
||||
if receiver.locale and receiver.locale != target.locale:
|
||||
return self._render_message(event, target, receiver.locale)
|
||||
return default_message
|
||||
|
||||
async def _send_to_target(
|
||||
self, event: ServiceEvent, target: TargetConfig
|
||||
) -> dict[str, Any]:
|
||||
"""Send event to a single target (potentially multiple receivers)."""
|
||||
default_message = self._render_message(event, target, target.locale)
|
||||
|
||||
send_method = {
|
||||
"telegram": self._send_telegram,
|
||||
@@ -98,11 +126,11 @@ class NotificationDispatcher:
|
||||
"matrix": self._send_matrix,
|
||||
}.get(target.type)
|
||||
if send_method:
|
||||
return await send_method(target, message, event)
|
||||
return await send_method(target, default_message, event)
|
||||
return {"success": False, "error": f"Unknown target type: {target.type}"}
|
||||
|
||||
async def _send_telegram(
|
||||
self, target: TargetConfig, message: str, event: ServiceEvent
|
||||
self, target: TargetConfig, default_message: str, event: ServiceEvent
|
||||
) -> dict[str, Any]:
|
||||
bot_token = target.config.get("bot_token")
|
||||
disable_preview = target.config.get("disable_url_preview", False)
|
||||
@@ -120,7 +148,6 @@ class NotificationDispatcher:
|
||||
|
||||
if not target.receivers:
|
||||
return {"success": False, "error": "No receivers configured"}
|
||||
receivers = target.receivers
|
||||
|
||||
# Prepare assets list once (shared across receivers)
|
||||
provider_urls = []
|
||||
@@ -146,28 +173,27 @@ class NotificationDispatcher:
|
||||
asset_cache=self._asset_cache,
|
||||
)
|
||||
|
||||
for receiver in receivers:
|
||||
chat_id = receiver.get("chat_id")
|
||||
if not chat_id:
|
||||
results.append({"success": False, "error": "Missing chat_id in receiver"})
|
||||
for receiver in target.receivers:
|
||||
if not isinstance(receiver, TelegramReceiver) or not receiver.chat_id:
|
||||
results.append({"success": False, "error": "Invalid telegram receiver"})
|
||||
continue
|
||||
|
||||
# Step 1: Send the text message
|
||||
message = self._message_for_receiver(receiver, default_message, event, target)
|
||||
|
||||
text_result = await client.send_message(
|
||||
chat_id=str(chat_id),
|
||||
chat_id=receiver.chat_id,
|
||||
text=message,
|
||||
disable_web_page_preview=bool(disable_preview),
|
||||
)
|
||||
if not text_result.get("success"):
|
||||
_LOGGER.warning("Failed to send to chat %s: %s", chat_id, text_result.get("error"))
|
||||
_LOGGER.warning("Failed to send to chat %s: %s", receiver.chat_id, text_result.get("error"))
|
||||
results.append(text_result)
|
||||
continue
|
||||
|
||||
# Step 2: Send assets as reply
|
||||
if assets:
|
||||
reply_to = text_result.get("message_id")
|
||||
media_result = await client.send_notification(
|
||||
chat_id=str(chat_id),
|
||||
chat_id=receiver.chat_id,
|
||||
assets=assets,
|
||||
reply_to_message_id=reply_to,
|
||||
max_group_size=max_group,
|
||||
@@ -177,43 +203,40 @@ class NotificationDispatcher:
|
||||
chat_action=chat_action or None,
|
||||
)
|
||||
if not media_result.get("success"):
|
||||
_LOGGER.warning("Text sent OK but media failed for chat %s: %s", chat_id, media_result.get("error"))
|
||||
_LOGGER.warning("Text sent OK but media failed for chat %s: %s", receiver.chat_id, media_result.get("error"))
|
||||
|
||||
results.append(text_result)
|
||||
|
||||
return self._aggregate_results(results)
|
||||
|
||||
async def _send_webhook(
|
||||
self, target: TargetConfig, message: str, event: ServiceEvent
|
||||
self, target: TargetConfig, default_message: str, event: ServiceEvent
|
||||
) -> dict[str, Any]:
|
||||
if not target.receivers:
|
||||
return {"success": False, "error": "No receivers configured"}
|
||||
receivers = target.receivers
|
||||
|
||||
payload = {
|
||||
"message": message,
|
||||
"event_type": event.event_type.value,
|
||||
"provider_type": event.provider_type.value,
|
||||
"collection_name": event.collection_name,
|
||||
"collection_id": event.collection_id,
|
||||
"timestamp": event.timestamp.isoformat(),
|
||||
}
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
async with aiohttp.ClientSession() as session:
|
||||
for receiver in receivers:
|
||||
url = receiver.get("url")
|
||||
headers = receiver.get("headers", {})
|
||||
if not url:
|
||||
results.append({"success": False, "error": "Missing url in receiver"})
|
||||
for receiver in target.receivers:
|
||||
if not isinstance(receiver, WebhookReceiver) or not receiver.url:
|
||||
results.append({"success": False, "error": "Invalid webhook receiver"})
|
||||
continue
|
||||
client = WebhookClient(session, url, headers)
|
||||
message = self._message_for_receiver(receiver, default_message, event, target)
|
||||
payload = {
|
||||
"message": message,
|
||||
"event_type": event.event_type.value,
|
||||
"provider_type": event.provider_type.value,
|
||||
"collection_name": event.collection_name,
|
||||
"collection_id": event.collection_id,
|
||||
"timestamp": event.timestamp.isoformat(),
|
||||
}
|
||||
client = WebhookClient(session, receiver.url, receiver.headers)
|
||||
results.append(await client.send(payload))
|
||||
|
||||
return self._aggregate_results(results)
|
||||
|
||||
async def _send_email(
|
||||
self, target: TargetConfig, message: str, event: ServiceEvent
|
||||
self, target: TargetConfig, default_message: str, event: ServiceEvent
|
||||
) -> dict[str, Any]:
|
||||
from .email.client import EmailClient, SmtpConfig
|
||||
|
||||
@@ -233,71 +256,68 @@ class NotificationDispatcher:
|
||||
|
||||
if not target.receivers:
|
||||
return {"success": False, "error": "No receivers configured"}
|
||||
receivers = target.receivers
|
||||
subject = f"[Notify Bridge] {event.event_type.value}: {event.collection_name}"
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
for receiver in receivers:
|
||||
email = receiver.get("email")
|
||||
if not email:
|
||||
results.append({"success": False, "error": "Missing email in receiver"})
|
||||
for receiver in target.receivers:
|
||||
if not isinstance(receiver, EmailReceiver) or not receiver.email:
|
||||
results.append({"success": False, "error": "Invalid email receiver"})
|
||||
continue
|
||||
message = self._message_for_receiver(receiver, default_message, event, target)
|
||||
result = await client.send(
|
||||
to_email=email,
|
||||
to_email=receiver.email,
|
||||
subject=subject,
|
||||
body_text=message,
|
||||
to_name=receiver.get("name", ""),
|
||||
to_name=receiver.name,
|
||||
)
|
||||
results.append(result)
|
||||
|
||||
return self._aggregate_results(results)
|
||||
|
||||
async def _send_discord(
|
||||
self, target: TargetConfig, message: str, event: ServiceEvent
|
||||
self, target: TargetConfig, default_message: str, event: ServiceEvent
|
||||
) -> dict[str, Any]:
|
||||
from .discord.client import DiscordClient
|
||||
|
||||
if not target.receivers:
|
||||
return {"success": False, "error": "No receivers configured"}
|
||||
receivers = target.receivers
|
||||
username = target.config.get("username")
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
async with aiohttp.ClientSession() as session:
|
||||
client = DiscordClient(session)
|
||||
for receiver in receivers:
|
||||
webhook_url = receiver.get("webhook_url")
|
||||
if not webhook_url:
|
||||
results.append({"success": False, "error": "Missing webhook_url"})
|
||||
for receiver in target.receivers:
|
||||
if not isinstance(receiver, DiscordReceiver) or not receiver.webhook_url:
|
||||
results.append({"success": False, "error": "Invalid discord receiver"})
|
||||
continue
|
||||
results.append(await client.send(webhook_url, message, username=username))
|
||||
message = self._message_for_receiver(receiver, default_message, event, target)
|
||||
results.append(await client.send(receiver.webhook_url, message, username=username))
|
||||
|
||||
return self._aggregate_results(results)
|
||||
|
||||
async def _send_slack(
|
||||
self, target: TargetConfig, message: str, event: ServiceEvent
|
||||
self, target: TargetConfig, default_message: str, event: ServiceEvent
|
||||
) -> dict[str, Any]:
|
||||
from .slack.client import SlackClient
|
||||
|
||||
if not target.receivers:
|
||||
return {"success": False, "error": "No receivers configured"}
|
||||
receivers = target.receivers
|
||||
username = target.config.get("username")
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
async with aiohttp.ClientSession() as session:
|
||||
client = SlackClient(session)
|
||||
for receiver in receivers:
|
||||
webhook_url = receiver.get("webhook_url")
|
||||
if not webhook_url:
|
||||
results.append({"success": False, "error": "Missing webhook_url"})
|
||||
for receiver in target.receivers:
|
||||
if not isinstance(receiver, SlackReceiver) or not receiver.webhook_url:
|
||||
results.append({"success": False, "error": "Invalid slack receiver"})
|
||||
continue
|
||||
results.append(await client.send(webhook_url, message, username=username))
|
||||
message = self._message_for_receiver(receiver, default_message, event, target)
|
||||
results.append(await client.send(receiver.webhook_url, message, username=username))
|
||||
|
||||
return self._aggregate_results(results)
|
||||
|
||||
async def _send_ntfy(
|
||||
self, target: TargetConfig, message: str, event: ServiceEvent
|
||||
self, target: TargetConfig, default_message: str, event: ServiceEvent
|
||||
) -> dict[str, Any]:
|
||||
from .ntfy.client import NtfyClient
|
||||
|
||||
@@ -305,28 +325,26 @@ class NotificationDispatcher:
|
||||
auth_token = target.config.get("auth_token")
|
||||
if not target.receivers:
|
||||
return {"success": False, "error": "No receivers configured"}
|
||||
receivers = target.receivers
|
||||
|
||||
title = f"{event.event_type.value}: {event.collection_name}"
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
async with aiohttp.ClientSession() as session:
|
||||
client = NtfyClient(session)
|
||||
for receiver in receivers:
|
||||
topic = receiver.get("topic")
|
||||
if not topic:
|
||||
results.append({"success": False, "error": "Missing topic"})
|
||||
for receiver in target.receivers:
|
||||
if not isinstance(receiver, NtfyReceiver) or not receiver.topic:
|
||||
results.append({"success": False, "error": "Invalid ntfy receiver"})
|
||||
continue
|
||||
priority = receiver.get("priority", 3)
|
||||
message = self._message_for_receiver(receiver, default_message, event, target)
|
||||
results.append(await client.send(
|
||||
server_url, topic, message,
|
||||
title=title, priority=priority, auth_token=auth_token,
|
||||
server_url, receiver.topic, message,
|
||||
title=title, priority=receiver.priority, auth_token=auth_token,
|
||||
))
|
||||
|
||||
return self._aggregate_results(results)
|
||||
|
||||
async def _send_matrix(
|
||||
self, target: TargetConfig, message: str, event: ServiceEvent
|
||||
self, target: TargetConfig, default_message: str, event: ServiceEvent
|
||||
) -> dict[str, Any]:
|
||||
from .matrix.client import MatrixClient
|
||||
|
||||
@@ -337,18 +355,17 @@ class NotificationDispatcher:
|
||||
|
||||
if not target.receivers:
|
||||
return {"success": False, "error": "No receivers configured"}
|
||||
receivers = target.receivers
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
async with aiohttp.ClientSession() as session:
|
||||
client = MatrixClient(session, homeserver, access_token)
|
||||
for receiver in receivers:
|
||||
room_id = receiver.get("room_id")
|
||||
if not room_id:
|
||||
results.append({"success": False, "error": "Missing room_id"})
|
||||
for receiver in target.receivers:
|
||||
if not isinstance(receiver, MatrixReceiver) or not receiver.room_id:
|
||||
results.append({"success": False, "error": "Invalid matrix receiver"})
|
||||
continue
|
||||
message = self._message_for_receiver(receiver, default_message, event, target)
|
||||
results.append(await client.send_message(
|
||||
room_id, message, html_message=message,
|
||||
receiver.room_id, message, html_message=message,
|
||||
))
|
||||
|
||||
return self._aggregate_results(results)
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
"""Receiver hierarchy — typed delivery endpoints with locale resolution."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class Receiver:
|
||||
"""Base receiver with locale and raw config."""
|
||||
|
||||
locale: str = ""
|
||||
config: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Return raw dict for backward compat (includes locale)."""
|
||||
return {**self.config, "locale": self.locale}
|
||||
|
||||
|
||||
@dataclass
|
||||
class TelegramReceiver(Receiver):
|
||||
"""Telegram chat receiver."""
|
||||
|
||||
chat_id: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class WebhookReceiver(Receiver):
|
||||
"""Webhook URL receiver."""
|
||||
|
||||
url: str = ""
|
||||
headers: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EmailReceiver(Receiver):
|
||||
"""Email receiver."""
|
||||
|
||||
email: str = ""
|
||||
name: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class DiscordReceiver(Receiver):
|
||||
"""Discord webhook receiver."""
|
||||
|
||||
webhook_url: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class SlackReceiver(Receiver):
|
||||
"""Slack webhook receiver."""
|
||||
|
||||
webhook_url: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class NtfyReceiver(Receiver):
|
||||
"""ntfy topic receiver."""
|
||||
|
||||
topic: str = ""
|
||||
priority: int = 3
|
||||
|
||||
|
||||
@dataclass
|
||||
class MatrixReceiver(Receiver):
|
||||
"""Matrix room receiver."""
|
||||
|
||||
room_id: str = ""
|
||||
|
||||
|
||||
def build_receiver(target_type: str, config: dict[str, Any], locale: str = "") -> Receiver:
|
||||
"""Factory: build typed Receiver from target type and config dict."""
|
||||
if target_type == "telegram":
|
||||
return TelegramReceiver(
|
||||
locale=locale,
|
||||
config=config,
|
||||
chat_id=str(config.get("chat_id", "")),
|
||||
)
|
||||
if target_type == "webhook":
|
||||
return WebhookReceiver(
|
||||
locale=locale,
|
||||
config=config,
|
||||
url=config.get("url", ""),
|
||||
headers=config.get("headers", {}),
|
||||
)
|
||||
if target_type == "email":
|
||||
return EmailReceiver(
|
||||
locale=locale,
|
||||
config=config,
|
||||
email=config.get("email", ""),
|
||||
name=config.get("name", ""),
|
||||
)
|
||||
if target_type == "discord":
|
||||
return DiscordReceiver(
|
||||
locale=locale,
|
||||
config=config,
|
||||
webhook_url=config.get("webhook_url", ""),
|
||||
)
|
||||
if target_type == "slack":
|
||||
return SlackReceiver(
|
||||
locale=locale,
|
||||
config=config,
|
||||
webhook_url=config.get("webhook_url", ""),
|
||||
)
|
||||
if target_type == "ntfy":
|
||||
return NtfyReceiver(
|
||||
locale=locale,
|
||||
config=config,
|
||||
topic=config.get("topic", ""),
|
||||
priority=config.get("priority", 3),
|
||||
)
|
||||
if target_type == "matrix":
|
||||
return MatrixReceiver(
|
||||
locale=locale,
|
||||
config=config,
|
||||
room_id=config.get("room_id", ""),
|
||||
)
|
||||
return Receiver(locale=locale, config=config)
|
||||
Reference in New Issue
Block a user