Slack botでOpsgenieのオンコール担当者をメンションする

"""Mention who is on call in Slack."""

import json
import logging
import os
from typing import Literal, TypeVar

import pydantic
import requests
import slack_sdk
from aws_lambda_typing import context as context_
from aws_lambda_typing import events, responses
from slack_sdk import errors

logger = logging.getLogger()
logger.setLevel(logging.INFO)


SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
GENIE_KEY = os.environ["GENIE_KEY"]


T = TypeVar("T")


def not_none(obj: T | None) -> T:
    """Not None assertion."""
    if obj is None:
        raise ValueError("Not none assertion fails")
    return obj


def _get_slack_channel_name_from_channel_id(channel_id: str) -> str:
    """Get a Slack channel name from a Slack channel id.

    https://api.slack.com/methods/conversations.info
    Required Bot token scopes: channels:read, group:read, im:read, and mpim:read
    """
    client = slack_sdk.WebClient(token=SLACK_BOT_TOKEN)
    try:
        result = client.conversations_info(channel=channel_id)
        logger.info(result)
    except errors.SlackApiError as e:
        logger.error("Error getting a Slack channel name: %s", e)
        raise e
    channel: dict[str, str] = not_none(result.get("channel"))
    return not_none(channel.get("name"))


class _WhoIsOnCallDataItem(pydantic.BaseModel):
    on_call_recipients: list[str] = pydantic.Field(alias="onCallRecipients")


class _WhoIsOnCallResponse(pydantic.BaseModel):
    """Who is on call response schema."""

    data: _WhoIsOnCallDataItem


def _get_who_is_on_call_from_opsgenie(
    on_call_schedule: str | None,
) -> str | None:
    """Get an email address of who is on call from Opsgenie.

    https://docs.opsgenie.com/docs/who-is-on-call-api
    """
    if on_call_schedule is None:
        return None
    try:
        response = requests.get(
            url="https://api.opsgenie.com/v2/schedules/"
            + on_call_schedule
            + "/on-calls?scheduleIdentifierType=name&flat=true",
            headers={"Authorization": "GenieKey " + GENIE_KEY},
            timeout=10,
        )
        logger.info(response)
    except requests.exceptions.RequestException as e:
        logger.error("Error getting who is on call from Opsgenie: %s", e)
        raise e
    on_call_email_address = _WhoIsOnCallResponse.parse_obj(
        response.json()
    ).data.on_call_recipients[0]
    return on_call_email_address


def _get_user_id_from_email_address(
    email_address: str | None,
) -> str | None:
    """Get a Slack user id from a Slack email address.

    https://api.slack.com/methods/users.lookupByEmail
    Required Bot token scopes: users:read.email
    """
    if email_address is None:
        return None
    client = slack_sdk.WebClient(token=SLACK_BOT_TOKEN)
    try:
        result = client.users_lookupByEmail(email=email_address)
        logger.info(result)
    except errors.SlackApiError as e:
        logger.error("Error getting a Slack user id: %s", e)
        raise e
    user: dict[str, str] = not_none(result.get("user"))
    return not_none(user.get("id"))


def _post_slack_message(
    channel_id: str, thread_ts: str, slack_user_id: str | None
) -> None:
    """Post a Slack message.

    https://api.slack.com/methods/chat.postMessage
    Required Bot token scopes: chat:write
    """
    client = slack_sdk.WebClient(token=SLACK_BOT_TOKEN)
    if slack_user_id is None:
        try:
            result = client.chat_postMessage(
                channel=channel_id,
                thread_ts=thread_ts,
                text="Who is on call bot cannot be used in this channel.",
            )
            logger.info(result)

        except errors.SlackApiError as e:
            logger.error("Error mentioning: %s", e)
            raise e
        return

    try:
        result = client.chat_postMessage(
            channel=channel_id,
            thread_ts=thread_ts,
            text="<" + "@" + slack_user_id + ">" + " is on call.",
        )
        logger.info(result)

    except errors.SlackApiError as e:
        logger.error("Error mentioning: %s", e)
        raise e
    return


def _get_on_call_schedule_from_slack_channel_name(channel_name: str) -> str | None:
    """Get an on call schedule from a Slack channel name."""
    if "conata" in channel_name:
        return "conata_schedule"
    else:
        logger.error("Channel name does not match schedule: %s", channel_name)
        return None


class _AppMentionEvent(pydantic.BaseModel):
    """Slack App Mention Event Schema.

    https://api.slack.com/events/app_mention#app_mention-event__example-event-payloads__standard-app-mention-when-your-app-is-already-in-channel
    """

    type: Literal["app_mention"] = pydantic.Field(description="The type of the event.")
    channel: str = pydantic.Field(description="Channel id.")
    event_ts: str = pydantic.Field(description="Timestamp of the event.")


def _mention_who_is_on_call_in_slack(app_mention_event: _AppMentionEvent) -> None:
    """Mention who is on call in Slack.

    https://api.slack.com/events/app_mention
    """
    channel_name = _get_slack_channel_name_from_channel_id(app_mention_event.channel)
    on_call_schedule = _get_on_call_schedule_from_slack_channel_name(channel_name)
    on_caller_email = _get_who_is_on_call_from_opsgenie(on_call_schedule)
    slack_user_id = _get_user_id_from_email_address(on_caller_email)
    _post_slack_message(
        app_mention_event.channel, app_mention_event.event_ts, slack_user_id
    )


def lambda_handler(
    event: events.APIGatewayProxyEventV2, _context: context_.Context
) -> responses.APIGatewayProxyResponseV2:
    """Entry point of lambda.

    Args:
        event (events.APIGatewayProxyEventV2): Event object (https://docs.aws.amazon.com/lambda/latest/dg/gettingstarted-concepts.html#gettingstarted-concepts-event).
            An event is a JSON-formatted document that contains data for a Lambda function to process.
            The Lambda runtime converts the event to an object and passes it to your function code. It is usually of the Python dict type. It can also be list, str, int, float, or the NoneType type.
        _context (context_.Context): Context object (https://docs.aws.amazon.com/lambda/latest/dg/python-context.html).
            A context object is passed to your function by Lambda at runtime. This object provides methods and properties that provide information about the invocation, function, and runtime environment.
    """
    # Suppress Slack's retrying
    # https://api.slack.com/apis/connections/events-api#retries
    request_headers: dict[str, str] = not_none(event.get("headers"))
    slack_retry_num = request_headers.get("X-Slack-Retry-Num")
    if slack_retry_num is not None and int(slack_retry_num) >= 1:
        return {"statusCode": 200}
    body = json.loads(not_none(event.get("body")))
    # Handle URL verification handshake
    # https://api.slack.com/apis/connections/events-api#handshake
    challenge: str | None = body.get("challenge")
    if challenge is not None:
        return {
            "statusCode": 200,
            "headers": {"Content-Type": "text/plain"},
            "body": challenge,
        }
    app_mention_event = _AppMentionEvent.parse_obj(body.get("event"))
    _mention_who_is_on_call_in_slack(app_mention_event)
    return {
        "statusCode": 200,
    }