afi-backup-monitor/src/shared/python/slack.py

"""Slack webhook helper for posting messages."""

import json
import os
from urllib import request, error


def post_message(blocks, text="Afi Backup Monitor"):
    """Post a Block Kit message to Slack via webhook."""
    webhook_url = os.environ["SLACK_WEBHOOK_URL"]
    payload = json.dumps({"text": text, "blocks": blocks}).encode()

    req = request.Request(
        webhook_url,
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )

    try:
        with request.urlopen(req) as resp:
            return resp.read().decode()
    except error.HTTPError as e:
        error_body = e.read().decode() if e.fp else ""
        raise RuntimeError(f"Slack webhook failed ({e.code}): {error_body}")


def header_block(text):
    return {"type": "header", "text": {"type": "plain_text", "text": text}}


def section_block(text):
    return {"type": "section", "text": {"type": "mrkdwn", "text": text}}


def divider_block():
    return {"type": "divider"}