integrations.slack.api

  1"""
  2integrations/slack/api.py
  3"""
  4
  5import logging
  6import textwrap
  7from pathlib import PosixPath
  8from typing import TYPE_CHECKING, cast
  9
 10import pandas as pd
 11
 12from integrations.base.interface import APIInterface
 13from libs.types import StyleOptions
 14from libs.utils import converter, formatter
 15
 16if TYPE_CHECKING:
 17    from slack_sdk.web import SlackResponse
 18    from slack_sdk.web.client import WebClient
 19
 20    from integrations.protocols import MessageParserProtocol
 21
 22
 23class AdapterAPI(APIInterface):
 24    """インターフェースAPI操作クラス"""
 25
 26    # slack object
 27    appclient: "WebClient"
 28    """WebClient(botトークン使用)"""
 29    webclient: "WebClient"
 30    """WebClient(userトークン使用)"""
 31
 32    def __init__(self):
 33        super().__init__()
 34
 35        try:
 36            from slack_sdk.errors import SlackApiError
 37            self.slack_api_error = SlackApiError
 38        except ModuleNotFoundError as err:
 39            raise ModuleNotFoundError(err.msg) from None
 40
 41    def post(self, m: "MessageParserProtocol"):
 42        """メッセージをポストする
 43
 44        Args:
 45            m (MessageParserProtocol): メッセージデータ
 46        """
 47
 48        def _header_text(title: str) -> str:
 49            if not title.isnumeric() and title:  # 数値のキーはヘッダにしない
 50                return f"*【{title}】*\n"
 51            return ""
 52
 53        def _table_data(data: dict) -> list:
 54            ret_list: list = []
 55            text_data = iter(data.values())
 56            # 先頭ブロックの処理(ヘッダ追加)
 57            v = next(text_data)
 58            ret_list.append(f"{header}```\n{v}\n```\n" if style.codeblock else f"{header}{v}\n")
 59            # 残りのブロック
 60            for v in text_data:
 61                ret_list.append(f"```\n{v}\n```\n" if style.codeblock else f"```\n{v}\n```\n")
 62            return ret_list
 63
 64        if not m.in_thread:
 65            m.post.thread = False
 66
 67        # 見出しポスト
 68        header_title = ""
 69        header_text = ""
 70        if m.post.headline:
 71            header_title, header_text = next(iter(m.post.headline.items()))
 72            if not all(v["options"].header_hidden for x in m.post.message for _, v in x.items()):
 73                res = self._call_chat_post_message(
 74                    channel=m.data.channel_id,
 75                    text=f"{_header_text(header_title)}{header_text.rstrip()}",
 76                    thread_ts=m.reply_ts,
 77                )
 78                if res.status_code == 200:  # 見出しがある場合はスレッドにする
 79                    m.post.ts = res.get("ts", "undetermined")
 80                else:
 81                    m.post.ts = "undetermined"
 82
 83        # 本文
 84        post_msg: list[str] = []
 85        style = StyleOptions()
 86        for data in m.post.message:
 87            for title, val in data.items():
 88                msg = val.get("data")
 89                style = val.get("options", StyleOptions())
 90                header = ""
 91
 92                if isinstance(msg, PosixPath) and msg.exists():
 93                    comment = textwrap.dedent(
 94                        f"{_header_text(header_title)}{header_text.rstrip()}"
 95                    ) if style.use_comment else ""
 96                    self._call_files_upload(
 97                        channel=m.data.channel_id,
 98                        title=title,
 99                        file=str(msg),
100                        initial_comment=comment,
101                        thread_ts=m.reply_ts,
102                        request_file_info=False,
103                    )
104
105                if isinstance(msg, str):
106                    if style.key_title and (title != header_title):
107                        header = _header_text(title)
108                    post_msg.append(
109                        f"{header}```\n{msg.rstrip()}\n```\n" if style.codeblock else f"{header}{msg.rstrip()}\n"
110                    )
111
112                if isinstance(msg, pd.DataFrame):
113                    if style.key_title and (title != header_title):
114                        header = _header_text(title)
115
116                    match m.status.command_type:
117                        case "results":
118                            match title:
119                                case "通算ポイント" | "ポイント差分":
120                                    post_msg.extend(_table_data(converter.df_to_text_table(msg, step=40)))
121                                case "役満和了" | "卓外ポイント" | "その他":
122                                    if "回数" in msg.columns:
123                                        post_msg.extend(_table_data(converter.df_to_count(msg, title, 1)))
124                                    else:
125                                        post_msg.extend(_table_data(converter.df_to_remarks(msg)))
126                                case "座席データ":
127                                    post_msg.extend(_table_data(converter.df_to_seat_data(msg, 1)))
128                                case "戦績":
129                                    if "東家 名前" in msg.columns:  # 縦持ちデータ
130                                        post_msg.extend(_table_data(converter.df_to_results_details(msg)))
131                                    else:
132                                        post_msg.extend(_table_data(converter.df_to_results_simple(msg)))
133                                case _:
134                                    post_msg.extend(_table_data(converter.df_to_remarks(msg)))
135                        case "rating":
136                            post_msg.extend(_table_data(converter.df_to_text_table(msg, step=20)))
137                        case "ranking":
138                            post_msg.extend(_table_data(converter.df_to_ranking(msg, title, step=50)))
139
140        if style.summarize:
141            post_msg = formatter.group_strings(post_msg)
142
143        for msg in post_msg:
144            self._call_chat_post_message(
145                channel=m.data.channel_id,
146                text=msg,
147                thread_ts=m.reply_ts,
148            )
149
150    def _call_chat_post_message(self, **kwargs) -> "SlackResponse":
151        """slackにメッセージをポストする
152
153        Returns:
154            SlackResponse: API response
155        """
156
157        res = cast("SlackResponse", {})
158        if kwargs["thread_ts"] == "0":
159            kwargs.pop("thread_ts")
160
161        try:
162            res = self.appclient.chat_postMessage(**kwargs)
163        except self.slack_api_error as err:
164            logging.critical(err)
165            logging.error("kwargs=%s", kwargs)
166
167        return res
168
169    def _call_files_upload(self, **kwargs) -> "SlackResponse":
170        """slackにファイルをアップロードする
171
172        Returns:
173            SlackResponse | Any: API response
174        """
175
176        res = cast("SlackResponse", {})
177        if kwargs.get("thread_ts", "0") == "0":
178            kwargs.pop("thread_ts")
179
180        try:
181            res = self.appclient.files_upload_v2(**kwargs)
182        except self.slack_api_error as err:
183            logging.critical(err)
184            logging.error("kwargs=%s", kwargs)
185
186        return res