integrations.slack.events.handler

  1"""
  2integrations/slack/events/handler.py
  3"""
  4
  5import logging
  6import os
  7import sys
  8from typing import TYPE_CHECKING, Any, cast
  9
 10import libs.dispatcher
 11from integrations.slack.events.handler_registry import register, register_all
 12from integrations.slack.events.home_tab import home
 13from libs.utils.timekit import Delimiter, Format
 14from libs.utils.timekit import ExtendedDatetime as ExtDt
 15
 16if TYPE_CHECKING:
 17    from slack_bolt import App
 18
 19    from integrations.protocols import MessageParserProtocol
 20    from integrations.slack.adapter import ServiceAdapter
 21
 22
 23def main(adapter: "ServiceAdapter") -> None:
 24    """
 25    メイン処理
 26
 27    Args:
 28        adapter (ServiceAdapter): アダプタインターフェース
 29
 30    Raises:
 31        ModuleNotFoundError: ライブラリ未インストール
 32
 33    """
 34    try:
 35        from slack_bolt import App
 36        from slack_bolt.adapter.socket_mode import SocketModeHandler
 37        from slack_sdk import WebClient
 38        from slack_sdk.errors import SlackApiError
 39    except ModuleNotFoundError as err:
 40        raise ModuleNotFoundError(err.msg) from None
 41
 42    def log_filter() -> None:
 43        """ログレベル変更"""
 44        for name in logging.Logger.manager.loggerDict:
 45            if name.startswith(("slack_", "slack")) or "socket_mode" in name:
 46                logging.getLogger(name).setLevel(logging.WARNING)
 47
 48    try:
 49        log_filter()
 50        app = App(token=os.environ["SLACK_BOT_TOKEN"])
 51        adapter.api.webclient = WebClient(token=os.environ["SLACK_WEB_TOKEN"])
 52        adapter.api.appclient = app.client
 53        log_filter()
 54        adapter.conf.bot_id = app.client.auth_test()["user_id"]
 55    except SlackApiError as err:
 56        logging.critical(err)
 57        sys.exit()
 58
 59    register_all(app, adapter)  # イベント遅延登録
 60    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()  # type: ignore[no-untyped-call]
 61
 62
 63@register
 64def register_event_handlers(app: "App", adapter: "ServiceAdapter") -> None:
 65    """イベントAPI"""
 66    m = cast("MessageParserProtocol", adapter.parser())
 67
 68    @app.event("message")
 69    def handle_message_events(body: Any) -> None:
 70        """
 71        メッセージイベント
 72
 73        Args:
 74            body (Any): ポストされたデータ
 75
 76        """
 77        m.reset()
 78        m.parser(body)
 79        libs.dispatcher.by_keyword(m)
 80
 81    @app.command(adapter.conf.slash_command)
 82    def slash_command(ack: Any, body: Any) -> None:
 83        """
 84        スラッシュコマンド
 85
 86        Args:
 87            ack (Any): ack
 88            body (Any): ポストされたデータ
 89
 90        """
 91        ack()
 92        m.reset()
 93        m.parser(body)
 94        libs.dispatcher.by_keyword(m)
 95
 96    @app.event("app_home_opened")
 97    def handle_home_events(event: Any) -> None:
 98        """
 99        ホームタブオープン
100
101        Args:
102            event (Any): イベント内容
103
104        """
105        adapter.conf.tab_var = {
106            "view": {},
107            "no": 0,
108            "user_id": None,
109            "view_id": None,
110            "screen": None,
111            "operation": None,
112            "sday": adapter.conf.tab_var.get("sday", ExtDt().format(Format.YMD, Delimiter.HYPHEN)),
113            "eday": adapter.conf.tab_var.get("eday", ExtDt().format(Format.YMD, Delimiter.HYPHEN)),
114        }
115
116        adapter.conf.tab_var["user_id"] = event["user"]
117        if "view" in event:
118            adapter.conf.tab_var["view_id"] = event["view"]["id"]
119
120        logging.trace(adapter.conf.tab_var)  # type: ignore
121
122        home.build_main_menu(adapter)
123        result = adapter.api.appclient.views_publish(
124            user_id=adapter.conf.tab_var["user_id"],
125            view=adapter.conf.tab_var["view"],
126        )
127        logging.trace(result)  # type: ignore