chatbot.py

# -*- coding: utf-8 -*-

import time

import asyncio
import http.cookies
import random
from typing import *

import aiohttp

import blivedm
import blivedm.models.web as web_models
from blivedm.clients.ws_base import BulletChat

import USER_INFO

# 直播间ID的取值看直播间URL
TEST_ROOM_ID = 9792252

# 这里填一个已登录账号的cookie的SESSDATA字段的值。不填也可以连接，但是收到弹幕的用户名会打码，UID会变成0
cookies = http.cookies.SimpleCookie()
cookies['SESSDATA'] = USER_INFO.SESSDATA
cookies['SESSDATA']['domain'] = 'bilibili.com'
cookies['bili_jct'] = USER_INFO.bili_jct
cookies['buvid3'] = USER_INFO.buvid3

session: Optional[aiohttp.ClientSession] = None

async def dm_monitor():
    init_session()
    try:
        await run_single_client()
    finally:
        await session.close()

def init_session():
    global session
    session = aiohttp.ClientSession()
    session.cookie_jar.update_cookies(cookies)

async def run_single_client():
    room_id = TEST_ROOM_ID

    global client
    client = blivedm.BLiveClient(room_id, session=session)
    handler = MyHandler()
    client.set_handler(handler)

    client.start()
    try:
        await client.join()
    finally:
        await client.stop_and_close()

class MyHandler(blivedm.BaseHandler):
    _CMD_CALLBACK_DICT = blivedm.BaseHandler._CMD_CALLBACK_DICT.copy()
    
    def __interact_word_callback(self, client: blivedm.BLiveClient, command: dict):
        uname = command['data']['uname']
        msg_type = command['data']['msg_type']
        if msg_type == 2:
            asyncio.create_task(client.send_bulletchat(BulletChat(msg=f'谢谢{uname}的关注喵')))

    _CMD_CALLBACK_DICT['INTERACT_WORD'] = __interact_word_callback

    def _on_heartbeat(self, client: blivedm.BLiveClient, message: web_models.HeartbeatMessage):
        pass

    def _on_danmaku(self, client: blivedm.BLiveClient, message: web_models.DanmakuMessage):
        pass

    def _on_gift(self, client: blivedm.BLiveClient, message: web_models.GiftMessage):
        pass

    def _on_buy_guard(self, client: blivedm.BLiveClient, message: web_models.GuardBuyMessage):
        pass

    def _on_super_chat(self, client: blivedm.BLiveClient, message: web_models.SuperChatMessage):
        pass


if __name__ == '__main__':
    asyncio.run(dm_monitor())