Skip to main content

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

Project description

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()

相关项目

鸣谢

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

starbot_bilibili_core-1.3.0.tar.gz (47.5 kB view details)

Uploaded Source

Built Distribution

starbot_bilibili_core-1.3.0-py3-none-any.whl (31.5 kB view details)

Uploaded Python 3

File details

Details for the file starbot_bilibili_core-1.3.0.tar.gz.

File metadata

  • Download URL: starbot_bilibili_core-1.3.0.tar.gz
  • Upload date:
  • Size: 47.5 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.6.1 CPython/3.11.6 Windows/10

File hashes

Hashes for starbot_bilibili_core-1.3.0.tar.gz
Algorithm Hash digest
SHA256 a83e1b3ec4c423b2081a30324da30b35c2e2866e84b6996f36948e3c7e6867fe
MD5 48732e2250dcaa0d139693b4d77e02ba
BLAKE2b-256 b662c81224281236b69c36a01d8c44ac0a8812ef5b0cae5e14862f1f74d993c0

See more details on using hashes here.

File details

Details for the file starbot_bilibili_core-1.3.0-py3-none-any.whl.

File metadata

File hashes

Hashes for starbot_bilibili_core-1.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 a2cf9074e267081b30c6579af5ba751e52dc46a03e948cd0313632593f4b797c
MD5 af13921b8ec0e69170ef3eb7a4d7e70c
BLAKE2b-256 43e7da51601ce76061976a85a8de6b4df26eb2d87f89267810aa12058b5303aa

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page