tursom/blivedm
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'github/HEAD' into dev

Browse Source
This commit is contained in:
tursom
2025-06-08 20:56:49 +08:00
parent da3cd7db31 a45ee8f677
commit ea4cadd3c5
11 changed files with 600 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -100,6 +100,9 @@ ENV/
.idea/
.pdm-python
pdm.lock
pdm.toml
cookie.txt
logs

6
README.md
View File

@@ -2,7 +2,7 @@
Python获取bilibili直播弹幕的库，使用WebSocket协议，支持web端和B站直播开放平台两种接口
[协议解释](https://blog.csdn.net/xfgryujk/article/details/80306776)（有点过时了，总体是没错的）
[协议解释](https://open-live.bilibili.com/document/657d8e34-f926-a133-16c0-300c1afc6e6b)
基于本库开发的一个应用：[blivechat](https://github.com/xfgryujk/blivechat)
@@ -15,4 +15,8 @@ Python获取bilibili直播弹幕的库，使用WebSocket协议，支持web端和
pip install -r requirements.txt
```
<<<<<<< HEAD
3. 例程看[sample.py](./main.py)和[open_live_sample.py](./open_live_sample.py)
=======
3. web端例程在[sample.py](./sample.py)，B站直播开放平台例程在[open_live_sample.py](./open_live_sample.py)
>>>>>>> github/HEAD

2
blivedm/__init__.py
View File

@@ -1,3 +1,5 @@
# -*- coding: utf-8 -*-
__version__ = '1.1.2'
from .handlers import *
from .clients import *

23
blivedm/clients/open_live.py
View File

@@ -60,6 +60,8 @@ class OpenLiveClient(ws_base.WebSocketClientBase):
# 在调用init_room后初始化的字段
self._room_owner_uid: Optional[int] = None
"""主播用户ID"""
self._room_owner_open_id: Optional[str] = None
"""主播Open ID"""
self._host_server_url_list: Optional[List[str]] = []
"""弹幕服务器URL列表"""
self._auth_body: Optional[str] = None
@@ -78,6 +80,13 @@ class OpenLiveClient(ws_base.WebSocketClientBase):
"""
return self._room_owner_uid
@property
def room_owner_open_id(self) -> Optional[str]:
"""
主播Open ID，调用init_room后初始化
"""
return self._room_owner_open_id
@property
def room_owner_auth_code(self):
"""
@@ -181,6 +190,7 @@ class OpenLiveClient(ws_base.WebSocketClientBase):
anchor_info = data['anchor_info']
self._room_id = anchor_info['room_id']
self._room_owner_uid = anchor_info['uid']
self._room_owner_open_id = anchor_info['open_id']
return True
async def _end_game(self):
@@ -281,3 +291,16 @@ class OpenLiveClient(ws_base.WebSocketClientBase):
发送认证包
"""
await self._websocket.send_bytes(self._make_packet(self._auth_body, ws_base.Operation.AUTH))
def _handle_command(self, command: dict):
cmd = command.get('cmd', '')
if cmd == 'LIVE_OPEN_PLATFORM_INTERACTION_END' and command['data']['game_id'] == self._game_id:
# 服务器主动停止推送，可能是心跳超时，需要重新开启项目
logger.warning('room=%d game end by server, game_id=%s', self._room_id, self._game_id)
self._need_init_room = True
if self._websocket is not None and not self._websocket.closed:
asyncio.create_task(self._websocket.close())
return
super()._handle_command(command)

11
blivedm/clients/web.py
View File

@@ -16,8 +16,8 @@ __all__ = (
logger = logging.getLogger('blivedm')
UID_INIT_URL = 'https://api.bilibili.com/x/web-interface/nav'
BUVID_INIT_URL = 'https://data.bilibili.com/v/'
ROOM_INIT_URL = 'https://api.live.bilibili.com/xlive/web-room/v1/index/getInfoByRoom'
BUVID_INIT_URL = 'https://www.bilibili.com/'
ROOM_INIT_URL = 'https://api.live.bilibili.com/room/v1/Room/get_info'
DANMAKU_SERVER_CONF_URL = 'https://api.live.bilibili.com/xlive/web-room/v1/index/getDanmuInfo'
DEFAULT_DANMAKU_SERVER_LIST = [
{'host': 'broadcastlv.chat.bilibili.com', 'port': 2243, 'wss_port': 443, 'ws_port': 2244}
@@ -196,11 +196,8 @@ class BLiveClient(ws_base.WebSocketClientBase):
return True
def _parse_room_init(self, data):
room_info = data['room_info']
self._room_id = room_info['room_id']
self._room_owner_uid = room_info['uid']
self.live_status = room_info['live_status']
self.live_start_time = room_info['live_start_time']
self._room_id = data['room_id']
self._room_owner_uid = data['uid']
return True
async def _init_host_server(self):

100
blivedm/handlers.py
View File

@@ -17,7 +17,6 @@ logged_unknown_cmds = {
'ENTRY_EFFECT',
'HOT_RANK_CHANGED',
'HOT_RANK_CHANGED_V2',
'INTERACT_WORD',
'LIVE',
'LIVE_INTERACTIVE_GAME',
'NOTICE_MSG',
@@ -35,6 +34,7 @@ logged_unknown_cmds = {
'ROOM_REAL_TIME_MESSAGE_UPDATE',
'STOP_LIVE_ROOM_LIST',
'SUPER_CHAT_MESSAGE_JPN',
'USER_TOAST_MSG',
'WIDGET_BANNER',
}
"""已打日志的未知cmd"""
@@ -76,30 +76,36 @@ class BaseHandler(HandlerInterface):
['BaseHandler', ws_base.WebSocketClientBase, dict],
Any
]]
] = {
]
"""cmd -> 处理回调"""
_CMD_CALLBACK_DICT = {
# 收到心跳包，这是blivedm自造的消息，原本的心跳包格式不一样
'_HEARTBEAT': _make_msg_callback('_on_heartbeat', web_models.HeartbeatMessage),
# 收到弹幕
# 弹幕
# go-common\app\service\live\live-dm\service\v1\send.go
'DANMU_MSG': __danmu_msg_callback,
# 有人送礼
# 礼物
'SEND_GIFT': _make_msg_callback('_on_gift', web_models.GiftMessage),
# 有人上舰
# 上舰
'GUARD_BUY': _make_msg_callback('_on_buy_guard', web_models.GuardBuyMessage),
# 另一个上舰消息
'USER_TOAST_MSG_V2': _make_msg_callback('_on_user_toast_v2', web_models.UserToastV2Message),
# 醒目留言
'SUPER_CHAT_MESSAGE': _make_msg_callback('_on_super_chat', web_models.SuperChatMessage),
# 删除醒目留言
'SUPER_CHAT_MESSAGE_DELETE': _make_msg_callback('_on_super_chat_delete', web_models.SuperChatDeleteMessage),
# 进入房间、关注主播等互动消息
'INTERACT_WORD': _make_msg_callback('_on_interact_word', web_models.InteractWordMessage),
#
# 开放平台消息
#
# 收到弹幕
# 弹幕
'LIVE_OPEN_PLATFORM_DM': _make_msg_callback('_on_open_live_danmaku', open_models.DanmakuMessage),
# 有人送礼
# 礼物
'LIVE_OPEN_PLATFORM_SEND_GIFT': _make_msg_callback('_on_open_live_gift', open_models.GiftMessage),
# 有人上舰
# 上舰
'LIVE_OPEN_PLATFORM_GUARD': _make_msg_callback('_on_open_live_buy_guard', open_models.GuardBuyMessage),
# 醒目留言
'LIVE_OPEN_PLATFORM_SUPER_CHAT': _make_msg_callback('_on_open_live_super_chat', open_models.SuperChatMessage),
@@ -109,8 +115,13 @@ class BaseHandler(HandlerInterface):
),
# 点赞
'LIVE_OPEN_PLATFORM_LIKE': _make_msg_callback('_on_open_live_like', open_models.LikeMessage),
# 进入房间
'LIVE_OPEN_PLATFORM_LIVE_ROOM_ENTER': _make_msg_callback('_on_open_live_enter_room', open_models.RoomEnterMessage),
# 开始直播
'LIVE_OPEN_PLATFORM_LIVE_START': _make_msg_callback('_on_open_live_start_live', open_models.LiveStartMessage),
# 结束直播
'LIVE_OPEN_PLATFORM_LIVE_END': _make_msg_callback('_on_open_live_end_live', open_models.LiveEndMessage),
}
"""cmd -> 处理回调"""
def handle(self, client: ws_base.WebSocketClientBase, command: dict):
cmd = command.get('cmd', '')
@@ -135,71 +146,58 @@ class BaseHandler(HandlerInterface):
pass
def _on_heartbeat(self, client: ws_base.WebSocketClientBase, message: web_models.HeartbeatMessage):
"""
收到心跳包
"""
"""收到心跳包"""
def _on_danmaku(self, client: ws_base.WebSocketClientBase, message: web_models.DanmakuMessage):
"""
收到弹幕
"""
"""弹幕"""
def _on_gift(self, client: ws_base.WebSocketClientBase, message: web_models.GiftMessage):
"""
收到礼物
"""
"""礼物"""
def _on_buy_guard(self, client: ws_base.WebSocketClientBase, message: web_models.GuardBuyMessage):
"""
有人上舰
"""
"""上舰"""
def _on_user_toast_v2(self, client: ws_base.WebSocketClientBase, message: web_models.UserToastV2Message):
"""另一个上舰消息"""
def _on_super_chat(self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatMessage):
"""
醒目留言
"""
"""醒目留言"""
def _on_super_chat_delete(
self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatDeleteMessage
):
"""
删除醒目留言
"""
def _on_super_chat_delete(self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatDeleteMessage):
"""删除醒目留言"""
def _on_interact_word(self, client: ws_base.WebSocketClientBase, message: web_models.InteractWordMessage):
"""进入房间、关注主播等互动消息"""
#
# 开放平台消息
#
def _on_open_live_danmaku(self, client: ws_base.WebSocketClientBase, message: open_models.DanmakuMessage):
"""
收到弹幕
"""
"""弹幕"""
def _on_open_live_gift(self, client: ws_base.WebSocketClientBase, message: open_models.GiftMessage):
"""
收到礼物
"""
"""礼物"""
def _on_open_live_buy_guard(self, client: ws_base.WebSocketClientBase, message: open_models.GuardBuyMessage):
"""
有人上舰
"""
"""上舰"""
def _on_open_live_super_chat(
self, client: ws_base.WebSocketClientBase, message: open_models.SuperChatMessage
):
"""
醒目留言
"""
def _on_open_live_super_chat(self, client: ws_base.WebSocketClientBase, message: open_models.SuperChatMessage):
"""醒目留言"""
def _on_open_live_super_chat_delete(
self, client: ws_base.WebSocketClientBase, message: open_models.SuperChatDeleteMessage
):
"""
删除醒目留言
"""
"""删除醒目留言"""
def _on_open_live_like(self, client: ws_base.WebSocketClientBase, message: open_models.LikeMessage):
"""
点赞
"""
"""点赞"""
def _on_open_live_enter_room(self, client: ws_base.WebSocketClientBase, message: open_models.RoomEnterMessage):
"""进入房间"""
def _on_open_live_start_live(self, client: ws_base.WebSocketClientBase, message: open_models.LiveStartMessage):
"""开始直播"""
def _on_open_live_end_live(self, client: ws_base.WebSocketClientBase, message: open_models.LiveEndMessage):
"""结束直播"""

166
blivedm/models/open_live.py
View File

@@ -23,8 +23,8 @@ class DanmakuMessage:
uname: str = ''
"""用户昵称"""
uid: int = 0
"""用户UID"""
open_id: str = ''
"""用户唯一标识"""
uface: str = ''
"""用户头像"""
timestamp: int = 0
@@ -47,12 +47,20 @@ class DanmakuMessage:
"""表情包图片地址"""
dm_type: int = 0
"""弹幕类型 0：普通弹幕 1：表情包弹幕"""
glory_level: int = 0
"""直播荣耀等级"""
reply_open_id: str = ''
"""被at用户唯一标识"""
reply_uname: str = ''
"""被at的用户昵称"""
is_admin: int = 0
"""发送弹幕的用户是否是房管，取值范围0或1，取值为1时是房管"""
@classmethod
def from_command(cls, data: dict):
return cls(
uname=data['uname'],
uid=data['uid'],
open_id=data['open_id'],
uface=data['uface'],
timestamp=data['timestamp'],
room_id=data['room_id'],
@@ -64,6 +72,10 @@ class DanmakuMessage:
fans_medal_level=data['fans_medal_level'],
emoji_img_url=data['emoji_img_url'],
dm_type=data['dm_type'],
glory_level=data['glory_level'],
reply_open_id=data['reply_open_id'],
reply_uname=data['reply_uname'],
is_admin=data['is_admin'],
)
@@ -75,6 +87,8 @@ class AnchorInfo:
uid: int = 0
"""收礼主播uid"""
open_id: str = ''
"""收礼主播唯一标识"""
uname: str = ''
"""收礼主播昵称"""
uface: str = ''
@@ -84,6 +98,7 @@ class AnchorInfo:
def from_dict(cls, data: dict):
return cls(
uid=data['uid'],
open_id=data['open_id'],
uname=data['uname'],
uface=data['uface'],
)
@@ -122,8 +137,8 @@ class GiftMessage:
room_id: int = 0
"""房间号"""
uid: int = 0
"""送礼用户UID"""
open_id: str = ''
"""用户唯一标识"""
uname: str = ''
"""送礼用户昵称"""
uface: str = ''
@@ -135,7 +150,20 @@ class GiftMessage:
gift_num: int = 0
"""赠送道具数量"""
price: int = 0
"""（礼物单价）支付金额(1000 = 1元 = 10电池),盲盒:爆出道具的价值""" # 这个B 站文档又不写清楚是单价还是总价
"""
礼物爆出单价，(1000 = 1元 = 10电池),盲盒:爆出道具的价值
注意：
- 免费礼物这个字段也可能不是0，而是银瓜子数
- 有些打折礼物这里不是实际支付的价值，实际价值应该用 `r_price`
"""
r_price: int = 0
"""
实际价值(1000 = 1元 = 10电池),盲盒:爆出道具的价值
注意：免费礼物这个字段也可能不是0
"""
paid: bool = False
"""是否是付费道具"""
fans_medal_level: int = 0
@@ -169,13 +197,14 @@ class GiftMessage:
return cls(
room_id=data['room_id'],
uid=data['uid'],
open_id=data['open_id'],
uname=data['uname'],
uface=data['uface'],
gift_id=data['gift_id'],
gift_name=data['gift_name'],
gift_num=data['gift_num'],
price=data['price'],
r_price=data['r_price'],
paid=data['paid'],
fans_medal_level=data['fans_medal_level'],
fans_medal_name=data['fans_medal_name'],
@@ -196,8 +225,8 @@ class UserInfo:
用户信息
"""
uid: int = 0
"""用户uid"""
open_id: str = ''
"""用户唯一标识"""
uname: str = ''
"""用户昵称"""
uface: str = ''
@@ -206,7 +235,7 @@ class UserInfo:
@classmethod
def from_dict(cls, data: dict):
return cls(
uid=data['uid'],
open_id=data['open_id'],
uname=data['uname'],
uface=data['uface'],
)
@@ -225,7 +254,9 @@ class GuardBuyMessage:
guard_num: int = 0
"""大航海数量"""
guard_unit: str = ''
"""大航海单位"""
"""大航海单位(正常单位为“月”，如为其他内容，无视`guard_num`以本字段内容为准，例如`*3天`)"""
price: int = 0
"""大航海金瓜子"""
fans_medal_level: int = 0
"""粉丝勋章等级"""
fans_medal_name: str = ''
@@ -246,6 +277,7 @@ class GuardBuyMessage:
guard_level=data['guard_level'],
guard_num=data['guard_num'],
guard_unit=data['guard_unit'],
price=data['price'],
fans_medal_level=data['fans_medal_level'],
fans_medal_name=data['fans_medal_name'],
fans_medal_wearing_status=data['fans_medal_wearing_status'],
@@ -263,8 +295,8 @@ class SuperChatMessage:
room_id: int = 0
"""直播间id"""
uid: int = 0
"""购买用户UID"""
open_id: str = ''
"""用户唯一标识"""
uname: str = ''
"""购买的用户昵称"""
uface: str = ''
@@ -296,7 +328,7 @@ class SuperChatMessage:
def from_command(cls, data: dict):
return cls(
room_id=data['room_id'],
uid=data['uid'],
open_id=data['open_id'],
uname=data['uname'],
uface=data['uface'],
message_id=data['message_id'],
@@ -340,13 +372,16 @@ class LikeMessage:
"""
点赞消息
请注意：用户端每分钟触发若干次的情况下只会推送一次该消息
请注意：
- 只有房间处于开播中，才会触发点赞事件
- 对单一用户最近2秒聚合发送一次点赞次数
"""
uname: str = ''
"""用户昵称"""
uid: int = 0
"""用户UID"""
open_id: str = ''
"""用户唯一标识"""
uface: str = ''
"""用户头像"""
timestamp: int = 0
@@ -355,7 +390,7 @@ class LikeMessage:
"""发生的直播间"""
like_text: str = ''
"""点赞文案(“xxx点赞了”)"""
like_count: int = 0 # 官方文档把这个字段名打错了，这个B文档真是一点都靠不住
like_count: int = 0
"""对单个用户最近2秒的点赞次数聚合"""
fans_medal_wearing_status: bool = False
"""该房间粉丝勋章佩戴情况"""
@@ -371,7 +406,7 @@ class LikeMessage:
def from_command(cls, data: dict):
return cls(
uname=data['uname'],
uid=data['uid'],
open_id=data['open_id'],
uface=data['uface'],
timestamp=data['timestamp'],
room_id=data['room_id'],
@@ -382,3 +417,96 @@ class LikeMessage:
fans_medal_level=data['fans_medal_level'],
msg_id=data.get('msg_id', ''), # 官方文档表格里没列出这个字段，但是参考JSON里面有
)
@dataclasses.dataclass
class RoomEnterMessage:
"""
进入房间消息
"""
room_id: int = 0
"""直播间id"""
uface: str = ''
"""用户头像"""
uname: str = ''
"""用户昵称"""
open_id: str = ''
"""用户唯一标识"""
timestamp: int = 0
"""发生的时间戳"""
msg_id: str = '' # 官方文档表格里没列出这个字段，但是实际上有
"""消息唯一id"""
@classmethod
def from_command(cls, data: dict):
return cls(
room_id=data['room_id'],
uface=data['uface'],
uname=data['uname'],
open_id=data['open_id'],
timestamp=data['timestamp'],
msg_id=data.get('msg_id', ''), # 官方文档表格里没列出这个字段，但是实际上有
)
@dataclasses.dataclass
class LiveStartMessage:
"""
开始直播消息
"""
room_id: int = 0
"""直播间id"""
open_id: str = ''
"""用户唯一标识"""
timestamp: int = 0
"""发生的时间戳"""
area_name: str = ''
"""开播二级分区名"""
title: str = ''
"""开播时刻，直播间的标题"""
msg_id: str = '' # 官方文档表格里没列出这个字段，但是实际上有
"""消息唯一id"""
@classmethod
def from_command(cls, data: dict):
return cls(
room_id=data['room_id'],
open_id=data['open_id'],
timestamp=data['timestamp'],
area_name=data['area_name'],
title=data['title'],
msg_id=data.get('msg_id', ''), # 官方文档表格里没列出这个字段，但是实际上有
)
@dataclasses.dataclass
class LiveEndMessage:
"""
结束直播消息
"""
room_id: int = 0
"""直播间id"""
open_id: str = ''
"""用户唯一标识"""
timestamp: int = 0
"""发生的时间戳"""
area_name: str = ''
"""开播二级分区名"""
title: str = ''
"""开播时刻，直播间的标题"""
msg_id: str = '' # 官方文档表格里没列出这个字段，但是实际上有
"""消息唯一id"""
@classmethod
def from_command(cls, data: dict):
return cls(
room_id=data['room_id'],
open_id=data['open_id'],
timestamp=data['timestamp'],
area_name=data['area_name'],
title=data['title'],
msg_id=data.get('msg_id', ''), # 官方文档表格里没列出这个字段，但是实际上有
)

201
blivedm/models/web.py
View File

@@ -67,6 +67,8 @@ class DanmakuMessage:
"""用户ID"""
uname: str = ''
"""用户名"""
face: str = ''
"""用户头像URL"""
admin: int = 0
"""是否房管"""
vip: int = 0
@@ -80,7 +82,7 @@ class DanmakuMessage:
uname_color: str = ''
"""用户名颜色"""
medal_level: str = ''
medal_level: int = 0
"""勋章等级"""
medal_name: str = ''
"""勋章名"""
@@ -108,23 +110,39 @@ class DanmakuMessage:
privilege_type: int = 0
"""舰队类型，0非舰队，1总督，2提督，3舰长"""
wealth_level: int = 0
"""荣耀等级"""
@classmethod
def from_command(cls, info: list):
mode_info = info[0][15]
try:
face = mode_info['user']['base']['face']
except (TypeError, KeyError):
face = ''
if len(info[3]) != 0:
medal_level = info[3][0]
medal_name = info[3][1]
runame = info[3][2]
room_id = info[3][3]
medal_room_id = info[3][3]
mcolor = info[3][4]
special_medal = info[3][5]
else:
medal_level = 0
medal_name = ''
runame = ''
room_id = 0
medal_room_id = 0
mcolor = 0
special_medal = 0
if len(info[5]) != 0:
old_title = info[5][0]
title = info[5][1]
else:
old_title = ''
title = ''
return cls(
mode=info[0][1],
font_size=info[0][2],
@@ -137,12 +155,13 @@ class DanmakuMessage:
dm_type=info[0][12],
emoticon_options=info[0][13],
voice_config=info[0][14],
mode_info=info[0][15],
mode_info=mode_info,
msg=info[1],
uid=info[2][0],
uname=info[2][1],
face=face,
admin=info[2][2],
vip=info[2][3],
svip=info[2][4],
@@ -153,7 +172,7 @@ class DanmakuMessage:
medal_level=medal_level,
medal_name=medal_name,
runame=runame,
medal_room_id=room_id,
medal_room_id=medal_room_id,
mcolor=mcolor,
special_medal=special_medal,
@@ -161,18 +180,23 @@ class DanmakuMessage:
ulevel_color=info[4][2],
ulevel_rank=info[4][3],
old_title=info[5][0],
title=info[5][1],
old_title=old_title,
title=title,
privilege_type=info[7],
wealth_level=info[16][0],
)
@property
def emoticon_options_dict(self) -> dict:
"""
示例：
```
{'bulge_display': 0, 'emoticon_unique': 'official_13', 'height': 60, 'in_player_area': 1, 'is_dynamic': 1,
'url': 'https://i0.hdslb.com/bfs/live/a98e35996545509188fe4d24bd1a56518ea5af48.png', 'width': 183}
```
"""
if isinstance(self.emoticon_options, dict):
return self.emoticon_options
@@ -185,11 +209,14 @@ class DanmakuMessage:
def voice_config_dict(self) -> dict:
"""
示例：
```
{'voice_url': 'https%3A%2F%2Fboss.hdslb.com%2Flive-dm-voice%2Fb5b26e48b556915cbf3312a59d3bb2561627725945.wav
%3FX-Amz-Algorithm%3DAWS4-HMAC-SHA256%26X-Amz-Credential%3D2663ba902868f12f%252F20210731%252Fshjd%252Fs3%25
2Faws4_request%26X-Amz-Date%3D20210731T100545Z%26X-Amz-Expires%3D600000%26X-Amz-SignedHeaders%3Dhost%26
X-Amz-Signature%3D114e7cb5ac91c72e231c26d8ca211e53914722f36309b861a6409ffb20f07ab8',
'file_format': 'wav', 'text': '汤，下午好。', 'file_duration': 1}
```
"""
if isinstance(self.voice_config, dict):
return self.voice_config
@@ -198,6 +225,30 @@ class DanmakuMessage:
except (json.JSONDecodeError, TypeError):
return {}
@property
def extra_dict(self) -> dict:
"""
示例：
```
{'send_from_me': False, 'mode': 0, 'color': 14893055, 'dm_type': 0, 'font_size': 25, 'player_mode': 4,
'show_player_type': 0, 'content': '确实', 'user_hash': '2904574201', 'emoticon_unique': '', 'bulge_display': 0,
'recommend_score': 5, 'main_state_dm_color': '', 'objective_state_dm_color': '', 'direction': 0,
'pk_direction': 0, 'quartet_direction': 0, 'anniversary_crowd': 0, 'yeah_space_type': '', 'yeah_space_url': '',
'jump_to_url': '', 'space_type': '', 'space_url': '', 'animation': {}, 'emots': None, 'is_audited': False,
'id_str': '6fa9959ab8feabcd1b337aa5066768334027', 'icon': None, 'show_reply': True, 'reply_mid': 0,
'reply_uname': '', 'reply_uname_color': '', 'reply_is_mystery': False, 'reply_type_enum': 0, 'hit_combo': 0,
'esports_jump_url': ''}
```
"""
try:
extra = self.mode_info['extra']
if isinstance(extra, dict):
return extra
return json.loads(extra)
except (KeyError, json.JSONDecodeError, TypeError):
return {}
@dataclasses.dataclass
class GiftMessage:
@@ -223,6 +274,8 @@ class GiftMessage:
"""礼物ID"""
gift_type: int = 0
"""礼物类型（未知）"""
gift_img_basic: str = ''
"""图标URL"""
action: str = ''
"""目前遇到的有'喂食'、'赠送'"""
price: int = 0
@@ -235,9 +288,29 @@ class GiftMessage:
"""总瓜子数"""
tid: str = ''
"""可能是事务ID，有时和rnd相同"""
medal_level: int = 0
"""勋章等级"""
medal_name: str = ''
"""勋章名"""
medal_room_id: int = 0
"""勋章房间ID，未登录时是0"""
medal_ruid: int = 0
"""勋章主播ID"""
@classmethod
def from_command(cls, data: dict):
medal_info = data.get('medal_info', None)
if medal_info is not None:
medal_level = medal_info['medal_level']
medal_name = medal_info['medal_name']
medal_room_id = medal_info['anchor_roomid']
medal_ruid = medal_info['target_id']
else:
medal_level = 0
medal_name = ''
medal_room_id = 0
medal_ruid = 0
return cls(
gift_name=data['giftName'],
num=data['num'],
@@ -248,12 +321,17 @@ class GiftMessage:
timestamp=data['timestamp'],
gift_id=data['giftId'],
gift_type=data['giftType'],
gift_img_basic=data['gift_info']['img_basic'],
action=data['action'],
price=data['price'],
rnd=data['rnd'],
coin_type=data['coin_type'],
total_coin=data['total_coin'],
tid=data['tid'],
medal_level=medal_level,
medal_name=medal_name,
medal_room_id=medal_room_id,
medal_ruid=medal_ruid,
)
@@ -269,7 +347,7 @@ class GuardBuyMessage:
"""用户名"""
guard_level: int = 0
"""舰队等级，0非舰队，1总督，2提督，3舰长"""
num: int = 0
num: int = 0 # 可以理解为礼物数量？
"""数量"""
price: int = 0
"""单价金瓜子数"""
@@ -297,6 +375,57 @@ class GuardBuyMessage:
)
@dataclasses.dataclass
class UserToastV2Message:
"""
另一个上舰消息，包含的数据更多
"""
uid: int = 0
"""用户ID"""
username: str = ''
"""用户名"""
guard_level: int = 0
"""舰队等级，0非舰队，1总督，2提督，3舰长"""
num: int = 0 # 可以理解为礼物数量？
"""数量"""
price: int = 0
"""单价金瓜子数"""
unit: str = ''
"""单位，根据开放平台的文档，正常单位为“月”，如为其他内容，无视`guard_num`以本字段内容为准，例如`*3天`"""
gift_id: int = 0
"""礼物ID"""
start_time: int = 0
"""开始时间戳，和结束时间戳相同"""
end_time: int = 0
"""结束时间戳，和开始时间戳相同"""
source: int = 0
"""猜测0是自己买的，2是别人送的，这个只影响是否播动画"""
toast_msg: str = ''
"""提示信息（"<%XXX%> 在主播XXX的直播间续费了舰长，今天是TA陪伴主播的第XXX天"）"""
@classmethod
def from_command(cls, data: dict):
sender_info = data['sender_uinfo']
guard_info = data['guard_info']
pay_info = data['pay_info']
gift_info = data['gift_info']
option = data['option']
return cls(
uid=sender_info['uid'],
username=sender_info['base']['name'],
guard_level=guard_info['guard_level'],
num=pay_info['num'],
price=pay_info['price'],
unit=pay_info['unit'],
gift_id=gift_info['gift_id'],
start_time=guard_info['start_time'],
end_time=guard_info['end_time'],
source=option['source'],
toast_msg=data['toast_msg'],
)
@dataclasses.dataclass
class SuperChatMessage:
"""
@@ -308,7 +437,7 @@ class SuperChatMessage:
message: str = ''
"""消息"""
message_trans: str = ''
"""消息日文翻译（目前只出现在SUPER_CHAT_MESSAGE_JPN）"""
"""消息日文翻译"""
start_time: int = 0
"""开始时间戳"""
end_time: int = 0
@@ -341,9 +470,29 @@ class SuperChatMessage:
"""背景图URL"""
background_price_color: str = ''
"""背景价格颜色，'#rrggbb'"""
medal_level: int = 0
"""勋章等级"""
medal_name: str = ''
"""勋章名"""
medal_room_id: int = 0
"""勋章房间ID"""
medal_ruid: int = 0
"""勋章主播ID"""
@classmethod
def from_command(cls, data: dict):
medal_info = data.get('medal_info', None)
if medal_info is not None:
medal_level = medal_info['medal_level']
medal_name = medal_info['medal_name']
medal_room_id = medal_info['anchor_roomid']
medal_ruid = medal_info['target_id']
else:
medal_level = 0
medal_name = ''
medal_room_id = 0
medal_ruid = 0
return cls(
price=data['price'],
message=data['message'],
@@ -364,6 +513,10 @@ class SuperChatMessage:
background_icon=data['background_icon'],
background_image=data['background_image'],
background_price_color=data['background_price_color'],
medal_level=medal_level,
medal_name=medal_name,
medal_room_id=medal_room_id,
medal_ruid=medal_ruid,
)
@@ -381,3 +534,33 @@ class SuperChatDeleteMessage:
return cls(
ids=data['ids'],
)
@dataclasses.dataclass
class InteractWordMessage:
"""
进入房间、关注主播等互动消息
"""
uid: int = 0
"""用户ID"""
username: str = ''
"""用户名"""
face: str = ''
"""用户头像URL"""
timestamp: int = 0
"""时间戳"""
msg_type: int = 0
"""`{1: '进入', 2: '关注了', 3: '分享了', 4: '特别关注了', 5: '互粉了', 6: '为主播点赞了'}`"""
@classmethod
def from_command(cls, data: dict):
user_info = data['uinfo']
user_base_info = user_info['base']
return cls(
uid=user_info['uid'],
username=user_base_info['name'],
face=user_base_info['face'],
timestamp=data['timestamp'],
msg_type=data['msg_type'],
)

9
open_live_sample.py
View File

@@ -71,6 +71,15 @@ class MyHandler(blivedm.BaseHandler):
def _on_open_live_like(self, client: blivedm.OpenLiveClient, message: open_models.LikeMessage):
print(f'[{message.room_id}] {message.uname} 点赞')
def _on_open_live_enter_room(self, client: blivedm.OpenLiveClient, message: open_models.RoomEnterMessage):
print(f'[{message.room_id}] {message.uname} 进入房间')
def _on_open_live_start_live(self, client: blivedm.OpenLiveClient, message: open_models.LiveStartMessage):
print(f'[{message.room_id}] 开始直播')
def _on_open_live_end_live(self, client: blivedm.OpenLiveClient, message: open_models.LiveEndMessage):
print(f'[{message.room_id}] 结束直播')
if __name__ == '__main__':
asyncio.run(main())

46
pyproject.toml Normal file
View File

@@ -0,0 +1,46 @@
[build-system]
requires = ["pdm-backend"]
build-backend = "pdm.backend"
[project]
name = "blivedm"
dynamic = ["version"]
description = "Python获取bilibili直播弹幕的库，使用WebSocket协议"
readme = "README.md"
keywords = ["bilibili", "bilibili-live", "danmaku"]
requires-python = ">=3.8"
authors = [
{name = "xfgryujk", email = "xfgryujk@126.com"},
]
license = {file = "LICENSE"}
classifiers = [
"Development Status :: 5 - Production/Stable",
"Framework :: AsyncIO",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: Internet :: WWW/HTTP",
"Topic :: Software Development :: Libraries",
]
dependencies = [
"aiohttp~=3.9.0",
"Brotli~=1.1.0",
"yarl~=1.9.3",
]
[project.urls]
Homepage = "https://github.com/xfgryujk/blivedm"
Repository = "https://github.com/xfgryujk/blivedm"
Issues = "https://github.com/xfgryujk/blivedm/issues"
[tool.pdm]
version = {source = "file", path = "blivedm/__init__.py"}
distribution = true

120
sample.py Normal file
View File

@@ -0,0 +1,120 @@
# -*- coding: utf-8 -*-
import asyncio
import http.cookies
import random
from typing import *
import aiohttp
import blivedm
import blivedm.models.web as web_models
# 直播间ID的取值看直播间URL
TEST_ROOM_IDS = [
12235923,
14327465,
21396545,
21449083,
23105590,
]
# 这里填一个已登录账号的cookie的SESSDATA字段的值。不填也可以连接，但是收到弹幕的用户名会打码，UID会变成0
SESSDATA = ''
session: Optional[aiohttp.ClientSession] = None
async def main():
init_session()
try:
await run_single_client()
await run_multi_clients()
finally:
await session.close()
def init_session():
cookies = http.cookies.SimpleCookie()
cookies['SESSDATA'] = SESSDATA
cookies['SESSDATA']['domain'] = 'bilibili.com'
global session
session = aiohttp.ClientSession()
session.cookie_jar.update_cookies(cookies)
async def run_single_client():
"""
演示监听一个直播间
"""
room_id = random.choice(TEST_ROOM_IDS)
client = blivedm.BLiveClient(room_id, session=session)
handler = MyHandler()
client.set_handler(handler)
client.start()
try:
# 演示5秒后停止
await asyncio.sleep(5)
client.stop()
await client.join()
finally:
await client.stop_and_close()
async def run_multi_clients():
"""
演示同时监听多个直播间
"""
clients = [blivedm.BLiveClient(room_id, session=session) for room_id in TEST_ROOM_IDS]
handler = MyHandler()
for client in clients:
client.set_handler(handler)
client.start()
try:
await asyncio.gather(*(
client.join() for client in clients
))
finally:
await asyncio.gather(*(
client.stop_and_close() for client in clients
))
class MyHandler(blivedm.BaseHandler):
# # 演示如何添加自定义回调
# _CMD_CALLBACK_DICT = blivedm.BaseHandler._CMD_CALLBACK_DICT.copy()
#
# # 看过数消息回调
# def __watched_change_callback(self, client: blivedm.BLiveClient, command: dict):
# print(f'[{client.room_id}] WATCHED_CHANGE: {command}')
# _CMD_CALLBACK_DICT['WATCHED_CHANGE'] = __watched_change_callback # noqa
def _on_heartbeat(self, client: blivedm.BLiveClient, message: web_models.HeartbeatMessage):
print(f'[{client.room_id}] 心跳')
def _on_danmaku(self, client: blivedm.BLiveClient, message: web_models.DanmakuMessage):
print(f'[{client.room_id}] {message.uname}：{message.msg}')
def _on_gift(self, client: blivedm.BLiveClient, message: web_models.GiftMessage):
print(f'[{client.room_id}] {message.uname} 赠送{message.gift_name}x{message.num}'
f' （{message.coin_type}瓜子x{message.total_coin}）')
# def _on_buy_guard(self, client: blivedm.BLiveClient, message: web_models.GuardBuyMessage):
# print(f'[{client.room_id}] {message.username} 上舰，guard_level={message.guard_level}')
def _on_user_toast_v2(self, client: blivedm.BLiveClient, message: web_models.UserToastV2Message):
print(f'[{client.room_id}] {message.username} 上舰，guard_level={message.guard_level}')
def _on_super_chat(self, client: blivedm.BLiveClient, message: web_models.SuperChatMessage):
print(f'[{client.room_id}] 醒目留言 ¥{message.price} {message.uname}：{message.message}')
# def _on_interact_word(self, client: blivedm.BLiveClient, message: web_models.InteractWordMessage):
# if message.msg_type == 1:
# print(f'[{client.room_id}] {message.username} 进入房间')
if __name__ == '__main__':
asyncio.run(main())