Source code for operators.communications.notify_slack

"""
NotifySlackOperator — post a notification message to a Slack channel.

Drop-in companion to EmailOperator for DAG failure/success callbacks.
Call .execute(context) inside a callback, or use as a standalone task.

Implementation notes
--------------------
- Uses SlackHook with Config.SLACK_DEFAULT_API ("slack_default_api" connection).
  Same hook pattern used across the codebase (dbt notifications, ingest_slack, etc.).
- mrkdwn=True enables Slack markdown (*bold*, `code`, :emoji:) in the message body.
- @mentions: pass slack_user_ids directly via mention_user_ids (static, no API call), or
  pass emails via mention_slack_emails. Email → Slack ID resolution order:
    1. Local lookup via _get_slack_id (.agents/airflow-sso-users.json) — no API call
    2. Slack API users.lookupByEmail (requires users:read.email scope — best-effort fallback)

nhat@ready.net | 2026-02-23
"""

import json
import os

from airflow.models import BaseOperator
from airflow.providers.slack.hooks.slack import SlackHook
from config import Config

# Inline SSO lookup — avoids importing from util (which would create a circular import
# via util/dag.py → NotifySlackOperator → util). Intentional small duplication.
_SSO_USERS: dict = {}
try:
    with open(os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
                           ".agents", "airflow-sso-users.json")) as _f:
        _SSO_USERS = json.load(_f)
except (FileNotFoundError, json.JSONDecodeError):
    pass


def _get_slack_id(email: str | None) -> str | None:
    if not email:
        return None
    entry = _SSO_USERS.get(email)
    return entry.get("slack_user_id") if entry else None




class NotifySlackOperator(BaseOperator):
    """
    Post a notification message to a Slack channel.

    Args:
        channel:              Slack channel name or ID, e.g. "#airflow"
        text:                 Message body — supports Slack mrkdwn (*bold*, `code`, :emoji:)
        mention_slack_emails: Emails to @mention at the top of the message. Resolved to
                              Slack user IDs via local SSO lookup first, then Slack API
                              users.lookupByEmail as fallback. Best-effort: unresolved
                              emails are silently skipped.
        mention_user_ids:     Slack user IDs to @mention directly (no lookup needed).
        slack_conn_id:        Airflow connection ID for the Slack API token
                              (defaults to Config.SLACK_DEFAULT_API = "slack_default_api")

    Usage in a callback::

        NotifySlackOperator(
            task_id="slack_notify",
            channel="#airflow",
            text=":red_circle: *FAILED* | `my_dag`\\nTriggered by: Nhat Nguyen | duration: 4s",
            mention_slack_emails=["nhat@ready.net"],
        ).execute(context)
    """

    ui_color = "#4A154B"  # Slack purple
    template_fields = ("text",)

    def __init__(
        self,
        channel: str,
        text: str,
        mention_slack_emails: list[str] | None = None,
        mention_user_ids: list[str] | None = None,
        slack_conn_id: str = Config.SLACK_DEFAULT_API,
        **kwargs,
    ):
        super().__init__(**kwargs)
        self.channel = channel
        self.text = text
        self.mention_slack_emails = mention_slack_emails or []
        self.mention_user_ids = mention_user_ids or []
        self.slack_conn_id = slack_conn_id



    def execute(self, context):
        hook = SlackHook(slack_conn_id=self.slack_conn_id)

        # Start with directly-provided user IDs (no lookup needed)
        mention_ids = list(self.mention_user_ids)

        for email in self.mention_slack_emails:
            # 1. Local SSO lookup — no API call, no OAuth scope required
            uid = _get_slack_id(email)
            if uid:
                if uid not in mention_ids:
                    mention_ids.append(uid)
                continue

            # 2. Slack API fallback — requires users:read.email scope
            try:
                response = hook.client.users_lookupByEmail(email=email)
                if response.get("ok"):
                    uid = response.get("user", {}).get("id")
                    if uid and uid not in mention_ids:
                        mention_ids.append(uid)
                else:
                    self.log.warning(
                        "users.lookupByEmail failed for %s: %s", email, response.get("error")
                    )
            except Exception as e:
                self.log.warning("users.lookupByEmail exception for %s: %s", email, e)

        mention_prefix = " ".join(f"<@{uid}>" for uid in mention_ids)
        full_text = f"{mention_prefix}\n{self.text}".strip() if mention_prefix else self.text
        hook.call(
            "chat.postMessage",
            json={"channel": self.channel, "text": full_text, "mrkdwn": True},
        )