Python基于Wechaty构建一个简单的微信机器人

python3 --version
pip3 --version

python3 -m venv .venv
source .venv/bin/activate

pip install wechaty

from wechaty import Wechaty, Message, Contact

class MyBot(Wechaty):

    async def on_message(self, msg: Message):
        text = msg.text()
        talker = msg.talker()

        print(f'Message from {talker.name}: {text}')

        if text.lower() == 'ping':
            await msg.say('pong')

if __name__ == '__main__':
    import asyncio
    asyncio.run(MyBot().start())

export WECHATY_PUPPET=wechaty-puppet-service
export WECHATY_PUPPET_SERVICE_TOKEN=your_token_here
python bot.py

bot = MyBot()
bot.options.puppet = 'wechaty-puppet-service'
bot.options.puppet_token = 'your_token_here'
asyncio.run(bot.start())

docker run -e WECHATY_PUPPET=wechaty-puppet-service \
           -e WECHATY_PUPPET_SERVICE_TOKEN=your_token \
           wechaty/wechaty

pip install wechaty

import asyncio
from wechaty import Wechaty, Message

class MyBot(Wechaty):
    async def on_message(self, msg: Message):
        talker = msg.talker()
        await talker.ready()
        if msg.text() == "ding":
            await talker.say('dong')
        elif msg.text() == 'image':
            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 talker.say(file_box)

async def main():
    bot = MyBot()
    await bot.start()

asyncio.run(main())

#encoding = utf-8                                                                                                                                                                                            
from wechaty import Wechaty
from wechaty.plugin import WechatyPlugin
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import asyncio
import datetime
#from wechaty import get_room_id

class DailyPlugin(WechatyPlugin):
  @property
  def name(self) -> str:
    return 'dayily'

  async def tick(self):
    rooms = await self.bot.Room.find_all()
    for currRoom in rooms:
        print(f'room pay load topic: {currRoom.payload.topic}, topic: {currRoom.topic}, currRoom.room_id: {currRoom.room_id}')

        topic =  currRoom.payload.topic
        if(topic.startswith("ding") or topic.startswith("曼玲-食神券坊外卖")):
            room = currRoom
            await room.say(f'您好，我是食神券坊智能机器人，今天正式上线！ -> {datetime.datetime.now()}')

  async def init_plugin(self, wechaty: Wechaty):
    await super().init_plugin(wechaty)
    scheduler = AsyncIOScheduler()
    scheduler.add_job(self.tick, 'cron', hour=9, minute=43, second=55)
    scheduler.start()

async def main():
  bot = Wechaty().use(DailyPlugin())
  await bot.start()

asyncio.run(main())