Source code for ehforwarderbot.status

# coding=utf-8

from abc import abstractmethod, ABC
from contextlib import suppress
from typing import Dict, Collection, Any, Optional

from . import Channel, Message, coordinator
from .channel import SlaveChannel
from .chat import Chat, ChatMember
from .types import Reactions, ReactionName, ChatID, MessageID

__all__ = ["Status", "ChatUpdates", "MemberUpdates", "MessageRemoval",
           "ReactToMessage", "MessageReactionsUpdate"]


[docs]class Status(ABC): """ Abstract class of a status Attributes: destination_channel (:obj:`.Channel`): The channel that this status is sent to, usually the master channel. """ @abstractmethod def __init__(self): self.destination_channel: 'Channel' = None raise NotImplementedError() @abstractmethod def verify(self): raise NotImplementedError() def __getstate__(self): state = self.__dict__.copy() if state['destination_channel'] is not None: state['destination_channel'] = state['destination_channel'].channel_id return state def __setstate__(self, state: Dict[str, Any]): self.__dict__.update(state) with suppress(NameError): dc = coordinator.get_module_by_id(state['destination_channel']) if isinstance(dc, Channel): self.destination_channel = dc
[docs]class ChatUpdates(Status): """Inform the master channel on updates of slave chats. Attributes: channel (:obj:`.SlaveChannel`): Slave channel that issues the update new_chats (Optional[Collection[str]]): Unique ID of new chats removed_chats (Optional[Collection[str]]): Unique ID of removed chats modified_chats (Optional[Collection[str]]): Unique ID of modified chats """ # noinspection PyMissingConstructor
[docs] def __init__(self, channel: 'SlaveChannel', new_chats: Collection[ChatID] = tuple(), removed_chats: Collection[ChatID] = tuple(), modified_chats: Collection[ChatID] = tuple()): """ Args: channel (:obj:`.SlaveChannel`): Slave channel that issues the update new_chats (Optional[Collection[str]]): Unique ID of new chats removed_chats (Optional[Collection[str]]): Unique ID of removed chats modified_chats (Optional[Collection[str]]): Unique ID of modified chats """ self.channel: 'SlaveChannel' = channel self.new_chats: Collection[ChatID] = new_chats self.removed_chats: Collection[ChatID] = removed_chats self.modified_chats: Collection[ChatID] = modified_chats self.destination_channel = coordinator.master self.verify()
def __str__(self): return "<ChatUpdates @ {s.channel.channel_name}; New: {s.new_chats}; " \ "Removed: {s.removed_chats}; Modified: {s.modified_chats}>".format(s=self) def verify(self): assert isinstance(self.channel, SlaveChannel), f"Slave channel {self.channel!r} is not valid." def __getstate__(self): state = super(ChatUpdates, self).__getstate__() if state['channel'] is not None: state['channel'] = state['channel'].channel_id return state def __setstate__(self, state: Dict[str, Any]): super(ChatUpdates, self).__setstate__(state) with suppress(NameError): c = coordinator.get_module_by_id(state['channel']) if isinstance(c, SlaveChannel): self.channel = c
[docs]class MemberUpdates(Status): """Inform the master channel on updates of members in a slave chat. Attributes: channel (:obj:`.SlaveChannel`): Slave channel that issues the update chat_id (str): Unique ID of the chat. new_members (Optional[Collection[str]]): Unique ID of new members removed_members (Optional[Collection[str]]): Unique ID of removed members modified_members (Optional[Collection[str]]): Unique ID of modified members """ # noinspection PyMissingConstructor
[docs] def __init__(self, channel: 'SlaveChannel', chat_id: ChatID, new_members: Collection[ChatID] = tuple(), removed_members: Collection[ChatID] = tuple(), modified_members: Collection[ChatID] = tuple()): """ Args: channel (:obj:`.SlaveChannel`): Slave channel that issues the update chat_id (str): Unique ID of the chat. new_members (Optional[Collection[str]]): Unique ID of new members removed_members (Optional[Collection[str]]): Unique ID of removed members modified_members (Optional[Collection[str]]): Unique ID of modified members """ self.channel: 'SlaveChannel' = channel self.chat_id: ChatID = chat_id self.new_members: Collection[ChatID] = new_members self.removed_members: Collection[ChatID] = removed_members self.modified_members: Collection[ChatID] = modified_members self.destination_channel = coordinator.master self.verify()
def __str__(self): return "<MemberUpdates: {s.chat_id} @ {s.channel.channel_name}; New: {s.new_chats}; " \ "Removed: {s.removed_chats}; Modified: {s.modified_chats}>".format(s=self) def verify(self): assert isinstance(self.channel, SlaveChannel), f"Slave channel {self.channel!r} is not valid." def __getstate__(self): state = super(MemberUpdates, self).__getstate__() if state['channel'] is not None: state['channel'] = state['channel'].channel_id return state def __setstate__(self, state: Dict[str, Any]): super(MemberUpdates, self).__setstate__(state) with suppress(NameError): c = coordinator.get_module_by_id(state['channel']) if isinstance(c, SlaveChannel): self.channel = c
[docs]class MessageRemoval(Status): """Inform a channel to remove a certain message. This is usually known as “delete from everyone”, “delete from recipient”, “recall a message”, “unsend”, or “revoke a message” as well, depends on the IM platform. Some channels MAY not support removal of messages, and raises a :obj:`.exceptions.EFBOperationNotSupported` exception. Feedback by sending another ``MessageRemoval`` back is not required when this object is sent from a master channel. Master channels SHOULD treat a successful delivery of this status as a successful removal. Attributes: source_channel (:obj:`.Channel`): Channel issued the status destination_channel (:obj:`.Channel`): Channel the status is issued to message (:obj:`~.message.Message`): Message to remove. This MAY not be a complete :obj:`.message.Message` object. Raises: :obj:`.exceptions.EFBOperationNotSupported`: When message removal is not supported in the channel. """ # noinspection PyMissingConstructor
[docs] def __init__(self, source_channel: 'Channel', destination_channel: 'Channel', message: 'Message'): """Create a message removal status Try to provided as much as you can, if not, provide a minimum information in the channel: * Slave channel ID and chat ID (:attr:`message.chat.module_id <.Chat.module_id>` and :attr:`message.chat.uid <.Chat.uid>`) * Message unique ID from the slave channel (:attr:`message.uid <.message.Message.uid>`) Args: source_channel (:obj:`~.channel.Channel`): Channel issued the status destination_channel (:obj:`~.channel.Channel`): Channel the status is issued to message (:obj:`~.message.Message`): Message to remove. """ self.source_channel: 'Channel' = source_channel self.destination_channel: 'Channel' = destination_channel self.message: 'Message' = message self.verify()
def __str__(self): return "<MessageRemoval: {s.message}; {s.source_channel.channel_name} " \ "-> {s.destination_channel.channel_name}>".format(s=self) def verify(self): assert isinstance(self.source_channel, Channel), \ f"Source channel {self.source_channel!r} is not valid." assert isinstance(self.destination_channel, Channel), \ f"Destination channel {self.destination_channel!r} is not valid." assert isinstance(self.message, Message), \ f"Message {self.message!r} is not valid." assert self.message.chat.module_id and self.message.chat.uid and self.message.uid, \ f"Message does not contain the minimum information required: {self.message!r}" def __getstate__(self): state = super(MessageRemoval, self).__getstate__() if state['source_channel'] is not None: state['source_channel'] = state['source_channel'].channel_id return state def __setstate__(self, state: Dict[str, Any]): super(MessageRemoval, self).__setstate__(state) with suppress(NameError): sc = coordinator.get_module_by_id(state['source_channel']) if isinstance(sc, Channel): self.source_channel = sc
[docs]class ReactToMessage(Status): """ Created when user react to a message, issued from master channel. When this status is sent, a :obj:`.status.MessageReactionsUpdate` is RECOMMENDED to be issued back to master channel. Args: chat (:obj:`Chat`): The chat where message is sent msg_id (str): ID of the message to react to reaction (Optional[str]): The reaction name to be sent, usually an emoji. Set to ``None`` to remove reaction. destination_channel (:obj:`.SlaveChannel`): Channel the status is issued to, extracted from the chat object. Raises: :obj:`.exceptions.EFBMessageReactionNotPossible`: Raised when the reaction is not valid (e.g. the specific reaction is not in the list of possible reactions). :obj:`.exceptions.EFBOperationNotSupported`: Raised when reaction in any form is not supported to the message at all. """ # noinspection PyMissingConstructor
[docs] def __init__(self, chat: 'Chat', msg_id: str, reaction: Optional[ReactionName]): """ Args: chat (:obj:`.Chat`): The chat where message is sent msg_id (str): ID of the message to react to reaction: The reaction name to be sent, usually an emoji """ self.chat: 'Chat' = chat self.msg_id: str = msg_id self.reaction: Optional[str] = reaction if getattr(self.chat, 'module_id', None): self.destination_channel = coordinator.slaves[self.chat.module_id] self.verify()
def verify(self): assert isinstance(self.chat, Chat), f"Chat {self.chat!r} is not valid." assert self.msg_id, f"Message ID {self.msg_id!r} is not valid." assert isinstance(self.destination_channel, SlaveChannel), \ f"Destination channel is not a slave channel, but {self.destination_channel}."
[docs]class MessageReactionsUpdate(Status): """ Update reacts of a message, issued from slave channel to master channel. Args: chat (:obj:`.Chat`): The chat where message is sent msg_id (str): ID of the message for the reacts reactions: Indicate reactions to the message. Dictionary key represents the reaction name, usually an emoji. Value is a collection of users who reacted to the message with that certain emoji. All :obj:`Chat` objects in this dict MUST be members in the chat of the message. destination_channel (:obj:`.MasterChannel`): Channel the status is issued to, which is always the master channel. """ # noinspection PyMissingConstructor
[docs] def __init__(self, chat: 'Chat', msg_id: MessageID, reactions: Reactions): """ Args: chat (:obj:`.Chat`): The chat where message is sent msg_id (str): ID of the message for the reacts reactions: Indicate reactions to the message. Dictionary key represents the reaction name, usually an emoji. Value is a collection of users who reacted to the message with that certain emoji. All :obj:`Chat` objects in this dict MUST be members in the chat of the message. """ self.chat: 'Chat' = chat self.msg_id: MessageID = msg_id self.reactions: Reactions = reactions self.destination_channel = coordinator.master self.verify()
def verify(self): assert isinstance(self.chat, Chat), f"Chat {self.chat!r} is not valid." assert self.msg_id, f"Message ID {self.msg_id} is not valid." for reaction, users in self.reactions.items(): for user in users: assert isinstance(user, ChatMember), f"Expected reactor of {reaction} to be a ChatMember, but {user!r} found."