listen for message event
(self, msg: Message)
| 39 | |
| 40 | # pylint: disable=R0912,R0914,R0915 |
| 41 | async def on_message(self, msg: Message) -> None: |
| 42 | """ |
| 43 | listen for message event |
| 44 | """ |
| 45 | from_contact: Contact = msg.talker() |
| 46 | text: str = msg.text() |
| 47 | room: Optional[Room] = msg.room() |
| 48 | msg_type: MessageType = msg.type() |
| 49 | file_box: Optional[FileBox] = None |
| 50 | if text == 'ding': |
| 51 | conversation: Union[ |
| 52 | Room, Contact] = from_contact if room is None else room |
| 53 | await conversation.ready() |
| 54 | await conversation.say('dong') |
| 55 | file_box = FileBox.from_url( |
| 56 | 'https://ss3.bdstatic.com/70cFv8Sh_Q1YnxGkpoWK1HF6hhy/it/' |
| 57 | 'u=1116676390,2305043183&fm=26&gp=0.jpg', |
| 58 | name='ding-dong.jpg') |
| 59 | await conversation.say(file_box) |
| 60 | |
| 61 | elif msg_type == MessageType.MESSAGE_TYPE_IMAGE: |
| 62 | logger.info('receving image file') |
| 63 | # file_box: FileBox = await msg.to_file_box() |
| 64 | image: Image = msg.to_image() |
| 65 | |
| 66 | hd_file_box: FileBox = await image.hd() |
| 67 | await hd_file_box.to_file('./hd-image.jpg', overwrite=True) |
| 68 | |
| 69 | thumbnail_file_box: FileBox = await image.thumbnail() |
| 70 | await thumbnail_file_box.to_file('./thumbnail-image.jpg', overwrite=True) |
| 71 | artwork_file_box: FileBox = await image.artwork() |
| 72 | await artwork_file_box.to_file('./artwork-image.jpg', overwrite=True) |
| 73 | # reply the image |
| 74 | await msg.say(hd_file_box) |
| 75 | |
| 76 | # pylint: disable=C0301 |
| 77 | elif msg_type in [MessageType.MESSAGE_TYPE_AUDIO, MessageType.MESSAGE_TYPE_ATTACHMENT, MessageType.MESSAGE_TYPE_VIDEO]: |
| 78 | logger.info('receving file ...') |
| 79 | file_box = await msg.to_file_box() |
| 80 | if file_box: |
| 81 | await file_box.to_file(file_box.name) |
| 82 | |
| 83 | elif msg_type == MessageType.MESSAGE_TYPE_MINI_PROGRAM: |
| 84 | logger.info('receving mini-program ...') |
| 85 | mini_program: Optional[MiniProgram] = await msg.to_mini_program() |
| 86 | if mini_program: |
| 87 | await msg.say(mini_program) |
| 88 | |
| 89 | elif text == 'get room members' and room: |
| 90 | logger.info('get room members ...') |
| 91 | room_members: List[Contact] = await room.member_list() |
| 92 | names: List[str] = [ |
| 93 | room_member.name for room_member in room_members] |
| 94 | await msg.say('\n'.join(names)) |
| 95 | |
| 96 | elif text.startswith('remove room member:'): |
| 97 | logger.info('remove room member:') |
| 98 | if not room: |
nothing calls this directly
no test coverage detected