Slack @-mention to ntfy.sh

bridge.py

import logging
import os
import threading
import time

import requests
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger("slack-ntfy-bridge")

SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_APP_TOKEN = os.environ["SLACK_APP_TOKEN"]
SLACK_USER_ID = os.environ["SLACK_USER_ID"]
ALERT_CHANNEL_ID = os.environ["ALERT_CHANNEL_ID"]
NTFY_TOPIC = os.environ["NTFY_TOPIC"]

app = App(token=SLACK_BOT_TOKEN)


def resolve_sender_name(client, user_id: str) -> str:
    try:
        info = client.users_info(user=user_id)
        profile = info["user"]["profile"]
        return profile.get("display_name") or profile.get("real_name") or user_id
    except Exception:
        logger.exception("Failed to look up user %s, falling back to raw ID", user_id)
        return user_id


def send_ntfy(title: str, body: str) -> None:
    try:
        requests.post(
            f"https://ntfy.sh/{NTFY_TOPIC}",
            data=body.encode("utf-8"),
            headers={
                "Title": title,
                "Priority": "high",
                "Tags": "rotating_light",
            },
            timeout=10,
        )
        logger.info("Sent ntfy notification: %s", title)
    except Exception:
        logger.exception("Failed to send ntfy notification")


@app.event("message")
def handle_message(event, client):
    channel = event.get("channel")
    text = event.get("text", "")

    if channel != ALERT_CHANNEL_ID:
        return

    if f"<@{SLACK_USER_ID}>" not in text:
        return

    sender_id = event.get("user", "unknown")
    sender_name = resolve_sender_name(client, sender_id)

    logger.info("Mention detected from %s: %s", sender_name, text)
    send_ntfy(f"{sender_name} in #alerts-production", text)


def heartbeat():
    while True:
        logger.info("heartbeat: alive")
        time.sleep(60)


if __name__ == "__main__":
    threading.Thread(target=heartbeat, daemon=True).start()
    logger.info("Starting slack-ntfy-bridge")
    handler = SocketModeHandler(app, SLACK_APP_TOKEN)
    handler.start()