integrations.standard_io.parser
integrations/standard_io/parser.py
1""" 2integrations/standard_io/parser.py 3""" 4 5from datetime import datetime 6from typing import Any, cast 7 8from integrations.base.interface import MessageParserDataMixin, MessageParserInterface 9from integrations.protocols import MsgData, PostData, StatusData 10from libs.types import ChannelType, MessageStatus 11 12 13class MessageParser(MessageParserDataMixin, MessageParserInterface): 14 """メッセージ解析クラス""" 15 16 def __init__(self) -> None: 17 MessageParserDataMixin.__init__(self) 18 self.data: MsgData = MsgData() 19 self.post: PostData = PostData() 20 self.status: StatusData = StatusData() 21 22 def parser(self, body: dict[str, Any]) -> None: 23 self.data.status = MessageStatus.APPEND 24 self.data.channel_id = "dummy" 25 self.data.event_ts = str(datetime.now().timestamp()) 26 self.data.thread_ts = self.data.event_ts 27 self.status.source = "standard_io" 28 29 if body.get("event"): 30 body = cast(dict[str, Any], body["event"]) 31 32 if body.get("text"): 33 self.data.text = str(body.get("text", "")) 34 else: 35 self.data.text = "" 36 37 if body.get("channel_name") == "directmessage": # スラッシュコマンド扱い 38 self.status.command_flg = True 39 self.data.channel_type = ChannelType.DIRECT_MESSAGE 40 self.data.channel_id = body.get("channel_id", "") 41 42 @property 43 def in_thread(self) -> bool: 44 return False 45 46 @property 47 def is_bot(self) -> bool: 48 return False 49 50 @property 51 def check_updatable(self) -> bool: 52 return True 53 54 @property 55 def ignore_user(self) -> bool: 56 return False
class
MessageParser(integrations.base.interface.MessageParserDataMixin, integrations.base.interface.MessageParserInterface):
14class MessageParser(MessageParserDataMixin, MessageParserInterface): 15 """メッセージ解析クラス""" 16 17 def __init__(self) -> None: 18 MessageParserDataMixin.__init__(self) 19 self.data: MsgData = MsgData() 20 self.post: PostData = PostData() 21 self.status: StatusData = StatusData() 22 23 def parser(self, body: dict[str, Any]) -> None: 24 self.data.status = MessageStatus.APPEND 25 self.data.channel_id = "dummy" 26 self.data.event_ts = str(datetime.now().timestamp()) 27 self.data.thread_ts = self.data.event_ts 28 self.status.source = "standard_io" 29 30 if body.get("event"): 31 body = cast(dict[str, Any], body["event"]) 32 33 if body.get("text"): 34 self.data.text = str(body.get("text", "")) 35 else: 36 self.data.text = "" 37 38 if body.get("channel_name") == "directmessage": # スラッシュコマンド扱い 39 self.status.command_flg = True 40 self.data.channel_type = ChannelType.DIRECT_MESSAGE 41 self.data.channel_id = body.get("channel_id", "") 42 43 @property 44 def in_thread(self) -> bool: 45 return False 46 47 @property 48 def is_bot(self) -> bool: 49 return False 50 51 @property 52 def check_updatable(self) -> bool: 53 return True 54 55 @property 56 def ignore_user(self) -> bool: 57 return False
メッセージ解析クラス
def
parser(self, body: dict[str, typing.Any]) -> None:
23 def parser(self, body: dict[str, Any]) -> None: 24 self.data.status = MessageStatus.APPEND 25 self.data.channel_id = "dummy" 26 self.data.event_ts = str(datetime.now().timestamp()) 27 self.data.thread_ts = self.data.event_ts 28 self.status.source = "standard_io" 29 30 if body.get("event"): 31 body = cast(dict[str, Any], body["event"]) 32 33 if body.get("text"): 34 self.data.text = str(body.get("text", "")) 35 else: 36 self.data.text = "" 37 38 if body.get("channel_name") == "directmessage": # スラッシュコマンド扱い 39 self.status.command_flg = True 40 self.data.channel_type = ChannelType.DIRECT_MESSAGE 41 self.data.channel_id = body.get("channel_id", "")
メッセージ解析
Arguments:
- body (Any): 解析データ