微信数据小助手

import asyncio
import logging
from typing import Optional, Union
import os
from wechaty_puppet import FileBox, ScanStatus  # type: ignore
from wechaty_puppet import MessageType
from wechaty import Wechaty, Contact
from wechaty.user import Message, Room
logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)
import time
class MyBot(Wechaty):
    """
    listen wechaty event with inherited functions, which is more friendly for
    oop developer
    """
    def __init__(self):
        super().__init__()
    async def on_message(self, msg: Message):
        """
        listen for message event
        """
        from_contact = msg.talker()
        text = msg.text()
        room = msg.room()
        if room:
            await room.ready()
        # 保存图片
        if msg.type() == MessageType.MESSAGE_TYPE_IMAGE:
            img = await msg.to_file_box()
            # save the image as local file
            await img.to_file(os.path.join('img', img.name))
        # 保存视频
        elif msg.type() == MessageType.MESSAGE_TYPE_VIDEO:
            video = await msg.to_file_box()
            # save the video as local file
            await video.to_file(os.path.join('video', video.name))
        # 保存语音
        elif msg.type() == MessageType.MESSAGE_TYPE_AUDIO:
            audio = await msg.to_file_box()
            # save the audio file as local file
            await audio.to_file(os.path.join('audio', audio.name))
        # 保存表情
        elif msg.type() == MessageType.MESSAGE_TYPE_EMOTICON:
            emoticon = await msg.to_file_box()
            # save the audio file as local file
            await emoticon.to_file(os.path.join('emoticon', emoticon.name))
        # 保存文件
        elif msg.type() == MessageType.MESSAGE_TYPE_ATTACHMENT:
            attachment = await msg.to_file_box()
            # save the audio file as local file
            await attachment.to_file(os.path.join('file', attachment.name))
        # 保存聊天记录
        elif msg.type() == MessageType.MESSAGE_TYPE_TEXT:
            txt = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + ',' + from_contact.name + ',' + text + '\n'
            print(txt)
            with open('chat_log.csv', 'a', encoding='utf-8') as f:
                f.write(txt)
        print('done')
    async def on_login(self, contact: Contact):
        log.info(f'user: {contact} has login')
    async def on_scan(self, status: ScanStatus, qr_code: Optional[str] = None,
                      data: Optional[str] = None):
        contact = self.Contact.load(self.contact_id)
        await contact.ready()
        print(f'user <{contact}> scan status: {status.name} , '
              f'qr_code: {qr_code}')
bot: Optional[MyBot] = None
async def main():
    """doc"""
    # pylint: disable=W0603
    global bot
    bot = MyBot()
    await bot.start()
asyncio.run(main())