libs.functions.slack_api

  1"""
  2libs/functions/slack_api.py
  3"""
  4
  5import logging
  6import re
  7from typing import Any, cast
  8
  9from slack_sdk.errors import SlackApiError
 10from slack_sdk.web import SlackResponse
 11
 12import libs.global_value as g
 13from libs.utils import formatter
 14
 15
 16def call_chat_post_message(**kwargs) -> SlackResponse | Any:
 17    """slackにメッセージをポストする
 18
 19    Returns:
 20        SlackResponse | Any: API response
 21    """
 22
 23    res = None
 24    if not kwargs["thread_ts"]:
 25        kwargs.pop("thread_ts")
 26    try:
 27        res = g.app.client.chat_postMessage(**kwargs)
 28    except SlackApiError as e:
 29        logging.critical(e)
 30        logging.error("kwargs=%s", kwargs)
 31        logging.error("msg: %s", vars(g.msg))
 32
 33    return res
 34
 35
 36def call_files_upload(**kwargs) -> SlackResponse | Any:
 37    """slackにファイルをアップロードする
 38
 39    Returns:
 40        SlackResponse | Any: API response
 41    """
 42
 43    res = None
 44    if not kwargs["thread_ts"]:
 45        kwargs.pop("thread_ts")
 46    try:
 47        res = g.app.client.files_upload_v2(**kwargs)
 48    except SlackApiError as e:
 49        logging.critical(e)
 50        logging.error("kwargs=%s", kwargs)
 51        logging.error("msg: %s", vars(g.msg))
 52
 53    return res
 54
 55
 56def post_message(message: str, ts=False) -> SlackResponse | Any:
 57    """chat_postMessageに渡すパラメータを設定
 58
 59    Args:
 60        message (str): ポストするメッセージ
 61        ts (bool, optional): スレッドに返す. Defaults to False.
 62
 63    Returns:
 64        SlackResponse | Any: API response
 65    """
 66
 67    res: dict | Any = {}
 68    if not ts and g.msg.thread_ts:
 69        ts = g.msg.thread_ts
 70
 71    if g.args.testcase:
 72        formatter.debug_out(message)
 73        res["ts"] = 0  # dummy
 74    else:
 75        res = call_chat_post_message(
 76            channel=g.msg.channel_id,
 77            text=f"{message.strip()}",
 78            thread_ts=ts,
 79        )
 80
 81    return res
 82
 83
 84def post_multi_message(msg: dict, ts: bool | None = False, summarize: bool = True) -> None:
 85    """メッセージを分割してポスト
 86
 87    Args:
 88        msg (dict): ポストするメッセージ
 89        ts (bool, optional): スレッドに返す. Defaults to False.
 90        summarize (bool, optional): 可能な限り1つのブロックにまとめる. Defaults to True.
 91    """
 92
 93    if g.args.testcase:
 94        formatter.debug_out("", msg)
 95    else:
 96        if isinstance(msg, dict):
 97            if summarize:  # まとめてポスト
 98                key_list = list(map(str, msg.keys()))
 99                post_msg = msg[key_list[0]]
100                for i in key_list[1:]:
101                    if len((post_msg + msg[i])) < 3800:  # 3800文字を超える直前までまとめる
102                        post_msg += msg[i]
103                    else:
104                        post_message(post_msg, ts)
105                        post_msg = msg[i]
106                post_message(post_msg, ts)
107            else:  # そのままポスト
108                for i in msg.keys():
109                    post_message(msg[i], ts)
110        else:
111            post_message(msg, ts)
112
113
114def post_text(event_ts: str, title: str, msg: str) -> SlackResponse | Any:
115    """コードブロック修飾付きポスト
116
117    Args:
118        event_ts (str): スレッドに返す
119        title (str): タイトル行
120        msg (str): 本文
121
122    Returns:
123        SlackResponse | Any: API response
124    """
125
126    # コードブロック修飾付きポスト
127    if len(re.sub(r"\n+", "\n", f"{msg.strip()}").splitlines()) == 1:
128        res = call_chat_post_message(
129            channel=g.msg.channel_id,
130            text=f"{title}\n{msg.strip()}",
131            thread_ts=event_ts,
132        )
133    else:
134        # ポスト予定のメッセージをstep行単位のブロックに分割
135        step = 50
136        post_msg = []
137        for count in range(int(len(msg.splitlines()) / step) + 1):
138            post_msg.append(
139                "\n".join(msg.splitlines()[count * step:(count + 1) * step])
140            )
141
142        # 最終ブロックがstepの半分以下なら直前のブロックにまとめる
143        if len(post_msg) > 1 and step / 2 > len(post_msg[count].splitlines()):
144            post_msg[count - 1] += "\n" + post_msg.pop(count)
145
146        # ブロック単位でポスト
147        for _, val in enumerate(post_msg):
148            res = call_chat_post_message(
149                channel=g.msg.channel_id,
150                text=f"\n{title}\n\n```{val.strip()}```",
151                thread_ts=event_ts,
152            )
153
154    return res
155
156
157def post_fileupload(title: str, file: str | bool, ts: str | bool = False) -> SlackResponse | None:
158    """files_upload_v2に渡すパラメータを設定
159
160    Args:
161        title (str): タイトル行
162        file (str): アップロードファイルパス
163        ts (str | bool, optional): スレッドに返す. Defaults to False.
164
165    Returns:
166        SlackResponse | None: 結果
167    """
168
169    if g.args.testcase:
170        formatter.debug_out(title, file)
171        return None
172
173    if not ts and g.msg.thread_ts:
174        ts = g.msg.thread_ts
175
176    res = call_files_upload(
177        channel=g.msg.channel_id,
178        title=title,
179        file=file,
180        thread_ts=ts,
181        request_file_info=False,
182    )
183
184    return res
185
186
187def slack_post(**kwargs):
188    """パラメータの内容によって呼び出すAPIを振り分ける"""
189
190    logging.debug(kwargs)
191    headline = str(kwargs.get("headline", ""))
192    message = kwargs.get("message")
193    summarize = bool(kwargs.get("summarize", True))
194    file_list = cast(dict, kwargs.get("file_list", {"dummy": ""}))
195
196    # 見出しポスト
197    if (res := post_message(headline)):
198        ts = res.get("ts", False)
199    else:
200        ts = False
201
202    # 本文ポスト
203    for x in file_list:
204        if (file_path := file_list.get(x)):
205            post_fileupload(str(x), str(file_path), ts)
206            message = {}  # ファイルがあるメッセージは不要
207
208    if message:
209        post_multi_message(message, ts, summarize)
210
211
212def call_reactions_add(icon: str, ch: str | None = None, ts: str | None = None):
213    """リアクションを付ける
214
215    Args:
216        icon (str): 付けるリアクション
217        ch (str | None, optional): チャンネルID. Defaults to None.
218        ts (str | None, optional): メッセージのタイムスタンプ. Defaults to None.
219    """
220
221    if not ch:
222        ch = g.msg.channel_id
223    if not ts:
224        ts = g.msg.event_ts
225
226    try:
227        res: SlackResponse = g.app.client.reactions_add(
228            channel=str(ch),
229            name=icon,
230            timestamp=str(ts),
231        )
232        logging.info("ts=%s, ch=%s, icon=%s, %s", ts, ch, icon, res.validate())
233    except SlackApiError as e:
234        match e.response.get("error"):
235            case "already_reacted":
236                pass
237            case _:
238                logging.critical(e)
239                logging.critical("ts=%s, ch=%s, icon=%s", ts, ch, icon)
240                logging.error("msg: %s", vars(g.msg))
241
242
243def call_reactions_remove(icon: str, ch: str | None = None, ts: str | None = None):
244    """リアクションを外す
245
246    Args:
247        icon (str): 外すリアクション
248        ch (str | None, optional): チャンネルID. Defaults to None.
249        ts (str | None, optional): メッセージのタイムスタンプ. Defaults to None.
250    """
251
252    if not ch:
253        ch = g.msg.channel_id
254    if not ts:
255        ts = g.msg.event_ts
256
257    try:
258        res = g.app.client.reactions_remove(
259            channel=ch,
260            name=icon,
261            timestamp=ts,
262        )
263        logging.info("ts=%s, ch=%s, icon=%s, %s", ts, ch, icon, res.validate())
264    except SlackApiError as e:
265        match e.response.get("error"):
266            case "no_reaction":
267                pass
268            case _:
269                logging.critical(e)
270                logging.critical("ts=%s, ch=%s, icon=%s", ts, ch, icon)
271                logging.error("msg: %s", vars(g.msg))