integrations.slack.parser

  1"""
  2integrations/slack/parser.py
  3"""
  4
  5import logging
  6from typing import TYPE_CHECKING, Any, cast
  7
  8import libs.global_value as g
  9from integrations.base.interface import MessageParserDataMixin, MessageParserInterface
 10from integrations.protocols import MsgData, PostData, StatusData
 11from libs.types import ChannelType, CommandType, MessageStatus
 12
 13if TYPE_CHECKING:
 14    from integrations.slack.adapter import ServiceAdapter
 15
 16
 17class MessageParser(MessageParserDataMixin, MessageParserInterface):
 18    """メッセージ解析クラス"""
 19
 20    def __init__(self) -> None:
 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[str, Any]) -> None:
 27        g.adapter = cast("ServiceAdapter", g.adapter)
 28        _event = cast(dict[str, Any], _body.get("event", _body))  # 対象のevent抽出
 29
 30        if _body.get("command") == g.adapter.conf.slash_command:  # スラッシュコマンド
 31            self.status.command_flg = True
 32            self.status.command_name = g.adapter.conf.slash_command
 33            self.data.status = MessageStatus.APPEND
 34            self.data.user_id = str(_body.get("user_id", ""))
 35            if _body.get("channel_name") == "directmessage":
 36                self.data.channel_type = ChannelType.DIRECT_MESSAGE
 37                self.data.channel_id = str(_body.get("channel_id", ""))
 38            else:  # チャンネル内コマンド
 39                if channel_id := str(_body.get("channel_id")):
 40                    self.status.source = f"slack_{channel_id}"
 41                self.data.channel_type = ChannelType.DIRECT_MESSAGE
 42                self.data.channel_id = g.adapter.functions.get_dm_channel_id(self.data.user_id)  # DM Open
 43        elif _body.get("container"):  # Homeタブ
 44            self.data.user_id = str(cast(dict[str, Any], _body["user"]).get("id", ""))
 45            self.data.channel_id = g.adapter.functions.get_dm_channel_id(self.data.user_id)
 46            self.data.channel_type = ChannelType.HOME_APP
 47            self.data.text = "dummy"
 48        elif _body.get("iid"):  # 検索結果
 49            if _channel_id := str(cast(dict[str, Any], _event["channel"]).get("id", "")):
 50                self.data.channel_id = _channel_id
 51                _event.pop("channel")
 52            self.data.channel_type = ChannelType.SEARCH
 53            self.data.status = MessageStatus.APPEND
 54        else:
 55            match _event.get("subtype"):
 56                case subtype if str(subtype).endswith(("_topic", "_purpose", "_name")):
 57                    self.data.status = MessageStatus.DO_NOTHING
 58                    return
 59                case "message_changed":
 60                    self.data.status = MessageStatus.CHANGED
 61                    _event.update(cast(dict[str, Any], _event["message"]))
 62                    if cast(dict[str, Any], _event["message"]).get("subtype") == "tombstone":  # スレッド元の削除
 63                        self.data.status = MessageStatus.DELETED
 64                    elif _edited := cast(dict[str, Any], _event["message"]).get("edited"):
 65                        self.data.edited_ts = str(cast(dict[str, Any], _edited).get("ts", "undetermined"))
 66                    if _previous_message := cast(dict[str, Any], _event.get("previous_message", {})):
 67                        if _previous_message.get("thread_ts"):
 68                            _event.update(thread_ts=_previous_message.get("thread_ts", "0"))
 69                        if cast(dict[str, Any], _event["message"]).get("text") == _previous_message.get("text"):
 70                            self.data.status = MessageStatus.DO_NOTHING
 71                case "message_deleted":
 72                    self.data.status = MessageStatus.DELETED
 73                    _event.update(cast(dict[str, Any], _event["previous_message"]))
 74                case "file_share" | "thread_broadcast":
 75                    self.data.status = MessageStatus.APPEND
 76                case None:
 77                    self.data.status = MessageStatus.APPEND
 78                case _:
 79                    self.data.status = MessageStatus.APPEND
 80                    logging.warning("unknown subtype: %s", _body)
 81
 82        match _event.get("channel_type"):
 83            case "directmessage" | "im":
 84                self.data.channel_type = ChannelType.DIRECT_MESSAGE
 85            case "channel":
 86                self.data.channel_type = ChannelType.CHANNEL
 87            case "group":
 88                self.data.channel_type = ChannelType.PRIVATE
 89            case _:
 90                pass
 91
 92        self.data.text = _event.get("text", self.data.text)
 93        self.data.channel_id = _event.get("channel", self.data.channel_id)
 94        self.data.user_id = _event.get("user", self.data.user_id)
 95        self.data.event_ts = _event.get("ts", "0")
 96        self.data.thread_ts = _event.get("thread_ts", "0")
 97        if not self.status.source:
 98            self.status.source = f"slack_{self.data.channel_id}"
 99
100        logging.debug(self.data)
101
102    @property
103    def in_thread(self) -> bool:
104        """スレッド内のメッセージか判定"""
105        if self.data.thread_ts == "0":
106            return False
107        if self.data.event_ts == self.data.thread_ts:
108            return False
109        return True
110
111    @property
112    def is_bot(self) -> bool:
113        if self.data.user_id == "USLACKBOT":
114            return True
115        return False
116
117    @property
118    def check_updatable(self) -> bool:
119        g.adapter = cast("ServiceAdapter", g.adapter)
120        ret: bool = False
121
122        # 突合処理中はチェック省略
123        if self.status.command_type == CommandType.COMPARISON:
124            return True
125
126        if g.adapter.conf.channel_limitations:
127            if self.data.channel_id in g.adapter.conf.channel_limitations:
128                ret = True
129        else:  # リストが空なら全チャンネルが対象
130            match self.data.channel_type:
131                case ChannelType.CHANNEL:  # public channel
132                    ret = True
133                case ChannelType.PRIVATE:  # private channel
134                    ret = True
135                case ChannelType.HOME_APP:  # home app
136                    ret = True
137                case ChannelType.DIRECT_MESSAGE:  # direct message
138                    ret = False
139                case ChannelType.SEARCH:
140                    ret = True
141                case _:
142                    pass
143
144        return ret
145
146    @property
147    def ignore_user(self) -> bool:
148        g.adapter = cast("ServiceAdapter", g.adapter)
149        return self.data.user_id in g.adapter.conf.ignore_userid