How to use the ehforwarderbot.MsgType function in ehforwarderbot

To help you get started, we’ve selected a few ehforwarderbot examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github blueset / ehForwarderBot / tests / test_message.py View on Github external
def test_verify_location_message(base_message):
    msg = base_message

    msg.type = MsgType.Location
    msg.attributes = EFBMsgLocationAttribute(latitude=0.0, longitude=0.0)
    msg.verify()

    with pytest.raises(ValueError) as exec_info:
        msg.attributes = EFBMsgLocationAttribute(latitude='0.0', longitude=1.0)
        msg.verify()
    assert 'Latitude' in exec_info.value.args[0]

    with pytest.raises(ValueError) as exec_info:
        msg.attributes = EFBMsgLocationAttribute(latitude=1.0, longitude=10)
        msg.verify()
    assert 'Longitude' in exec_info.value.args[0]
github blueset / ehForwarderBot / tests / mocks / slave.py View on Github external
from typing import Set, Optional, List
from logging import getLogger

from ehforwarderbot import EFBChannel, EFBMsg, EFBStatus, ChannelType, MsgType, EFBChat, ChatType
from ehforwarderbot.exceptions import EFBChatNotFound
from ehforwarderbot.utils import extra
from ehforwarderbot.types import ModuleID, MessageID


class MockSlaveChannel(EFBChannel):

    channel_name: str = "Mock Slave"
    channel_emoji: str = "➖"
    channel_id: ModuleID = ModuleID("tests.mocks.slave.MockSlaveChannel")
    channel_type: ChannelType = ChannelType.Slave
    supported_message_types: Set[MsgType] = {MsgType.Text, MsgType.Link}
    __version__: str = '0.0.1'

    logger = getLogger(channel_id)

    polling = threading.Event()

    __picture_dict = {
        "alice": "A.png",
        "bob": "B.png",
        "carol": "C.png",
        "dave": "D.png",
        "wonderland001": "W.png"
    }

    def __init__(self, instance_id=None):
        super().__init__(instance_id)
github blueset / efb-telegram-master / tests / integration / test_master_message.py View on Github external
def compare_message(self, tg_msg: Message, efb_msg: EFBMessage) -> None:
        assert efb_msg.type == MsgType.Sticker
        assert efb_msg.file
        assert efb_msg.file.seek(0, 2)
        # Cannot compare file size as WebP pictures are converted to PNG here.
github milkice233 / efb-qq-slave / efb_qq_slave / __init__.py View on Github external
from . import __version__ as version

from .Clients.BaseClient import BaseClient
from .ClientMgr import ClientMgr


class QQMessengerChannel(EFBChannel):
    channel_name: str = "QQ Slave"
    channel_emoji: str = "🐧"
    channel_id = "milkice.qq"
    channel_type: ChannelType = ChannelType.Slave

    __version__ = version.__version__

    supported_message_types = {MsgType.Text, MsgType.Sticker, MsgType.Image,
                               MsgType.Link, MsgType.Voice, MsgType.Animation}

    # todo supported_message can be dynamically defined by Client

    config = dict()
    QQClientMgr: ClientMgr
    QQClient: BaseClient
    logger: logging.Logger = logging.getLogger(__name__)

    def __init__(self, instance_id: str = None):
        super().__init__(instance_id)
        """
        Load Config
        """
        self.load_config()

    def check_updates(self):
github milkice233 / efb-qq-slave / efb_qq_slave / Clients / CoolQ / MsgDecorator.py View on Github external
def qq_image_wrapper(self, data, chat: Chat = None):
        efb_msg = Message()
        if 'url' not in data:
            efb_msg.type = MsgType.Text
            efb_msg.text = self._('[Image Source missing]')
            return [efb_msg]

        efb_msg.file = cq_get_image(data['url'])
        if efb_msg.file is None:
            efb_msg.type = MsgType.Text
            efb_msg.text = self._('[Download image failed, please check on your QQ client]')
            return [efb_msg]

        efb_msg.type = MsgType.Image
        mime = magic.from_file(efb_msg.file.name, mime=True)
        if isinstance(mime, bytes):
            mime = mime.decode()
        efb_msg.filename = data['file'] if 'file' in data else efb_msg.file.name
        efb_msg.path = efb_msg.file.name
        efb_msg.mime = mime
github blueset / efb-wechat-slave / efb_wechat_slave / slave_message.py View on Github external
def wechat_shared_image_msg(self, msg: wxpy.Message, source: str, text: str = "", mode: str = "image") -> Message:
        efb_msg = Message()
        efb_msg.type = MsgType.Image
        efb_msg.text = self._("Via {source}").format(source=source)
        if text:
            efb_msg.text = "%s\n%s" % (text, efb_msg.text)
        efb_msg.path, efb_msg.mime, efb_msg.file = self.save_file(msg, app_message=mode)
        return efb_msg
github blueset / efb-telegram-master / efb_telegram_master / db.py View on Github external
def build_etm_msg(self, chat_manager: ChatObjectCacheManager,
                      recur: bool = True) -> ETMMsg:
        c_module, c_id, _ = chat_id_str_to_id(self.slave_origin_uid)
        a_module, a_id, a_grp = chat_id_str_to_id(self.slave_member_uid)
        chat: 'ETMChatType' = chat_manager.get_chat(c_module, c_id, build_dummy=True)
        author: 'ETMChatMember' = chat_manager.get_chat_member(a_module, a_grp, a_id, build_dummy=True)  # type: ignore
        msg = ETMMsg(
            uid=self.slave_message_id,
            chat=chat,
            author=author,
            text=self.text,
            type=MsgType(self.msg_type),
            type_telegram=TGMsgType(self.media_type),
            mime=self.mime or None,
            file_id=self.file_id or None,
        )
        with suppress(NameError):
            to_module = coordinator.get_module_by_id(self.sent_to)
            if isinstance(to_module, Channel):
                msg.deliver_to = to_module

        # - ``target``: ``master_msg_id`` of the target message
        # - ``is_system``
        # - ``attributes``
        # - ``commands``
        # - ``substitutions``: ``Dict[Tuple[int, int], SlaveChatID]``
        # - ``reactions``: ``Dict[str, Collection[SlaveChatID]]``
        if self.pickle:
github milkice233 / efb-qq-slave / efb_qq_slave / Clients / CoolQ / CoolQ.py View on Github external
context['uid_prefix'] = 'group_request'
            context['group_name'] = self._('[Request]') + self.get_group_info(context['group_id'])['group_name']
            context['group_id_orig'] = context['group_id']
            context['group_id'] = str(context['group_id']) + "_notification"
            context['message_type'] = 'group'
            context['event_description'] = '\u2139 New Group Join Request'
            original_group = self.get_group_info(context['group_id'], False)
            group_name = context['group_id']
            if original_group is not None and 'group_name' in original_group:
                group_name = original_group['group_name']
            msg = Message()
            msg.uid = 'group' + '_' + str(context['group_id'])
            msg.author = self.chat_manager.build_efb_chat_as_system_user(context)
            msg.chat = self.chat_manager.build_efb_chat_as_group(context)
            msg.deliver_to = coordinator.master
            msg.type = MsgType.Text
            name = ""
            if not self.get_friend_remark(context['user_id']):
                name = "{}({})[{}] ".format(
                    self.get_stranger_info(context['user_id'])['nickname'], self.get_friend_remark(context['user_id']),
                    context['user_id'])
            else:
                name = "{}[{}] ".format(self.get_stranger_info(context['user_id'])['nickname'], context['user_id'])
            msg.text = "{} wants to join the group {}({}). \nHere is the comment: {}".format(
                name, group_name, context['group_id_orig'], context['comment']
            )
            msg.commands = MessageCommands([MessageCommand(
                name=self._("Accept"),
                callable_name="process_group_request",
                kwargs={'result': 'accept',
                        'flag': context['flag'],
                        'sub_type': context['sub_type']}
github blueset / efb-telegram-master / efb_telegram_master / message.py View on Github external
def __init__(self, attributes: Optional[MessageAttribute] = None, author: ChatMember = None, chat: Chat = None,
                 commands: Optional[MessageCommands] = None, deliver_to: Channel = None, edit: bool = False,
                 edit_media: bool = False, file: Optional[BinaryIO] = None, filename: Optional[str] = None,
                 is_system: bool = False, mime: Optional[str] = None, path: Optional[Path] = None,
                 reactions: Reactions = None, substitutions: Optional[Substitutions] = None,
                 target: 'Optional[Message]' = None, text: str = "", type: MsgType = MsgType.Unsupported,
                 uid: Optional[MessageID] = None, vendor_specific: Dict[str, Any] = None,
                 type_telegram: TGMsgType = TGMsgType.System, file_id: Optional[str] = None):
        super().__init__(attributes=attributes, chat=chat, author=author, commands=commands, deliver_to=deliver_to,
                         edit=edit, edit_media=edit_media, file=file, filename=filename, is_system=is_system, mime=mime,
                         path=path, reactions=reactions, substitutions=substitutions, target=target, text=text,
                         type=type, uid=uid, vendor_specific=vendor_specific)
        self.__initialized = False
        self.type_telegram = type_telegram
        self.file_id = file_id