Add Apprise

app/fanout/apprise_mod.py

@@ -0,0 +1,125 @@
+"""Fanout module for Apprise push notifications."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+from app.fanout.base import FanoutModule
+
+logger = logging.getLogger(__name__)
+
+
+def _parse_urls(raw: str) -> list[str]:
+    """Split multi-line URL string into individual URLs."""
+    return [line.strip() for line in raw.splitlines() if line.strip()]
+
+
+def _normalize_discord_url(url: str) -> str:
+    """Add avatar=no to Discord URLs to suppress identity override."""
+    parts = urlsplit(url)
+    scheme = parts.scheme.lower()
+    host = parts.netloc.lower()
+
+    is_discord = scheme in ("discord", "discords") or (
+        scheme in ("http", "https")
+        and host in ("discord.com", "discordapp.com")
+        and parts.path.lower().startswith("/api/webhooks/")
+    )
+    if not is_discord:
+        return url
+
+    query = dict(parse_qsl(parts.query, keep_blank_values=True))
+    query["avatar"] = "no"
+    return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment))
+
+
+def _format_body(data: dict, *, include_path: bool) -> str:
+    """Build a human-readable notification body from message data."""
+    msg_type = data.get("type", "")
+    text = data.get("text", "")
+    sender_name = data.get("sender_name") or "Unknown"
+
+    via = ""
+    if include_path:
+        paths = data.get("paths")
+        if paths and isinstance(paths, list) and len(paths) > 0:
+            path_str = paths[0].get("path", "") if isinstance(paths[0], dict) else ""
+        else:
+            path_str = None
+
+        if msg_type == "PRIV" and path_str is None:
+            via = " **via:** [`direct`]"
+        elif path_str is not None:
+            path_str = path_str.strip().lower()
+            if path_str == "":
+                via = " **via:** [`direct`]"
+            else:
+                hops = [path_str[i : i + 2] for i in range(0, len(path_str), 2)]
+                if hops:
+                    hop_list = ", ".join(f"`{h}`" for h in hops)
+                    via = f" **via:** [{hop_list}]"
+
+    if msg_type == "PRIV":
+        return f"**DM:** {sender_name}: {text}{via}"
+
+    channel_name = data.get("channel_name") or data.get("conversation_key", "channel")
+    return f"**{channel_name}:** {sender_name}: {text}{via}"
+
+
+def _send_sync(urls_raw: str, body: str, *, preserve_identity: bool) -> bool:
+    """Send notification synchronously via Apprise. Returns True on success."""
+    import apprise as apprise_lib
+
+    urls = _parse_urls(urls_raw)
+    if not urls:
+        return False
+
+    notifier = apprise_lib.Apprise()
+    for url in urls:
+        if preserve_identity:
+            url = _normalize_discord_url(url)
+        notifier.add(url)
+
+    return bool(notifier.notify(title="", body=body))
+
+
+class AppriseModule(FanoutModule):
+    """Sends push notifications via Apprise for incoming messages."""
+
+    def __init__(self, config_id: str, config: dict) -> None:
+        super().__init__(config_id, config)
+        self._last_error: str | None = None
+
+    async def on_message(self, data: dict) -> None:
+        # Skip outgoing messages — only notify on incoming
+        if data.get("outgoing"):
+            return
+
+        urls = self.config.get("urls", "")
+        if not urls or not urls.strip():
+            return
+
+        preserve_identity = self.config.get("preserve_identity", True)
+        include_path = self.config.get("include_path", True)
+        body = _format_body(data, include_path=include_path)
+
+        try:
+            success = await asyncio.to_thread(
+                _send_sync, urls, body, preserve_identity=preserve_identity
+            )
+            self._last_error = None if success else "Apprise notify returned failure"
+            if not success:
+                logger.warning("Apprise notification failed for module %s", self.config_id)
+        except Exception as exc:
+            self._last_error = str(exc)
+            logger.exception("Apprise send error for module %s", self.config_id)
+
+    @property
+    def status(self) -> str:
+        if not self.config.get("urls", "").strip():
+            return "disconnected"
+        if self._last_error:
+            return "error"
+        return "connected"

app/fanout/manager.py

@@ -17,6 +17,7 @@ def _register_module_types() -> None:
     """Lazily populate the type registry to avoid circular imports."""
     if _MODULE_TYPES:
         return
+    from app.fanout.apprise_mod import AppriseModule
     from app.fanout.bot import BotModule
     from app.fanout.mqtt_community import MqttCommunityModule
     from app.fanout.mqtt_private import MqttPrivateModule
@@ -26,6 +27,7 @@ def _register_module_types() -> None:
     _MODULE_TYPES["mqtt_community"] = MqttCommunityModule
     _MODULE_TYPES["bot"] = BotModule
     _MODULE_TYPES["webhook"] = WebhookModule
+    _MODULE_TYPES["apprise"] = AppriseModule
 
 
 def _matches_filter(filter_value: Any, key: str) -> bool:

app/routers/fanout.py

@@ -12,7 +12,7 @@ from app.repository.fanout import FanoutConfigRepository
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/fanout", tags=["fanout"])
 
-_VALID_TYPES = {"mqtt_private", "mqtt_community", "bot", "webhook"}
+_VALID_TYPES = {"mqtt_private", "mqtt_community", "bot", "webhook", "apprise"}
 
 _IATA_RE = re.compile(r"^[A-Z]{3}$")
 
@@ -65,6 +65,13 @@ def _validate_bot_config(config: dict) -> None:
         ) from None
 
 
+def _validate_apprise_config(config: dict) -> None:
+    """Validate apprise config blob."""
+    urls = config.get("urls", "")
+    if not urls or not urls.strip():
+        raise HTTPException(status_code=400, detail="At least one Apprise URL is required")
+
+
 def _validate_webhook_config(config: dict) -> None:
     """Validate webhook config blob."""
     url = config.get("url", "")
@@ -86,7 +93,7 @@ def _enforce_scope(config_type: str, scope: dict) -> dict:
         return {"messages": "none", "raw_packets": "all"}
     if config_type == "bot":
         return {"messages": "all", "raw_packets": "none"}
-    if config_type == "webhook":
+    if config_type in ("webhook", "apprise"):
         messages = scope.get("messages", "all")
         if messages not in ("all", "none") and not isinstance(messages, dict):
             messages = "all"
@@ -130,6 +137,8 @@ async def create_fanout_config(body: FanoutConfigCreate) -> dict:
             _validate_bot_config(body.config)
         elif body.type == "webhook":
             _validate_webhook_config(body.config)
+        elif body.type == "apprise":
+            _validate_apprise_config(body.config)
 
     scope = _enforce_scope(body.type, body.scope)
 
@@ -180,6 +189,8 @@ async def update_fanout_config(config_id: str, body: FanoutConfigUpdate) -> dict
             _validate_bot_config(config_to_validate)
         elif existing["type"] == "webhook":
             _validate_webhook_config(config_to_validate)
+        elif existing["type"] == "apprise":
+            _validate_apprise_config(config_to_validate)
 
     updated = await FanoutConfigRepository.update(config_id, **kwargs)
     if updated is None: