starbot-bilibili-core

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


Keywords
starbot, bilibili, bot
License
AGPL-3.0-only
Install
pip install starbot-bilibili-core==0.0.1b0

Documentation

StarBotCore

PyPI Python License STARS

一个优雅的异步 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()

相关项目

鸣谢