class_docstr: Slack input channel implementation. Based on the HTTPInputChannel.
class_docstr_tok: ["Slack","input","channel","implementation",".","Based","on","the","HTTPInputChannel","."]
code: def blueprint(
      self, on_new_message: Callable[[UserMessage], Awaitable[Any]]
  ) -> Blueprint:
      slack_webhook = Blueprint("slack_webhook", __name__)
  
      @slack_webhook.route("/", methods=["GET"])
      async def health(_: Request) -> HTTPResponse:
          return response.json({"status": "ok"})
  
      @slack_webhook.route("/webhook", methods=["GET", "POST"])
      async def webhook(request: Request) -> HTTPResponse:
          content_type = request.headers.get("content-type")
  
          if not self.is_request_from_slack_authentic(request):
              return response.text(
                  "Message is not properly signed with a valid "
                  "X-Slack-Signature header",
                  status=HTTPStatus.BAD_REQUEST,
              )
  
          # Slack API sends either a JSON-encoded or a URL-encoded body
          # depending on the content
  
          if content_type == "application/json":
              # if JSON-encoded message is received
              output = request.json
              event = output.get("event", {})
              user_message = event.get("text", "")
              sender_id = event.get("user", "")
              metadata = self.get_metadata(request)
              channel_id = metadata.get("out_channel")
              thread_id = metadata.get("thread_id")
              conversation_id = self._get_conversation_id(
                  sender_id, channel_id, thread_id
              )
  
              if "challenge" in output:
                  return response.json(output.get("challenge"))
  
              if not self._is_user_message(output):
                  logger.debug(
                      "Received message from Slack which doesn't look like "
                      "a user message. Skipping message."
                  )
                  return response.text("Bot message delivered.")
  
              if not self._is_supported_channel(output, metadata):
                  logger.warning(
                      f"Received message on unsupported "
                      f"channel: {metadata['out_channel']}"
                  )
                  return response.text("channel not supported.")
  
              return await self.process_message(
                  request,
                  on_new_message,
                  text=self._sanitize_user_message(
                      user_message, metadata["users"]
                  ),
                  sender_id=conversation_id,
                  metadata=metadata,
              )
          elif content_type == "application/x-www-form-urlencoded":
              # if URL-encoded message is received
              output = request.form
              payload = json.loads(output["payload"][0])
  
              if self._is_interactive_message(payload):
                  sender_id = payload["user"]["id"]
                  text = self._get_interactive_response(
                      payload["actions"][0]
                  )
                  if text is not None:
                      metadata = self.get_metadata(request)
                      channel_id = metadata.get("out_channel")
                      thread_id = metadata.get("thread_id")
                      conversation_id = self._get_conversation_id(
                          sender_id, channel_id, thread_id
                      )
                      return await self.process_message(
                          request,
                          on_new_message,
                          text,
                          conversation_id,
                          metadata,
                      )
                  if payload["actions"][0]["type"] == "button":
                      # link buttons don't have "value", don't send their clicks to
                      # bot
                      return response.text("User clicked link button")
              return response.text(
                  "The input message could not be processed.",
                  status=HTTPStatus.INTERNAL_SERVER_ERROR,
              )
  
          return response.text("Bot message delivered.")
  
      return slack_webhook
  
imports: import asyncio
  import hashlib
  import hmac
  from http import HTTPStatus
  import json
  import logging
  import re
  import time
  from typing import Any, Awaitable, Callable, Dict, List, Optional, Set, Text
  from rasa.core.channels.channel import InputChannel, OutputChannel, UserMessage
  from rasa.shared.constants import DOCS_URL_CONNECTORS_SLACK
  from rasa.shared.exceptions import InvalidConfigException
  import rasa.shared.utils.io
  from sanic import Blueprint, response
  from sanic.request import Request
  from sanic.response import HTTPResponse
  from slack_sdk.web.async_client import AsyncWebClient
label_desc: Use image node_id 13 for calling the SlackInput obj's underlying member method code with example usage: obj.blueprint(on_new_message) and returns: slack_webhook, response, response, response, unknown, response, response, response, response, unknown, response