StarBotCore

一个优雅的异步 bilibili 直播间监听库

特性

• 极简代码调用，可快速实现高级功能
• 细化原始事件，自动区分文字弹幕与表情弹幕，免费礼物、付费礼物与盲盒礼物等
• 基于分级消息主题的订阅发布模式，可灵活监听单个直播间或多个直播间
• 自动解析原始数据并包装为事件发布，无需手动使用代码处理原始数据

快速开始

安装

pip install starbot-bilibili-core

代码框架

在开始编写代码前，需要引入异步任务执行器 executor，并执行初始化操作

from starbot_executor import executor

async def main():
    # 业务逻辑
    pass

loop = executor.init()
loop.run_until_complete(main())
loop.run_forever()

监听直播间

引入直播间事件监听器 listener 后，仅须一行代码即可添加监听直播间，并自动持续发布该直播间的相关事件

await listener.add(room_id=22889484)

已支持的事件列表

• 连接成功事件（ConnectedEvent）
• 断开连接事件（DisconnectedEvent）
• 心跳响应超时事件（TimeoutEvent）
• 开播事件（LiveOnEvent）
• 下播事件（LiveOffEvent）
• 文字弹幕事件（DanmuEvent）
• 大表情弹幕事件（EmojiEvent）
• 免费礼物事件（SilverGiftEvent）
• 付费礼物事件（GoldGiftEvent）
• 盲盒礼物事件（BlindGiftEvent）
• 醒目留言事件（SuperChatEvent）
• 开通舰长事件（CaptainEvent）
• 开通提督事件（CommanderEvent）
• 开通总督事件（GovernorEvent）
• 进房事件（EnterRoomEvent）
• 关注事件（FollowEvent）
• 观看过人数更新事件（WatchedUpdateEvent）
• 点赞事件（LikeClickEvent）
• 点赞数更新事件（LikeUpdateEvent）

订阅-发布机制

消息主题为层级结构，较高层级的消息主题可同时监听到较低层级的事件，举例如下：

弹幕事件对应消息主题为三层，一级主题为：EventType.LiveEvent，二级主题为：LiveEvent.DanmuEvent，三级主题为：房间号
使用 @executor.on(EventType.LiveEvent, LiveEvent.DanmuEvent, 房间号) 装饰的方法，仅可监听到相应房间号的弹幕事件
使用 @executor.on(EventType.LiveEvent, LiveEvent.DanmuEvent) 装饰的方法，可监听到所有房间的弹幕事件
使用 @executor.on(EventType.LiveEvent) 或 @executor.on() 装饰的方法，可监听到所有房间的所有事件

完整示例

from loguru import logger
from starbot_executor import executor

from starbot_core import listener, EventType, LiveEvent, ConnectedEvent, DisconnectedEvent, DanmuEvent, BlindGiftEvent
from starbot_core.exception.LoginException import LoginException
from starbot_core.utils.network import credential


async def main():
    # 登录 B 站账号，不登录账号无法抓取完整数据，Cookie 获取方式：https://bot.starlwr.com/depoly/document
    try:
        await credential.login(
            sessdata="在此填入 SESSDATA 字段",
            bili_jct="在此填入 bili_jct 字段",
            buvid3="在此填入 buvid3 字段"
        )
    except LoginException as e:
        logger.error(e.msg, e)
        return

    # 例 1: 监听所有直播间的所有事件
    @executor.on()
    async def on_all(subjects, event):
        logger.info(f"{subjects}: {event}")

    # 例 2: 监听所有直播间的连接成功事件
    @executor.on(EventType.LiveEvent, LiveEvent.ConnectedEvent)
    async def on_connected(event: ConnectedEvent):
        logger.success(f"已成功连接到 {event.source.uname} 的直播间 {event.source.room_id}")

    # 例 3: 监听所有直播间的断开连接事件
    @executor.on(EventType.LiveEvent, LiveEvent.DisconnectedEvent)
    async def on_disconnected(event: DisconnectedEvent):
        logger.opt(colors=True).info(f"已断开连接 <cyan>{event.source.uname}</> 的直播间 <cyan>{event.source.room_id}</>")

    # 例 4: 监听 22889484 直播间的弹幕事件
    @executor.on(EventType.LiveEvent, LiveEvent.DanmuEvent, 22889484)
    async def on_danmu(event: DanmuEvent):
        if event.fans_medal is None:
            logger.info(f"{event.sender.uname} (无粉丝勋章): {event.content}")
        else:
            logger.info(
                f"{event.sender.uname} (Lv {event.fans_medal.level} - {event.fans_medal.name}): {event.content}")

    # 例 5: 监听所有直播间的盲盒礼物事件
    @executor.on(EventType.LiveEvent, LiveEvent.BlindGiftEvent)
    async def on_blind_gift(event: BlindGiftEvent):
        if event.profit >= 0:
            logger.info(f"{event.sender.uname} 通过 {event.blind_name} 开出了 {event.count} 个 {event.gift_name}, 赚了 {abs(event.profit)} 元")
        else:
            logger.info(f"{event.sender.uname} 通过 {event.blind_name} 开出了 {event.count} 个 {event.gift_name}, 亏了 {abs(event.profit)} 元")

    # 循环添加监听直播间
    room_ids = (22889484, 55634, 21013446)
    for room_id in room_ids:
        await listener.add(room_id=room_id)


listener.auto_complete = True  # 设置为 True 可以获得更完整的信息（部分事件中的头像、昵称等），但会因网络请求导致一定的延迟，请酌情使用
loop = executor.init()
loop.run_until_complete(main())
loop.run_forever()

相关项目

• StarBotExecutor: 一个基于订阅发布模式的异步执行器

鸣谢

• bilibili-api: 哔哩哔哩的 API 调用模块
• bilibili-API-collect: 哔哩哔哩 API 收集整理