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

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

import json
import os
from urllib import request, error

_webhook_url = None


def _get_webhook_url():
    global _webhook_url
    if _webhook_url is None:
        import boto3

        client = boto3.client("secretsmanager")
        secret_arn = os.environ["SLACK_WEBHOOK_SECRET_ARN"]
        response = client.get_secret_value(SecretId=secret_arn)
        _webhook_url = response["SecretString"]
    return _webhook_url


def post_message(blocks, text="Afi Backup Monitor"):
    """Post a Block Kit message to Slack via webhook."""
    webhook_url = _get_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"}