- Python-wechaty is a unified conversational RPA SDK for chatbot makers.
- Developer at least use six lines code to create chatbot based on multi-ims,
- Eg: WeChat, WeChat official account, dingtalk, lark, whatsapp, giter, and so on .
There are Chinese documents and English documents for developers to create their own chatbots.
Prerequisites:
pip install -r projects/chatbot/requirements.txt
Requirements:
- wechaty
Run the script:
python projects/chatbot/bot.py
Source Code:
bot.py
"""example code for ding-dong-bot with oop style"""
from typing import List, Optional, Union
import asyncio
from datetime import datetime
from wechaty_puppet import get_logger
from wechaty import (
MessageType,
FileBox,
RoomMemberQueryFilter,
Wechaty,
Contact,
Room,
Message,
Image,
MiniProgram,
Friendship,
FriendshipType,
EventReadyPayload
)
logger = get_logger(__name__)
class MyBot(Wechaty):
"""
listen wechaty event with inherited functions, which is more friendly for
oop developer
"""
def __init__(self) -> None:
"""initialization function
"""
self.login_user: Optional[Contact] = None
super().__init__()
async def on_ready(self, payload: EventReadyPayload) -> None:
"""listen for on-ready event"""
logger.info('ready event %s...', payload)
# pylint: disable=R0912,R0914,R0915
async def on_message(self, msg: Message) -> None:
"""
listen for message event
"""
from_contact: Contact = msg.talker()
text: str = msg.text()
room: Optional[Room] = msg.room()
msg_type: MessageType = msg.type()
file_box: Optional[FileBox] = None
if text == 'ding':
conversation: Union[
Room, Contact] = from_contact if room is None else room
await conversation.ready()
await conversation.say('dong')
file_box = FileBox.from_url(
'https://ss3.bdstatic.com/70cFv8Sh_Q1YnxGkpoWK1HF6hhy/it/'
'u=1116676390,2305043183&fm=26&gp=0.jpg',
name='ding-dong.jpg')
await conversation.say(file_box)
elif msg_type == MessageType.MESSAGE_TYPE_IMAGE:
logger.info('receving image file')
# file_box: FileBox = await msg.to_file_box()
image: Image = msg.to_image()
hd_file_box: FileBox = await image.hd()
await hd_file_box.to_file('./hd-image.jpg', overwrite=True)
thumbnail_file_box: FileBox = await image.thumbnail()
await thumbnail_file_box.to_file('./thumbnail-image.jpg', overwrite=True)
artwork_file_box: FileBox = await image.artwork()
await artwork_file_box.to_file('./artwork-image.jpg', overwrite=True)
# reply the image
await msg.say(hd_file_box)
# pylint: disable=C0301
elif msg_type in [MessageType.MESSAGE_TYPE_AUDIO, MessageType.MESSAGE_TYPE_ATTACHMENT, MessageType.MESSAGE_TYPE_VIDEO]:
logger.info('receving file ...')
file_box = await msg.to_file_box()
if file_box:
await file_box.to_file(file_box.name)
elif msg_type == MessageType.MESSAGE_TYPE_MINI_PROGRAM:
logger.info('receving mini-program ...')
mini_program: Optional[MiniProgram] = await msg.to_mini_program()
if mini_program:
await msg.say(mini_program)
elif text == 'get room members' and room:
logger.info('get room members ...')
room_members: List[Contact] = await room.member_list()
names: List[str] = [
room_member.name for room_member in room_members]
await msg.say('\n'.join(names))
elif text.startswith('remove room member:'):
logger.info('remove room member:')
if not room:
await msg.say('this is not room zone')
return
room_member_name = text[len('remove room member:') + 1:]
room_member: Optional[Contact] = await room.member(
query=RoomMemberQueryFilter(name=room_member_name)
)
if room_member:
if self.login_user and self.login_user.contact_id in room.payload.admin_ids:
await room.delete(room_member)
else:
await msg.say('登录用户不是该群管理员...')
else:
await msg.say(f'can not fine room member by name<{room_member_name}>')
elif text.startswith('get room topic'):
logger.info('get room topic')
if room:
topic: Optional[str] = await room.topic()
if topic:
await msg.say(topic)
elif text.startswith('rename room topic:'):
logger.info('rename room topic ...')
if room:
new_topic = text[len('rename room topic:') + 1:]
await msg.say(new_topic)
elif text.startswith('add new friend:'):
logger.info('add new friendship ...')
identity_info = text[len('add new friend:'):]
weixin_contact: Optional[Contact] = await self.Friendship.search(weixin=identity_info)
phone_contact: Optional[Contact] = await self.Friendship.search(phone=identity_info)
contact: Optional[Contact] = weixin_contact or phone_contact
if contact:
await self.Friendship.add(contact, 'hello world ...')
elif text.startswith('at me'):
if room:
talker = msg.talker()
await room.say('hello', mention_ids=[talker.contact_id])
elif text.startswith('my alias'):
talker = msg.talker()
alias = await talker.alias()
await msg.say('your alias is:' + (alias or ''))
elif text.startswith('set alias:'):
talker = msg.talker()
new_alias = text[len('set alias:'):]
# set your new alias
alias = await talker.alias(new_alias)
# get your new alias
alias = await talker.alias()
await msg.say('your new alias is:' + (alias or ''))
elif text.startswith('find friends:'):
friend_name: str = text[len('find friends:'):]
friend = await self.Contact.find(friend_name)
if friend:
logger.info('find only one friend <%s>', friend)
friends: List[Contact] = await self.Contact.find_all(friend_name)
logger.info('find friend<%d>', len(friends))
logger.info(friends)
else:
pass
if msg.type() == MessageType.MESSAGE_TYPE_UNSPECIFIED:
talker = msg.talker()
assert isinstance(talker, Contact)
async def on_login(self, contact: Contact) -> None:
"""login event
Args:
contact (Contact): the account logined
"""
logger.info('Contact<%s> has logined ...', contact)
self.login_user = contact
async def on_friendship(self, friendship: Friendship) -> None:
"""when receive a new friendship application, or accept a new friendship
Args:
friendship (Friendship): contains the status and friendship info,
eg: hello text, friend contact object
"""
MAX_ROOM_MEMBER_COUNT = 500
# 1. receive a new friendship from someone
if friendship.type() == FriendshipType.FRIENDSHIP_TYPE_RECEIVE:
hello_text: str = friendship.hello()
# accept friendship when there is a keyword in hello text
if 'wechaty' in hello_text.lower():
await friendship.accept()
# 2. you have a new friend to your contact list
elif friendship.type() == FriendshipType.FRIENDSHIP_TYPE_CONFIRM:
# 2.1 invite the user to wechaty group
# find the topic of room which contains Wechaty keyword
wechaty_rooms: List[Room] = await self.Room.find_all('Wechaty')
# 2.2 find the suitable room
for wechaty_room in wechaty_rooms:
members: List[Contact] = await wechaty_room.member_list()
if len(members) < MAX_ROOM_MEMBER_COUNT:
contact: Contact = friendship.contact()
await wechaty_room.add(contact)
break
async def on_room_join(self, room: Room, invitees: List[Contact],
inviter: Contact, date: datetime) -> None:
"""on_room_join when there are new contacts to the room
Args:
room (Room): the room instance
invitees (List[Contact]): the new contacts to the room
inviter (Contact): the inviter who share qrcode or manual invite someone
date (datetime): the datetime to join the room
"""
# 1. say something to welcome the new arrivals
names: List[str] = []
for invitee in invitees:
await invitee.ready()
names.append(invitee.name)
await room.say(f'welcome {",".join(names)} to the wechaty group !')
async def main() -> None:
"""doc"""
bot = MyBot()
await bot.start()
asyncio.run(main())
Code language: HTML, XML (xml)
simple-bot.py
import asyncio
from wechaty import Wechaty, Message
async def on_message(msg: Message):
if msg.text() == 'ding':
await msg.say('dong')
async def main():
bot = Wechaty()
bot.on('message', on_message)
await bot.start()
asyncio.run(main())
Code language: JavaScript (javascript)