integrations.discord.functions

  1"""
  2integrations/discord/functions.py
  3"""
  4
  5from typing import TYPE_CHECKING, Any, cast
  6
  7from discord import Forbidden, NotFound
  8from discord.channel import TextChannel
  9
 10from integrations.base.interface import FunctionsInterface
 11from libs.types import ActionStatus
 12from libs.utils.timekit import ExtendedDatetime as ExtDt
 13
 14if TYPE_CHECKING:
 15    from discord import ClientUser, Message
 16
 17    from integrations.discord.api import AdapterAPI
 18    from integrations.discord.config import SvcConfig
 19    from integrations.discord.parser import MessageParser
 20    from integrations.protocols import MessageParserProtocol
 21
 22
 23class SvcFunctions(FunctionsInterface):
 24    """discord専用関数"""
 25
 26    def __init__(self, api: "AdapterAPI", conf: "SvcConfig") -> None:
 27        super().__init__()
 28
 29        self.api = api
 30        self.conf = conf
 31        """個別設定"""
 32
 33    def post_processing(self, m: "MessageParserProtocol") -> None:
 34        """
 35        後処理（非同期処理ラッパー）
 36
 37        Args:
 38            m (MessageParserProtocol): メッセージデータ
 39
 40        """
 41        match m.status.action:
 42            case ActionStatus.NOTHING:
 43                return
 44            case ActionStatus.CHANGE:
 45                self.api.bot.loop.create_task(self.update_reaction(m))
 46            case ActionStatus.DELETE:
 47                self.api.bot.loop.create_task(self.delete_reaction(m))
 48
 49    async def update_reaction(self, m: "MessageParserProtocol") -> None:
 50        """
 51        後処理
 52
 53        Args:
 54            m (MessageParserProtocol): メッセージデータ
 55
 56        """
 57        m = cast("MessageParser", m)
 58        self.conf.bot_name = cast("ClientUser", self.conf.bot_name)
 59
 60        if not hasattr(m, "discord_msg"):
 61            return
 62        if await self.is_deleted_message(m.discord_msg):
 63            return
 64
 65        emoji = {
 66            "ok": "\N{WHITE HEAVY CHECK MARK}",
 67            "ng": "\N{CROSS MARK}",
 68        }
 69
 70        # 既に付いているリアクションを取得
 71        has_ok = False
 72        has_ng = False
 73        for reaction in m.discord_msg.reactions:
 74            if str(reaction.emoji) == emoji["ok"]:
 75                async for user in reaction.users():
 76                    if user == self.conf.bot_name:
 77                        has_ok = True
 78                        continue
 79            if str(reaction.emoji) == emoji["ng"]:
 80                async for user in reaction.users():
 81                    if user == self.conf.bot_name:
 82                        has_ng = True
 83                        continue
 84
 85        # リアクション処理
 86        if m.status.reaction:  # NGを外してOKを付ける
 87            if has_ng:
 88                await m.discord_msg.remove_reaction(emoji["ng"], self.conf.bot_name)
 89            if not has_ok:
 90                await m.discord_msg.add_reaction(emoji["ok"])
 91        else:  # OKを外してNGを付ける
 92            if has_ok:
 93                await m.discord_msg.remove_reaction(emoji["ok"], self.conf.bot_name)
 94            if not has_ng:
 95                await m.discord_msg.add_reaction(emoji["ng"])
 96
 97    async def delete_reaction(self, m: "MessageParserProtocol") -> None:
 98        """
 99        botが付けたリアクションをすべて削除する
100
101        Args:
102            m (MessageParserProtocol): メッセージデータ
103
104        """
105        m = cast("MessageParser", m)
106
107        if hasattr(m, "discord_msg"):
108            if await self.is_deleted_message(m.discord_msg):
109                return  # 削除済み
110        else:  # Messageを持っていない場合はevent_tsとchannel_idを使って検索を試みる
111            if not m.data.channel_id:
112                return  # チャンネルIDが空
113            channel = self.api.bot.get_channel(int(m.data.channel_id))
114            if not isinstance(channel, TextChannel):
115                return  # 対象外チャンネル
116
117            async for msg in channel.history(
118                limit=10,
119                oldest_first=True,
120                after=ExtDt(float(m.data.event_ts), seconds=-1).dt,
121                before=ExtDt(float(m.data.event_ts), seconds=1).dt,
122            ):
123                if str(msg.created_at.timestamp()) == m.data.event_ts:
124                    m.discord_msg = msg
125                    break
126                return  # 該当メッセージなし(削除済み)
127
128        for reaction in m.discord_msg.reactions:
129            async for user in reaction.users():
130                if user == self.conf.bot_name:
131                    await reaction.remove(user)
132
133    async def is_deleted_message(self, message: "Message") -> bool:
134        """
135        メッセージが削除済みか調べる
136
137        Args:
138            message (Message): discordオブジェクト
139
140        Returns:
141            bool: 真偽
142
143        """
144        try:
145            await message.channel.fetch_message(message.id)
146            return False
147        except NotFound:
148            return True
149        except Forbidden:
150            return True
151
152    def get_conversations(self, m: "MessageParserProtocol") -> dict[str, Any]:
153        _ = m
154        return {}