integrations.slack.parser

  1"""
  2integrations/slack/parser.py
  3"""
  4
  5import logging
  6from typing import TYPE_CHECKING, cast
  7
  8import libs.global_value as g
  9from integrations.base.interface import (MessageParserDataMixin,
 10                                         MessageParserInterface)
 11from integrations.protocols import MsgData, PostData, StatusData
 12
 13if TYPE_CHECKING:
 14    from integrations.slack.adapter import ServiceAdapter
 15
 16
 17class MessageParser(MessageParserDataMixin, MessageParserInterface):
 18    """メッセージ解析クラス"""
 19
 20    def __init__(self):
 21        MessageParserDataMixin.__init__(self)
 22        self.data: MsgData = MsgData()
 23        self.post: PostData = PostData()
 24        self.status: StatusData = StatusData()
 25
 26    def parser(self, _body: dict):
 27        g.adapter = cast("ServiceAdapter", g.adapter)
 28        # 対象のevent抽出
 29        _event = cast(dict, _body.get("event", _body))
 30
 31        if _body.get("command") == g.adapter.conf.slash_command:  # スラッシュコマンド
 32            self.status.command_flg = True
 33            self.status.command_name = g.adapter.conf.slash_command
 34            if _body.get("channel_name") == "directmessage":
 35                self.data.channel_type = "im"
 36                self.data.status = "message_append"
 37                self.data.channel_id = _body.get("channel_id", "")
 38            else:
 39                self.data.channel_id = g.adapter.functions.get_dm_channel_id(_body.get("user_id", ""))
 40        elif _body.get("container"):  # Homeタブ
 41            self.data.user_id = _body["user"].get("id")
 42            self.data.channel_id = g.adapter.functions.get_dm_channel_id(self.data.user_id)
 43            self.data.channel_type = "channel"
 44            self.data.text = "dummy"
 45        elif _body.get("iid"):  # 検索結果
 46            if (channel_id := str(cast(dict, _event["channel"]).get("id", ""))):
 47                self.data.channel_id = channel_id
 48                _event.pop("channel")
 49            self.data.channel_type = "search_messages"
 50            self.data.status = "message_append"
 51        else:
 52            match _event.get("subtype"):
 53                case "message_changed":
 54                    self.data.status = "message_changed"
 55                    _event.update(cast(dict, _event["message"]))
 56                    if (previous_message := cast(dict, _event.get("previous_message", {}))):
 57                        _event.update(thread_ts=previous_message.get("thread_ts", "0"))
 58                case "message_deleted":
 59                    self.data.status = "message_deleted"
 60                    _event.update(cast(dict, _event["previous_message"]))
 61                case "file_share":
 62                    self.data.status = "message_append"
 63                case None:
 64                    self.data.status = "message_append"
 65                case _:
 66                    self.data.status = "message_append"
 67                    logging.warning("unknown subtype: %s", _body)
 68
 69        self.data.text = _event.get("text", self.data.text)
 70        self.data.channel_id = _event.get("channel", self.data.channel_id)
 71        self.data.channel_type = _event.get("channel_type", self.data.channel_type)
 72        self.data.user_id = _event.get("user", self.data.user_id)
 73        self.data.event_ts = _event.get("ts", "0")
 74        self.data.thread_ts = _event.get("thread_ts", "0")
 75        self.status.source = f"slack_{self.data.channel_id}"
 76
 77        logging.debug(self.data)
 78
 79    @property
 80    def in_thread(self) -> bool:
 81        """スレッド内のメッセージか判定"""
 82        if self.data.thread_ts == "0":
 83            return False
 84        if self.data.event_ts == self.data.thread_ts:
 85            return False
 86        return True
 87
 88    @property
 89    def is_bot(self) -> bool:
 90        if self.data.user_id == "USLACKBOT":
 91            return True
 92        return False
 93
 94    @property
 95    def check_updatable(self) -> bool:
 96        g.adapter = cast("ServiceAdapter", g.adapter)
 97        ret: bool = False
 98
 99        # 突合処理中はチェック省略
100        if self.status.command_type == "comparison":
101            return True
102
103        if g.adapter.conf.channel_limitations:
104            if self.data.channel_id in g.adapter.conf.channel_limitations:
105                ret = True
106        else:  # リストが空なら全チャンネルが対象
107            match self.data.channel_type:
108                case "channel":  # public channel
109                    ret = True
110                case "group":  # private channel
111                    ret = True
112                case "im":  # direct message
113                    ret = False
114                case "search_messages":
115                    ret = True
116
117        return ret
118
119    @property
120    def ignore_user(self) -> bool:
121        g.adapter = cast("ServiceAdapter", g.adapter)
122        return self.data.user_id in g.adapter.conf.ignore_userid