slack_notifier.py

"""Outbound Slack deployment notifications."""

from __future__ import annotations

import os
from typing import Iterable

import requests


SLACK_POST_MESSAGE_URL = "https://slack.com/api/chat.postMessage"


def _is_enabled() -> bool:
    return os.getenv("SLACK_NOTIFICATIONS_ENABLED", "true").strip().lower() not in {
        "0",
        "false",
        "no",
        "off",
    }


def _get_channel_id() -> str:
    return (
        os.getenv("SLACK_DEPLOYMENT_CHANNEL_ID", "").strip()
        or os.getenv("SLACK_CHANNEL_ID", "").strip()
    )


def _channel_hint(channel: str) -> str:
    if not channel:
        return "missing"
    return f"{channel[:4]}...{channel[-3:]}" if len(channel) > 7 else channel


def send_slack_message(text: str) -> tuple[bool, str]:
    """Send a plain outbound Slack message.

    Missing Slack configuration is treated as a clean no-op so deployments never
    fail because notifications are not configured.
    """
    if not _is_enabled():
        print("[Slack] notification skipped: disabled", flush=True)
        return False, "Slack notifications disabled"

    token = os.getenv("SLACK_BOT_TOKEN", "").strip()
    channel = _get_channel_id()
    if not token:
        print("[Slack] notification skipped: SLACK_BOT_TOKEN missing", flush=True)
        return False, "Slack token not configured"
    if not channel:
        print("[Slack] notification skipped: SLACK_DEPLOYMENT_CHANNEL_ID missing", flush=True)
        return False, "Slack token or channel not configured"

    try:
        response = requests.post(
            SLACK_POST_MESSAGE_URL,
            headers={
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json; charset=utf-8",
            },
            json={
                "channel": channel,
                "text": text,
                "mrkdwn": True,
                "unfurl_links": False,
                "unfurl_media": False,
            },
            timeout=10,
        )
        response.raise_for_status()
        payload = response.json()
        if not payload.get("ok"):
            error = payload.get("error", "Slack API returned ok=false")
            print(f"[Slack] notification failed: {error} channel={_channel_hint(channel)}", flush=True)
            return False, error
        print(f"[Slack] notification sent channel={_channel_hint(channel)}", flush=True)
        return True, "sent"
    except Exception as exc:
        print(f"[Slack] notification exception: {exc}", flush=True)
        return False, str(exc)


def notify_deployment_event(
    title: str,
    status: str,
    details: Iterable[tuple[str, object]] | None = None,
) -> tuple[bool, str]:
    """Send a compact deployment status message to Slack."""
    lines = [f"*{title}*", f"Status: {status}"]
    for label, value in details or []:
        if value in (None, "", [], {}):
            continue
        lines.append(f"{label}: {value}")
    return send_slack_message("\n".join(lines))


def notify_data_fallback_event(
    *,
    company: str,
    use_case: str,
    scenario_type: str,
    fallback_events: Iterable[dict] | None = None,
) -> tuple[bool, str]:
    """Notify DemoPrep channel when dataset generation pads domain values."""
    details = [
        ("Company", company),
        ("Use case", use_case),
        ("Scenario", scenario_type),
    ]
    events = list(fallback_events or [])
    if events:
        preview = []
        for event in events[:5]:
            values = ", ".join(str(v) for v in event.get("fallback_values", [])[:4])
            preview.append(f"{event.get('dimension')}: {values}")
        details.append(("Fallback dimensions", "; ".join(preview)))
    return notify_deployment_event("DemoPrep data fallback used", "Warning", details)


def notify_dataset_quality_failure(
    *,
    company: str,
    use_case: str,
    scenario_type: str,
    summary: str,
) -> tuple[bool, str]:
    """Notify DemoPrep channel when generated data is blocked before deploy."""
    return notify_deployment_event(
        "DemoPrep dataset quality gate blocked deploy",
        "Failed",
        [
            ("Company", company),
            ("Use case", use_case),
            ("Scenario", scenario_type),
            ("Issues", summary[:1200]),
        ],
    )