Module slack_bolt.listener_matcher.builtins

def action(
    constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    if isinstance(constraints, (str, Pattern)):

        def func(body: Dict[str, Any]) -> bool:
            return (
                _block_action(constraints, body)
                or _attachment_action(constraints, body)
                or _dialog_submission(constraints, body)
                or _dialog_cancellation(constraints, body)
                or _workflow_step_edit(constraints, body)
            )

        return build_listener_matcher(func, asyncio, base_logger)

    elif "type" in constraints:
        action_type = constraints["type"]
        if action_type == "block_actions":
            return block_action(constraints, asyncio)
        if action_type == "interactive_message":
            return attachment_action(constraints["callback_id"], asyncio)
        if action_type == "dialog_submission":
            return dialog_submission(constraints["callback_id"], asyncio)
        if action_type == "dialog_cancellation":
            return dialog_cancellation(constraints["callback_id"], asyncio)
        # https://api.slack.com/workflows/steps
        if action_type == "workflow_step_edit":
            return workflow_step_edit(constraints["callback_id"], asyncio)

        raise BoltError(f"type: {action_type} is unsupported")
    elif "action_id" in constraints or "block_id" in constraints:
        # The default value is "block_actions"
        return block_action(constraints, asyncio)

    raise BoltError(f"action ({constraints}: {type(constraints)}) must be any of str, Pattern, and dict")

def attachment_action(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return _attachment_action(callback_id, body)

    return build_listener_matcher(func, asyncio, base_logger)

def block_action(
    constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return _block_action(constraints, body)

    return build_listener_matcher(func, asyncio, base_logger)

def block_suggestion(
    action_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return _block_suggestion(action_id, body)

    return build_listener_matcher(func, asyncio, base_logger)

def build_listener_matcher(
    func: Callable[..., bool],
    asyncio: bool,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    if asyncio:
        from .async_builtins import AsyncBuiltinListenerMatcher

        async def async_fun(body: Dict[str, Any]) -> bool:
            return func(body)

        return AsyncBuiltinListenerMatcher(func=async_fun, base_logger=base_logger)
    else:
        return BuiltinListenerMatcher(func=func, base_logger=base_logger)

def command(
    command: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return is_slash_command(body) and _matches(command, body["command"])

    return build_listener_matcher(func, asyncio, base_logger)

def dialog_cancellation(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return _dialog_cancellation(callback_id, body)

    return build_listener_matcher(func, asyncio, base_logger)

def dialog_submission(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return _dialog_submission(callback_id, body)

    return build_listener_matcher(func, asyncio, base_logger)

def dialog_suggestion(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return _dialog_suggestion(callback_id, body)

    return build_listener_matcher(func, asyncio, base_logger)

def event(
    constraints: Union[
        str,
        Pattern,
        Dict[str, Optional[Union[str, Sequence[Optional[Union[str, Pattern]]]]]],
    ],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    if isinstance(constraints, (str, Pattern)):
        event_type: Union[str, Pattern] = constraints
        _verify_message_event_type(event_type)

        def func(body: Dict[str, Any]) -> bool:
            return is_event(body) and _matches(event_type, body["event"]["type"])

        return build_listener_matcher(func, asyncio, base_logger)

    elif "type" in constraints:
        _verify_message_event_type(constraints["type"])  # type: ignore[arg-type]

        def func(body: Dict[str, Any]) -> bool:
            if is_event(body):
                return _check_event_subtype(
                    event_payload=body["event"],
                    constraints=constraints,
                )
            return False

        return build_listener_matcher(func, asyncio, base_logger)

    raise BoltError(f"event ({constraints}: {type(constraints)}) must be any of str, Pattern, and dict")

def function_executed(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return is_function(body) and _matches(callback_id, body.get("event", {}).get("function", {}).get("callback_id", ""))

    return build_listener_matcher(func, asyncio, base_logger)

def global_shortcut(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return is_global_shortcut(body) and _matches(callback_id, body["callback_id"])

    return build_listener_matcher(func, asyncio, base_logger)

def message_event(
    constraints: Dict[str, Optional[Union[str, Sequence[Optional[Union[str, Pattern]]]]]],
    keyword: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    if "type" in constraints and keyword is not None:
        _verify_message_event_type(constraints["type"])  # type: ignore[arg-type]

        def func(body: Dict[str, Any]) -> bool:
            if is_event(body):
                is_valid_subtype = _check_event_subtype(
                    event_payload=body["event"],
                    constraints=constraints,
                )
                if is_valid_subtype is True:
                    # Check keyword matching
                    text = body.get("event", {}).get("text", "")
                    match_result = re.findall(keyword, text)
                    if match_result is not None and match_result != []:
                        return True
            return False

        return build_listener_matcher(func, asyncio, base_logger)

    raise BoltError(f"event ({constraints}: {type(constraints)}) must be dict")

def message_shortcut(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return is_message_shortcut(body) and _matches(callback_id, body["callback_id"])

    return build_listener_matcher(func, asyncio, base_logger)

def options(
    constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    if isinstance(constraints, (str, Pattern)):

        def func(body: Dict[str, Any]) -> bool:
            return _block_suggestion(constraints, body) or _dialog_suggestion(constraints, body)

        return build_listener_matcher(func, asyncio, base_logger)

    if "action_id" in constraints:
        return block_suggestion(constraints["action_id"], asyncio)
    if "callback_id" in constraints:
        return dialog_suggestion(constraints["callback_id"], asyncio)
    else:
        raise BoltError(f"options ({constraints}: {type(constraints)}) must be any of str, Pattern, and dict")

def shortcut(
    constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    if isinstance(constraints, (str, Pattern)):
        callback_id: Union[str, Pattern] = constraints

        def func(body: Dict[str, Any]) -> bool:
            return is_shortcut(body) and _matches(callback_id, body["callback_id"])

        return build_listener_matcher(func, asyncio, base_logger)

    elif "type" in constraints and "callback_id" in constraints:
        if constraints["type"] == "shortcut":
            return global_shortcut(constraints["callback_id"], asyncio)
        if constraints["type"] == "message_action":
            return message_shortcut(constraints["callback_id"], asyncio)

    raise BoltError(f"shortcut ({constraints}: {type(constraints)}) must be any of str, Pattern, and dict")

def view(
    constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    if isinstance(constraints, (str, Pattern)):
        return view_submission(constraints, asyncio)
    elif "type" in constraints:
        if constraints["type"] == "view_submission":
            return view_submission(constraints["callback_id"], asyncio)
        if constraints["type"] == "view_closed":
            return view_closed(constraints["callback_id"], asyncio)

    raise BoltError(f"view ({constraints}: {type(constraints)}) must be any of str, Pattern, and dict")

def view_closed(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return is_view_closed(body) and _matches(callback_id, body["view"]["callback_id"])

    return build_listener_matcher(func, asyncio, base_logger)

def view_submission(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return is_view_submission(body) and _matches(callback_id, body["view"]["callback_id"])

    return build_listener_matcher(func, asyncio, base_logger)

def workflow_step_edit(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return _workflow_step_edit(callback_id, body)

    return build_listener_matcher(func, asyncio, base_logger)

def workflow_step_execute(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return (
            is_event(body)
            and _matches("workflow_step_execute", body["event"]["type"])
            and "workflow_step" in body["event"]
            and _matches(callback_id, body["event"]["callback_id"])
        )

    return build_listener_matcher(func, asyncio, base_logger)

def workflow_step_save(
    callback_id: Union[str, Pattern],
    asyncio: bool = False,
    base_logger: Optional[Logger] = None,
) -> Union[ListenerMatcher, "AsyncListenerMatcher"]:  # type: ignore[name-defined]
    def func(body: Dict[str, Any]) -> bool:
        return is_workflow_step_save(body) and _matches(callback_id, body["view"]["callback_id"])

    return build_listener_matcher(func, asyncio, base_logger)

class BuiltinListenerMatcher(ListenerMatcher):
    def __init__(
        self,
        *,
        func: Callable[..., Union[bool, Awaitable[bool]]],
        base_logger: Optional[Logger] = None,
    ):
        self.func = func
        self.arg_names = get_arg_names_of_callable(func)
        self.logger = get_bolt_logger(self.func, base_logger)

    def matches(self, req: BoltRequest, resp: BoltResponse) -> bool:
        return self.func(  # type: ignore[return-value]
            **build_required_kwargs(
                logger=self.logger,
                required_arg_names=self.arg_names,
                request=req,
                response=resp,
                this_func=self.func,
            )
        )