Source code for slidge.plugins.discord.util

from typing import TYPE_CHECKING, Optional, Union

import discord as di

from slidge.core.mixins.message import ContentMessageMixin
from slidge.core.mixins.presence import PresenceMixin
from slidge.util.types import LegacyAttachment, MessageReference

from ...util import strip_illegal_chars

    from .group import MUC
    from .session import Session

[docs]class MessageMixin(ContentMessageMixin):
[docs] session: "Session"
[docs] legacy_id: int # type:ignore
[docs] avatar: str
[docs] discord_user: Union[di.User, di.ClientUser] # type: ignore
[docs] MARKS = False
[docs] async def update_reactions(self, m: di.Message): legacy_reactions = [] user = self.discord_user for r in m.reactions: if r.is_custom_emoji(): continue assert isinstance(r.emoji, str) try: async for u in r.users(): if == legacy_reactions.append(r.emoji) except di.NotFound: # the message has now been deleted # seems to happen quite a lot. I guess # there are moderation bot that are triggered # by reactions from users # oh, discord… return self.react(, legacy_reactions)
[docs] async def _reply_to(self, message: di.Message): if not (ref := message.reference): return quoted_msg_id = ref.message_id if quoted_msg_id is None: return reply_to = MessageReference(quoted_msg_id) try: if message.type == di.MessageType.thread_starter_message: assert isinstance(, di.Thread) assert isinstance(, di.TextChannel) quoted_msg = await else: quoted_msg = await except di.errors.NotFound: reply_to.body = "[quoted message could not be fetched]" return reply_to reply_to.body = quoted_msg.clean_content author = if author == self.session.discord.user: = self.session.user return reply_to muc: "MUC" = getattr(self, "muc", None) # type: ignore if muc: = await muc.get_participant_by_discord_user(author) else: = self # type: ignore return reply_to
[docs] async def send_message( self, message: di.Message, archive_only=False, correction=False ): reply_to = await self._reply_to(message) mtype = message.type if mtype == di.MessageType.thread_created: text = f"/me created a thread named '{message.clean_content}'" elif mtype == di.MessageType.thread_starter_message: text = "I started a new thread from this message ↑" else: text = message.clean_content channel = if isinstance(channel, di.Thread): thread = if message.type == di.MessageType.channel_name_change: text = f"/me renamed this thread: {text}" else: thread = None # it seems attachments cannot be edited in discord anyway, only the text # of the message attachments = ( [Attachment.from_discord(a) for a in message.attachments] if not correction else [] ) await self.send_files( attachments, body_first=True,, when=message.created_at, thread=thread, body=text, reply_to=reply_to, archive_only=archive_only, == self.session.discord.user, correction=correction, )
[docs]class StatusMixin(PresenceMixin):
[docs] def update_status( self, status: di.Status, activity: Optional[ Union[di.Activity, di.Game, di.CustomActivity, di.Streaming, di.Spotify] ], ): # TODO: implement timeouts for activities (the Activity object has timestamps # attached to it) msg = strip_illegal_chars(str(activity)) if activity else None if status == elif status == di.Status.offline: self.offline(msg) elif status == di.Status.idle: self.away(msg) elif status == di.Status.dnd: self.busy(msg)
[docs]class Attachment(LegacyAttachment): @staticmethod
[docs] def from_discord(di_attachment: di.Attachment): return Attachment( url=di_attachment.url, name=di_attachment.filename, content_type=di_attachment.content_type,, )