alexei.dolgolyov/notify-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 31 Wiki Activity

feat: provider-strict configs, slot-based templates, broadcast targets, email bots, command templates

Browse Source 
Major architectural improvements:
- Provider-type enforcement: configs validated against provider type at assignment
- TemplateConfig migrated to slot-based pattern (TemplateSlot child table)
- Broadcast targets: TargetReceiver child table for multi-receiver dispatch
- EmailBot: first-class email sender entity with SMTP config, test connection
- CommandTemplateConfig: generic slot-based command response templates
- Provider capability registry: dynamic slot/event/command definitions per provider
- CommandTracker play/pause button matches NotificationTracker style
This commit is contained in:
Admin
2026-03-21 16:33:24 +03:00
parent 371ea70756
commit 846d480d38
27 changed files with 2355 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/notify_bridge_server/services/notifier.py
+134 -20
View File
@@ -5,7 +5,11 @@ from typing import Any
import aiohttp
from ..database.models import NotificationTarget
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from ..database.engine import get_engine
from ..database.models import NotificationTarget, TargetReceiver
_LOGGER = logging.getLogger(__name__)
@@ -26,52 +30,162 @@ def _get_test_message(locale: str, target_type: str) -> str:
return msgs.get(target_type, msgs.get("webhook", "Test"))
async def _load_receivers(target_id: int) -> list[dict]:
"""Load enabled receivers for a target from DB."""
engine = get_engine()
async with AsyncSession(engine) as session:
result = await session.exec(
select(TargetReceiver).where(
TargetReceiver.target_id == target_id,
TargetReceiver.enabled == True,
)
)
return [dict(r.config) for r in result.all()]
async def send_to_target(target: NotificationTarget, message: str) -> dict:
"""Send a message to a target, respecting all target config settings.
"""Send a message to a target, broadcasting to all receivers.
This is the SINGLE send path used by dispatch, test, and real-data notifications.
"""
try:
receivers = await _load_receivers(target.id)
if target.type == "telegram":
return await _send_telegram(target, message)
return await _send_telegram_broadcast(target, message, receivers)
elif target.type == "webhook":
return await _send_webhook(target, message)
return await _send_webhook_broadcast(target, message, receivers)
elif target.type == "email":
return await _send_email_broadcast(target, message, receivers)
return {"success": False, "error": f"Unknown target type: {target.type}"}
except Exception as e:
_LOGGER.error("Send failed: %s", e)
return {"success": False, "error": str(e)}
async def _send_telegram(target: NotificationTarget, message: str) -> dict:
async def _send_telegram_broadcast(target: NotificationTarget, message: str, receivers: list[dict]) -> dict:
from notify_bridge_core.notifications.telegram.client import TelegramClient
bot_token = target.config.get("bot_token")
chat_id = target.config.get("chat_id")
disable_preview = target.config.get("disable_url_preview", False)
if not bot_token or not chat_id:
return {"success": False, "error": "Missing bot_token or chat_id"}
if not bot_token:
return {"success": False, "error": "Missing bot_token"}
# Fall back to legacy chat_id if no receivers
if not receivers:
chat_id = target.config.get("chat_id")
if chat_id:
receivers = [{"chat_id": str(chat_id)}]
else:
return {"success": False, "error": "No receivers configured"}
results: list[dict] = []
async with aiohttp.ClientSession() as session:
client = TelegramClient(session, bot_token)
return await client.send_message(
chat_id=str(chat_id),
text=message,
disable_web_page_preview=bool(disable_preview),
)
for recv in receivers:
chat_id = recv.get("chat_id")
if not chat_id:
continue
result = await client.send_message(
chat_id=str(chat_id),
text=message,
disable_web_page_preview=bool(disable_preview),
)
results.append(result)
successes = sum(1 for r in results if r.get("success"))
if successes == len(results) and results:
return {"success": True, "receivers": len(results)}
elif successes > 0:
return {"success": True, "receivers": len(results), "partial_failures": len(results) - successes}
elif results:
return results[0]
return {"success": False, "error": "No valid receivers"}
async def _send_webhook(target: NotificationTarget, message: str, event_type: str = "notification") -> dict:
async def _send_webhook_broadcast(target: NotificationTarget, message: str, receivers: list[dict]) -> dict:
from notify_bridge_core.notifications.webhook.client import WebhookClient
url = target.config.get("url")
headers = target.config.get("headers", {})
if not url:
return {"success": False, "error": "Missing url in target config"}
# Fall back to legacy url if no receivers
if not receivers:
url = target.config.get("url")
headers = target.config.get("headers", {})
if url:
receivers = [{"url": url, "headers": headers}]
else:
return {"success": False, "error": "No receivers configured"}
results: list[dict] = []
async with aiohttp.ClientSession() as session:
client = WebhookClient(session, url, headers)
return await client.send({"message": message, "event_type": event_type})
for recv in receivers:
url = recv.get("url")
headers = recv.get("headers", {})
if not url:
continue
client = WebhookClient(session, url, headers)
results.append(await client.send({"message": message, "event_type": "notification"}))
successes = sum(1 for r in results if r.get("success"))
if successes == len(results) and results:
return {"success": True, "receivers": len(results)}
elif successes > 0:
return {"success": True, "receivers": len(results), "partial_failures": len(results) - successes}
elif results:
return results[0]
return {"success": False, "error": "No valid receivers"}
async def _send_email_broadcast(target: NotificationTarget, message: str, receivers: list[dict]) -> dict:
from notify_bridge_core.notifications.email.client import EmailClient, SmtpConfig
from ..database.models import EmailBot
email_bot_id = target.config.get("email_bot_id")
if not email_bot_id:
return {"success": False, "error": "No email bot configured for this target"}
engine = get_engine()
async with AsyncSession(engine) as session:
email_bot = await session.get(EmailBot, email_bot_id)
if not email_bot:
return {"success": False, "error": "Email bot not found"}
smtp_cfg = SmtpConfig(
host=email_bot.smtp_host,
port=email_bot.smtp_port,
username=email_bot.smtp_username,
password=email_bot.smtp_password,
from_address=email_bot.email,
from_name=email_bot.name,
use_tls=email_bot.smtp_use_tls,
)
if not smtp_cfg.host or not smtp_cfg.from_address:
return {"success": False, "error": "Email bot SMTP not configured"}
if not receivers:
return {"success": False, "error": "No email receivers configured"}
client = EmailClient(smtp_cfg)
results: list[dict] = []
for recv in receivers:
email = recv.get("email")
if not email:
continue
result = await client.send(
to_email=email,
subject="Notification from Notify Bridge",
body_text=message,
to_name=recv.get("name", ""),
)
results.append(result)
successes = sum(1 for r in results if r.get("success"))
if successes == len(results) and results:
return {"success": True, "receivers": len(results)}
elif successes > 0:
return {"success": True, "receivers": len(results), "partial_failures": len(results) - successes}
elif results:
return results[0]
return {"success": False, "error": "No valid email receivers"}
# --- Public API used by routes ---
packages/server/src/notify_bridge_server/services/watcher.py
+46 -12
View File
@@ -17,13 +17,16 @@ from notify_bridge_core.storage import JsonFileBackend
from ..database.engine import get_engine
from ..database.models import (
EmailBot,
EventLog,
NotificationTarget,
NotificationTracker,
NotificationTrackerState,
NotificationTrackerTarget,
ServiceProvider,
TargetReceiver,
TemplateConfig,
TemplateSlot,
TrackingConfig,
)
@@ -129,19 +132,59 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
if not target:
continue
# Load receivers for this target
recv_result = await session.exec(
select(TargetReceiver).where(
TargetReceiver.target_id == target.id,
TargetReceiver.enabled == True,
)
)
receivers = [dict(r.config) for r in recv_result.all()]
tracking_config = None
if tt.tracking_config_id:
tracking_config = await session.get(TrackingConfig, tt.tracking_config_id)
template_config = None
template_slots: dict[str, str] | None = None
if tt.template_config_id:
template_config = await session.get(TemplateConfig, tt.template_config_id)
if template_config:
slot_result = await session.exec(
select(TemplateSlot).where(TemplateSlot.config_id == template_config.id)
)
raw_slots = {s.slot_name: s.template for s in slot_result.all()}
# Map slot names to event_type values for dispatcher lookup
template_slots = {}
for slot_name, tmpl_text in raw_slots.items():
# Strip "message_" prefix for event-type slots
event_key = slot_name.removeprefix("message_") if slot_name.startswith("message_") else slot_name
template_slots[event_key] = tmpl_text
target_config = dict(target.config)
# Inject SMTP config for email targets from EmailBot
if target.type == "email":
email_bot_id = target.config.get("email_bot_id")
if email_bot_id:
email_bot = await session.get(EmailBot, email_bot_id)
if email_bot:
target_config["smtp"] = {
"host": email_bot.smtp_host,
"port": email_bot.smtp_port,
"username": email_bot.smtp_username,
"password": email_bot.smtp_password,
"from_address": email_bot.email,
"from_name": email_bot.name,
"use_tls": email_bot.smtp_use_tls,
}
link_data.append({
"target_type": target.type,
"target_config": dict(target.config),
"target_config": target_config,
"receivers": receivers,
"tracking_config": tracking_config,
"template_config": template_config,
"template_slots": template_slots,
})
# Snapshot the data we need
@@ -249,26 +292,17 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
_LOGGER.info(" Skipped by tracking config filter")
continue
# Build template slots from template config
tmpl = ld["template_config"]
slots = None
if tmpl:
slots = {
"assets_added": tmpl.message_assets_added,
"assets_removed": tmpl.message_assets_removed,
"collection_renamed": tmpl.message_collection_renamed,
"collection_deleted": tmpl.message_collection_deleted,
"sharing_changed": tmpl.message_sharing_changed,
}
target_configs.append(TargetConfig(
type=ld["target_type"],
config=ld["target_config"],
template_slots=slots,
template_slots=ld["template_slots"],
date_format=tmpl.date_format if tmpl else "%d.%m.%Y, %H:%M UTC",
date_only_format=tmpl.date_only_format if tmpl and tmpl.date_only_format else "%d.%m.%Y",
provider_api_key=provider_config.get("api_key"),
provider_internal_url=provider_config.get("url", ""),
provider_external_url=provider_config.get("external_domain", ""),
receivers=ld["receivers"],
))
if target_configs: