github-repos/blivedm
1
0
Fork 0
mirror of https://github.com/xfgryujk/blivedm.git synced 2024-12-26 04:40:11 +08:00
Code Issues Projects Releases Wiki Activity

添加开放平台消息模型

Browse Source
This commit is contained in:
John Smith 2023-09-03 16:53:08 +08:00
parent ff766d68db
commit 74b9cdc100
9 changed files with 913 additions and 456 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -1,6 +1,6 @@
# blivedm
Python获取bilibili直播弹幕的库，使用WebSocket协议
Python获取bilibili直播弹幕的库，使用WebSocket协议，支持web端和B站直播开放平台两种接口
[协议解释](https://blog.csdn.net/xfgryujk/article/details/80306776)（有点过时了，总体是没错的）
@ -15,4 +15,4 @@ Python获取bilibili直播弹幕的库，使用WebSocket协议
pip install -r requirements.txt
```
3. 例程看[sample.py](./sample.py)
3. 例程看[sample.py](./sample.py)和[open_live_sample.py](./open_live_sample.py)

1
blivedm/__init__.py
View File

@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
from .models import *
from .handlers import *
from .clients import *

7
blivedm/clients/web.py
View File

@ -117,6 +117,13 @@ class BLiveClient(ws_base.WebSocketClientBase):
return res
async def _init_uid(self):
cookies = self._session.cookie_jar.filter_cookies(yarl.URL(UID_INIT_URL))
sessdata_cookie = cookies.get('SESSDATA', None)
if sessdata_cookie is None or sessdata_cookie.value == '':
# cookie都没有，不用请求了
self._uid = 0
return True
try:
async with self._session.get(
UID_INIT_URL,

122
blivedm/handlers.py
View File

@ -3,7 +3,7 @@ import logging
from typing import *
from .clients import ws_base
from . import models
from .models import web as web_models, open_live as open_models
__all__ = (
'HandlerInterface',
@ -12,7 +12,7 @@ __all__ = (
logger = logging.getLogger('blivedm')
IGNORED_CMDS = (
logged_unknown_cmds = {
'COMBO_SEND',
'ENTRY_EFFECT',
'HOT_RANK_CHANGED',
@ -36,10 +36,7 @@ IGNORED_CMDS = (
'STOP_LIVE_ROOM_LIST',
'SUPER_CHAT_MESSAGE_JPN',
'WIDGET_BANNER',
)
"""常见可忽略的cmd"""
logged_unknown_cmds = set()
}
"""已打日志的未知cmd"""
@ -54,28 +51,22 @@ class HandlerInterface:
# TODO 加个异常停止的回调
def _make_msg_callback(method_name, message_cls):
def callback(self: 'BaseHandler', client: ws_base.WebSocketClientBase, command: dict):
method = getattr(self, method_name)
return method(client, message_cls.from_command(command['data']))
return callback
class BaseHandler(HandlerInterface):
"""
一个简单的消息处理器实现，带消息分发和消息类型转换。继承并重写_on_xxx方法即可实现自己的处理器
"""
def __heartbeat_callback(self, client: ws_base.WebSocketClientBase, command: dict):
return self._on_heartbeat(client, models.HeartbeatMessage.from_command(command['data']))
def __danmu_msg_callback(self, client: ws_base.WebSocketClientBase, command: dict):
return self._on_danmaku(client, models.DanmakuMessage.from_command(command['info'], command.get('dm_v2', '')))
def __send_gift_callback(self, client: ws_base.WebSocketClientBase, command: dict):
return self._on_gift(client, models.GiftMessage.from_command(command['data']))
def __guard_buy_callback(self, client: ws_base.WebSocketClientBase, command: dict):
return self._on_buy_guard(client, models.GuardBuyMessage.from_command(command['data']))
def __super_chat_message_callback(self, client: ws_base.WebSocketClientBase, command: dict):
return self._on_super_chat(client, models.SuperChatMessage.from_command(command['data']))
def __super_chat_message_delete_callback(self, client: ws_base.WebSocketClientBase, command: dict):
return self._on_super_chat_delete(client, models.SuperChatDeleteMessage.from_command(command['data']))
return self._on_danmaku(
client, web_models.DanmakuMessage.from_command(command['info'], command.get('dm_v2', ''))
)
_CMD_CALLBACK_DICT: Dict[
str,
@ -85,24 +76,39 @@ class BaseHandler(HandlerInterface):
]]
] = {
# 收到心跳包，这是blivedm自造的消息，原本的心跳包格式不一样
'_HEARTBEAT': __heartbeat_callback,
'_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': __send_gift_callback,
'SEND_GIFT': _make_msg_callback('_on_gift', web_models.GiftMessage),
# 有人上舰
'GUARD_BUY': __guard_buy_callback,
'GUARD_BUY': _make_msg_callback('_on_buy_guard', web_models.GuardBuyMessage),
# 醒目留言
'SUPER_CHAT_MESSAGE': __super_chat_message_callback,
'SUPER_CHAT_MESSAGE': _make_msg_callback('_on_super_chat', web_models.SuperChatMessage),
# 删除醒目留言
'SUPER_CHAT_MESSAGE_DELETE': __super_chat_message_delete_callback,
'SUPER_CHAT_MESSAGE_DELETE': _make_msg_callback('_on_super_chat_delete', web_models.SuperChatDeleteMessage),
#
# 开放平台消息
#
# 收到弹幕
'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),
# 删除醒目留言
'LIVE_OPEN_PLATFORM_SUPER_CHAT_DEL': _make_msg_callback(
'_on_open_live_super_chat_delete', open_models.SuperChatDeleteMessage
),
# 点赞
'LIVE_OPEN_PLATFORM_LIKE': _make_msg_callback('_on_open_live_like', open_models.LikeMessage),
}
"""cmd -> 处理回调"""
# 忽略其他常见cmd
for cmd in IGNORED_CMDS:
_CMD_CALLBACK_DICT[cmd] = None
del cmd
async def handle(self, client: ws_base.WebSocketClientBase, command: dict):
cmd = command.get('cmd', '')
@ -121,32 +127,72 @@ class BaseHandler(HandlerInterface):
if callback is not None:
await callback(self, client, command)
async def _on_heartbeat(self, client: ws_base.WebSocketClientBase, message: models.HeartbeatMessage):
async def _on_heartbeat(self, client: ws_base.WebSocketClientBase, message: web_models.HeartbeatMessage):
"""
收到心跳包（人气值）
收到心跳包
"""
async def _on_danmaku(self, client: ws_base.WebSocketClientBase, message: models.DanmakuMessage):
async def _on_danmaku(self, client: ws_base.WebSocketClientBase, message: web_models.DanmakuMessage):
"""
收到弹幕
"""
async def _on_gift(self, client: ws_base.WebSocketClientBase, message: models.GiftMessage):
async def _on_gift(self, client: ws_base.WebSocketClientBase, message: web_models.GiftMessage):
"""
收到礼物
"""
async def _on_buy_guard(self, client: ws_base.WebSocketClientBase, message: models.GuardBuyMessage):
async def _on_buy_guard(self, client: ws_base.WebSocketClientBase, message: web_models.GuardBuyMessage):
"""
有人上舰
"""
async def _on_super_chat(self, client: ws_base.WebSocketClientBase, message: models.SuperChatMessage):
async def _on_super_chat(self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatMessage):
"""
醒目留言
"""
async def _on_super_chat_delete(self, client: ws_base.WebSocketClientBase, message: models.SuperChatDeleteMessage):
async def _on_super_chat_delete(
self, client: ws_base.WebSocketClientBase, message: web_models.SuperChatDeleteMessage
):
"""
删除醒目留言
"""
#
# 开放平台消息
#
async def _on_open_live_danmaku(self, client: ws_base.WebSocketClientBase, message: open_models.DanmakuMessage):
"""
收到弹幕
"""
async def _on_open_live_gift(self, client: ws_base.WebSocketClientBase, message: open_models.GiftMessage):
"""
收到礼物
"""
async def _on_open_live_buy_guard(self, client: ws_base.WebSocketClientBase, message: open_models.GuardBuyMessage):
"""
有人上舰
"""
async def _on_open_live_super_chat(
self, client: ws_base.WebSocketClientBase, message: open_models.SuperChatMessage
):
"""
醒目留言
"""
async def _on_open_live_super_chat_delete(
self, client: ws_base.WebSocketClientBase, message: open_models.SuperChatDeleteMessage
):
"""
删除醒目留言
"""
async def _on_open_live_like(self, client: ws_base.WebSocketClientBase, message: open_models.LikeMessage):
"""
点赞
"""

400
blivedm/models/__init__.py
View File

@ -1,401 +1 @@
# -*- coding: utf-8 -*-
import base64
import binascii
import dataclasses
import json
from typing import *
from . import pb
__all__ = (
'HeartbeatMessage',
'DanmakuMessage',
'GiftMessage',
'GuardBuyMessage',
'SuperChatMessage',
'SuperChatDeleteMessage',
)
@dataclasses.dataclass
class HeartbeatMessage:
"""
心跳消息
"""
popularity: int = 0
"""人气值，已废弃"""
@classmethod
def from_command(cls, data: dict):
return cls(
popularity=data['popularity'],
)
@dataclasses.dataclass
class DanmakuMessage:
"""
弹幕消息
"""
mode: int = 0
"""弹幕显示模式（滚动、顶部、底部）"""
font_size: int = 0
"""字体尺寸"""
color: int = 0
"""颜色"""
timestamp: int = 0
"""时间戳（毫秒）"""
rnd: int = 0
"""随机数，前端叫作弹幕ID，可能是去重用的"""
uid_crc32: str = ''
"""用户ID文本的CRC32"""
msg_type: int = 0
"""是否礼物弹幕（节奏风暴）"""
bubble: int = 0
"""右侧评论栏气泡"""
dm_type: int = 0
"""弹幕类型，0文本，1表情，2语音"""
emoticon_options: Union[dict, str] = ''
"""表情参数"""
voice_config: Union[dict, str] = ''
"""语音参数"""
mode_info: dict = dataclasses.field(default_factory=dict)
"""一些附加参数"""
msg: str = ''
"""弹幕内容"""
uid: int = 0
"""用户ID"""
uname: str = ''
"""用户名"""
face: str = ''
"""用户头像URL"""
admin: int = 0
"""是否房管"""
vip: int = 0
"""是否月费老爷"""
svip: int = 0
"""是否年费老爷"""
urank: int = 0
"""用户身份，用来判断是否正式会员，猜测非正式会员为5000，正式会员为10000"""
mobile_verify: int = 0
"""是否绑定手机"""
uname_color: str = ''
"""用户名颜色"""
medal_level: str = ''
"""勋章等级"""
medal_name: str = ''
"""勋章名"""
runame: str = ''
"""勋章房间主播名"""
medal_room_id: int = 0
"""勋章房间ID"""
mcolor: int = 0
"""勋章颜色"""
special_medal: str = ''
"""特殊勋章"""
user_level: int = 0
"""用户等级"""
ulevel_color: int = 0
"""用户等级颜色"""
ulevel_rank: str = ''
"""用户等级排名，>50000时为'>50000'"""
old_title: str = ''
"""旧头衔"""
title: str = ''
"""头衔"""
privilege_type: int = 0
"""舰队类型，0非舰队，1总督，2提督，3舰长"""
@classmethod
def from_command(cls, info: list, dm_v2=''):
proto: Optional[pb.SimpleDm] = None
if dm_v2 != '':
try:
proto = pb.SimpleDm.loads(base64.b64decode(dm_v2))
except (binascii.Error, KeyError, TypeError, ValueError):
pass
if proto is not None:
face = proto.user.face
else:
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]
mcolor = info[3][4]
special_medal = info[3][5]
else:
medal_level = 0
medal_name = ''
runame = ''
room_id = 0
mcolor = 0
special_medal = 0
return cls(
mode=info[0][1],
font_size=info[0][2],
color=info[0][3],
timestamp=info[0][4],
rnd=info[0][5],
uid_crc32=info[0][7],
msg_type=info[0][9],
bubble=info[0][10],
dm_type=info[0][12],
emoticon_options=info[0][13],
voice_config=info[0][14],
mode_info=info[0][15],
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],
urank=info[2][5],
mobile_verify=info[2][6],
uname_color=info[2][7],
medal_level=medal_level,
medal_name=medal_name,
runame=runame,
medal_room_id=room_id,
mcolor=mcolor,
special_medal=special_medal,
user_level=info[4][0],
ulevel_color=info[4][2],
ulevel_rank=info[4][3],
old_title=info[5][0],
title=info[5][1],
privilege_type=info[7],
)
@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
try:
return json.loads(self.emoticon_options)
except (json.JSONDecodeError, TypeError):
return {}
@property
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
try:
return json.loads(self.voice_config)
except (json.JSONDecodeError, TypeError):
return {}
@dataclasses.dataclass
class GiftMessage:
"""
礼物消息
"""
gift_name: str = ''
"""礼物名"""
num: int = 0
"""数量"""
uname: str = ''
"""用户名"""
face: str = ''
"""用户头像URL"""
guard_level: int = 0
"""舰队等级，0非舰队，1总督，2提督，3舰长"""
uid: int = 0
"""用户ID"""
timestamp: int = 0
"""时间戳"""
gift_id: int = 0
"""礼物ID"""
gift_type: int = 0
"""礼物类型（未知）"""
action: str = ''
"""目前遇到的有'喂食'、'赠送'"""
price: int = 0
"""礼物单价瓜子数"""
rnd: str = ''
"""随机数，可能是去重用的。有时是时间戳+去重ID，有时是UUID"""
coin_type: str = ''
"""瓜子类型，'silver'或'gold'，1000金瓜子 = 1元"""
total_coin: int = 0
"""总瓜子数"""
tid: str = ''
"""可能是事务ID，有时和rnd相同"""
@classmethod
def from_command(cls, data: dict):
return cls(
gift_name=data['giftName'],
num=data['num'],
uname=data['uname'],
face=data['face'],
guard_level=data['guard_level'],
uid=data['uid'],
timestamp=data['timestamp'],
gift_id=data['giftId'],
gift_type=data['giftType'],
action=data['action'],
price=data['price'],
rnd=data['rnd'],
coin_type=data['coin_type'],
total_coin=data['total_coin'],
tid=data['tid'],
)
@dataclasses.dataclass
class GuardBuyMessage:
"""
上舰消息
"""
uid: int = 0
"""用户ID"""
username: str = ''
"""用户名"""
guard_level: int = 0
"""舰队等级，0非舰队，1总督，2提督，3舰长"""
num: int = 0
"""数量"""
price: int = 0
"""单价金瓜子数"""
gift_id: int = 0
"""礼物ID"""
gift_name: str = ''
"""礼物名"""
start_time: int = 0
"""开始时间戳，和结束时间戳相同"""
end_time: int = 0
"""结束时间戳，和开始时间戳相同"""
@classmethod
def from_command(cls, data: dict):
return cls(
uid=data['uid'],
username=data['username'],
guard_level=data['guard_level'],
num=data['num'],
price=data['price'],
gift_id=data['gift_id'],
gift_name=data['gift_name'],
start_time=data['start_time'],
end_time=data['end_time'],
)
@dataclasses.dataclass
class SuperChatMessage:
"""
醒目留言消息
"""
price: int = 0
"""价格（人民币）"""
message: str = ''
"""消息"""
message_trans: str = ''
"""消息日文翻译（目前只出现在SUPER_CHAT_MESSAGE_JPN）"""
start_time: int = 0
"""开始时间戳"""
end_time: int = 0
"""结束时间戳"""
time: int = 0
"""剩余时间（约等于 结束时间戳 - 开始时间戳）"""
id: int = 0
"""醒目留言ID，删除时用"""
gift_id: int = 0
"""礼物ID"""
gift_name: str = ''
"""礼物名"""
uid: int = 0
"""用户ID"""
uname: str = ''
"""用户名"""
face: str = ''
"""用户头像URL"""
guard_level: int = 0
"""舰队等级，0非舰队，1总督，2提督，3舰长"""
user_level: int = 0
"""用户等级"""
background_bottom_color: str = ''
"""底部背景色，'#rrggbb'"""
background_color: str = ''
"""背景色，'#rrggbb'"""
background_icon: str = ''
"""背景图标"""
background_image: str = ''
"""背景图URL"""
background_price_color: str = ''
"""背景价格颜色，'#rrggbb'"""
@classmethod
def from_command(cls, data: dict):
return cls(
price=data['price'],
message=data['message'],
message_trans=data['message_trans'],
start_time=data['start_time'],
end_time=data['end_time'],
time=data['time'],
id=data['id'],
gift_id=data['gift']['gift_id'],
gift_name=data['gift']['gift_name'],
uid=data['uid'],
uname=data['user_info']['uname'],
face=data['user_info']['face'],
guard_level=data['user_info']['guard_level'],
user_level=data['user_info']['user_level'],
background_bottom_color=data['background_bottom_color'],
background_color=data['background_color'],
background_icon=data['background_icon'],
background_image=data['background_image'],
background_price_color=data['background_price_color'],
)
@dataclasses.dataclass
class SuperChatDeleteMessage:
"""
删除醒目留言消息
"""
ids: List[int] = dataclasses.field(default_factory=list)
"""醒目留言ID数组"""
@classmethod
def from_command(cls, data: dict):
return cls(
ids=data['ids'],
)

371
blivedm/models/open_live.py Normal file
View File

@ -0,0 +1,371 @@
# -*- coding: utf-8 -*-
import dataclasses
from typing import *
__all__ = (
'DanmakuMessage',
'GiftMessage',
'GuardBuyMessage',
'SuperChatMessage',
'SuperChatDeleteMessage',
'LikeMessage',
)
# 注释都是复制自官方文档的，看不懂的话问B站
# https://open-live.bilibili.com/document/f9ce25be-312e-1f4a-85fd-fef21f1637f8
@dataclasses.dataclass
class DanmakuMessage:
"""
弹幕消息
"""
uname: str = ''
"""用户昵称"""
uid: int = 0
"""用户UID"""
uface: str = ''
"""用户头像"""
timestamp: int = 0
"""弹幕发送时间秒级时间戳"""
room_id: int = 0
"""弹幕接收的直播间"""
msg: str = ''
"""弹幕内容"""
msg_id: str = ''
"""消息唯一id"""
guard_level: int = 0
"""对应房间大航海等级"""
fans_medal_wearing_status: bool = False
"""该房间粉丝勋章佩戴情况"""
fans_medal_name: str = ''
"""粉丝勋章名"""
fans_medal_level: int = 0
"""对应房间勋章信息"""
emoji_img_url: str = ''
"""表情包图片地址"""
dm_type: int = 0
"""弹幕类型 0：普通弹幕 1：表情包弹幕"""
@classmethod
def from_command(cls, data: dict):
return cls(
uname=data['uname'],
uid=data['uid'],
uface=data['uface'],
timestamp=data['timestamp'],
room_id=data['room_id'],
msg=data['msg'],
msg_id=data['msg_id'],
guard_level=data['guard_level'],
fans_medal_wearing_status=data['fans_medal_wearing_status'],
fans_medal_name=data['fans_medal_name'],
fans_medal_level=data['fans_medal_level'],
emoji_img_url=data['emoji_img_url'],
dm_type=data['dm_type'],
)
@dataclasses.dataclass
class AnchorInfo:
"""
主播信息
"""
uid: int = 0
"""收礼主播uid"""
uname: str = ''
"""收礼主播昵称"""
uface: str = ''
"""收礼主播头像"""
@classmethod
def from_dict(cls, data: dict):
return cls(
uid=data['uid'],
uname=data['uname'],
uface=data['uface'],
)
@dataclasses.dataclass
class ComboInfo:
"""
连击信息
"""
combo_base_num: int = 0
"""每次连击赠送的道具数量"""
combo_count: int = 0
"""连击次数"""
combo_id: str = ''
"""连击id"""
combo_timeout: int = 0
"""连击有效期秒"""
@classmethod
def from_dict(cls, data: dict):
return cls(
combo_base_num=data['combo_base_num'],
combo_count=data['combo_count'],
combo_id=data['combo_id'],
combo_timeout=data['combo_timeout'],
)
@dataclasses.dataclass
class GiftMessage:
"""
礼物消息
"""
room_id: int = 0
"""房间号"""
uid: int = 0
"""送礼用户UID"""
uname: str = ''
"""送礼用户昵称"""
uface: str = ''
"""送礼用户头像"""
gift_id: int = 0
"""道具id(盲盒:爆出道具id)"""
gift_name: str = ''
"""道具名(盲盒:爆出道具名)"""
gift_num: int = 0
"""赠送道具数量"""
price: int = 0
"""支付金额(1000 = 1元 = 10电池),盲盒:爆出道具的价值"""
paid: bool = False
"""是否是付费道具"""
fans_medal_level: int = 0
"""实际送礼人的勋章信息"""
fans_medal_name: str = ''
"""粉丝勋章名"""
fans_medal_wearing_status: bool = False
"""该房间粉丝勋章佩戴情况"""
guard_level: int = 0
"""大航海等级"""
timestamp: int = 0
"""收礼时间秒级时间戳"""
anchor_info: AnchorInfo = dataclasses.field(default_factory=AnchorInfo)
"""主播信息"""
msg_id: str = ''
"""消息唯一id"""
gift_icon: str = ''
"""道具icon"""
combo_gift: bool = False
"""是否是combo道具"""
combo_info: ComboInfo = dataclasses.field(default_factory=ComboInfo)
"""连击信息"""
@classmethod
def from_command(cls, data: dict):
return cls(
room_id=data['room_id'],
uid=data['uid'],
uname=data['uname'],
uface=data['uface'],
gift_id=data['gift_id'],
gift_name=data['gift_name'],
gift_num=data['gift_num'],
price=data['price'],
paid=data['paid'],
fans_medal_level=data['fans_medal_level'],
fans_medal_name=data['fans_medal_name'],
fans_medal_wearing_status=data['fans_medal_wearing_status'],
guard_level=data['guard_level'],
timestamp=data['timestamp'],
anchor_info=AnchorInfo.from_dict(data['anchor_info']),
msg_id=data['msg_id'],
gift_icon=data['gift_icon'],
combo_gift=data['combo_gift'],
combo_info=ComboInfo.from_dict(data['combo_info']),
)
@dataclasses.dataclass
class UserInfo:
"""
用户信息
"""
uid: int = 0
"""用户uid"""
uname: str = ''
"""用户昵称"""
uface: str = ''
"""用户头像"""
@classmethod
def from_dict(cls, data: dict):
return cls(
uid=data['uid'],
uname=data['uname'],
uface=data['uface'],
)
@dataclasses.dataclass
class GuardBuyMessage:
"""
上舰消息
"""
user_info: UserInfo = dataclasses.field(default_factory=UserInfo)
"""用户信息"""
guard_level: int = 0
"""大航海等级"""
guard_num: int = 0
"""大航海数量"""
guard_unit: str = ''
"""大航海单位"""
fans_medal_level: int = 0
"""粉丝勋章等级"""
fans_medal_name: str = ''
"""粉丝勋章名"""
fans_medal_wearing_status: bool = False
"""该房间粉丝勋章佩戴情况"""
room_id: int = 0
"""房间号"""
msg_id: str = ''
"""消息唯一id"""
timestamp: int = 0
"""上舰时间秒级时间戳"""
@classmethod
def from_command(cls, data: dict):
return cls(
user_info=UserInfo.from_dict(data['user_info']),
guard_level=data['guard_level'],
guard_num=data['guard_num'],
guard_unit=data['guard_unit'],
fans_medal_level=data['fans_medal_level'],
fans_medal_name=data['fans_medal_name'],
fans_medal_wearing_status=data['fans_medal_wearing_status'],
room_id=data['room_id'],
msg_id=data['msg_id'],
timestamp=data['timestamp'],
)
@dataclasses.dataclass
class SuperChatMessage:
"""
醒目留言消息
"""
room_id: int = 0
"""直播间id"""
uid: int = 0
"""购买用户UID"""
uname: str = ''
"""购买的用户昵称"""
uface: str = ''
"""购买用户头像"""
message_id: int = 0
"""留言id(风控场景下撤回留言需要)"""
message: str = ''
"""留言内容"""
rmb: int = 0
"""支付金额(元)"""
timestamp: int = 0
"""赠送时间秒级"""
start_time: int = 0
"""生效开始时间"""
end_time: int = 0
"""生效结束时间"""
guard_level: int = 0
"""对应房间大航海等级"""
fans_medal_level: int = 0
"""对应房间勋章信息"""
fans_medal_name: str = ''
"""对应房间勋章名字"""
fans_medal_wearing_status: bool = False
"""该房间粉丝勋章佩戴情况"""
msg_id: str = ''
"""消息唯一id"""
@classmethod
def from_command(cls, data: dict):
return cls(
room_id=data['room_id'],
uid=data['uid'],
uname=data['uname'],
uface=data['uface'],
message_id=data['message_id'],
message=data['message'],
rmb=data['rmb'],
timestamp=data['timestamp'],
start_time=data['start_time'],
end_time=data['end_time'],
guard_level=data['guard_level'],
fans_medal_level=data['fans_medal_level'],
fans_medal_name=data['fans_medal_name'],
fans_medal_wearing_status=data['fans_medal_wearing_status'],
msg_id=data['msg_id'],
)
@dataclasses.dataclass
class SuperChatDeleteMessage:
"""
删除醒目留言消息
"""
room_id: int = 0
"""直播间id"""
message_ids: List[int] = dataclasses.field(default_factory=list)
"""留言id"""
msg_id: str = ''
"""消息唯一id"""
@classmethod
def from_command(cls, data: dict):
return cls(
room_id=data['room_id'],
message_ids=data['message_ids'],
msg_id=data['msg_id'],
)
@dataclasses.dataclass
class LikeMessage:
"""
点赞消息
请注意：用户端每分钟触发若干次的情况下只会推送一次该消息
"""
uname: str = ''
"""用户昵称"""
uid: int = 0
"""用户UID"""
uface: str = ''
"""用户头像"""
timestamp: int = 0
"""时间秒级时间戳"""
room_id: int = 0
"""发生的直播间"""
like_text: str = ''
"""点赞文案(“xxx点赞了”)"""
fans_medal_wearing_status: bool = False
"""该房间粉丝勋章佩戴情况"""
fans_medal_name: str = ''
"""粉丝勋章名"""
fans_medal_level: int = 0
"""对应房间勋章信息"""
@classmethod
def from_command(cls, data: dict):
return cls(
uname=data['uname'],
uid=data['uid'],
uface=data['uface'],
timestamp=data['timestamp'],
room_id=data['room_id'],
like_text=data['like_text'],
fans_medal_wearing_status=data['fans_medal_wearing_status'],
fans_medal_name=data['fans_medal_name'],
fans_medal_level=data['fans_medal_level'],
)

401
blivedm/models/web.py Normal file
View File

@ -0,0 +1,401 @@
# -*- coding: utf-8 -*-
import base64
import binascii
import dataclasses
import json
from typing import *
from . import pb
__all__ = (
'HeartbeatMessage',
'DanmakuMessage',
'GiftMessage',
'GuardBuyMessage',
'SuperChatMessage',
'SuperChatDeleteMessage',
)
@dataclasses.dataclass
class HeartbeatMessage:
"""
心跳消息
"""
popularity: int = 0
"""人气值，已废弃"""
@classmethod
def from_command(cls, data: dict):
return cls(
popularity=data['popularity'],
)
@dataclasses.dataclass
class DanmakuMessage:
"""
弹幕消息
"""
mode: int = 0
"""弹幕显示模式（滚动、顶部、底部）"""
font_size: int = 0
"""字体尺寸"""
color: int = 0
"""颜色"""
timestamp: int = 0
"""时间戳（毫秒）"""
rnd: int = 0
"""随机数，前端叫作弹幕ID，可能是去重用的"""
uid_crc32: str = ''
"""用户ID文本的CRC32"""
msg_type: int = 0
"""是否礼物弹幕（节奏风暴）"""
bubble: int = 0
"""右侧评论栏气泡"""
dm_type: int = 0
"""弹幕类型，0文本，1表情，2语音"""
emoticon_options: Union[dict, str] = ''
"""表情参数"""
voice_config: Union[dict, str] = ''
"""语音参数"""
mode_info: dict = dataclasses.field(default_factory=dict)
"""一些附加参数"""
msg: str = ''
"""弹幕内容"""
uid: int = 0
"""用户ID"""
uname: str = ''
"""用户名"""
face: str = ''
"""用户头像URL"""
admin: int = 0
"""是否房管"""
vip: int = 0
"""是否月费老爷"""
svip: int = 0
"""是否年费老爷"""
urank: int = 0
"""用户身份，用来判断是否正式会员，猜测非正式会员为5000，正式会员为10000"""
mobile_verify: int = 0
"""是否绑定手机"""
uname_color: str = ''
"""用户名颜色"""
medal_level: str = ''
"""勋章等级"""
medal_name: str = ''
"""勋章名"""
runame: str = ''
"""勋章房间主播名"""
medal_room_id: int = 0
"""勋章房间ID"""
mcolor: int = 0
"""勋章颜色"""
special_medal: str = ''
"""特殊勋章"""
user_level: int = 0
"""用户等级"""
ulevel_color: int = 0
"""用户等级颜色"""
ulevel_rank: str = ''
"""用户等级排名，>50000时为'>50000'"""
old_title: str = ''
"""旧头衔"""
title: str = ''
"""头衔"""
privilege_type: int = 0
"""舰队类型，0非舰队，1总督，2提督，3舰长"""
@classmethod
def from_command(cls, info: list, dm_v2=''):
proto: Optional[pb.SimpleDm] = None
if dm_v2 != '':
try:
proto = pb.SimpleDm.loads(base64.b64decode(dm_v2))
except (binascii.Error, KeyError, TypeError, ValueError):
pass
if proto is not None:
face = proto.user.face
else:
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]
mcolor = info[3][4]
special_medal = info[3][5]
else:
medal_level = 0
medal_name = ''
runame = ''
room_id = 0
mcolor = 0
special_medal = 0
return cls(
mode=info[0][1],
font_size=info[0][2],
color=info[0][3],
timestamp=info[0][4],
rnd=info[0][5],
uid_crc32=info[0][7],
msg_type=info[0][9],
bubble=info[0][10],
dm_type=info[0][12],
emoticon_options=info[0][13],
voice_config=info[0][14],
mode_info=info[0][15],
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],
urank=info[2][5],
mobile_verify=info[2][6],
uname_color=info[2][7],
medal_level=medal_level,
medal_name=medal_name,
runame=runame,
medal_room_id=room_id,
mcolor=mcolor,
special_medal=special_medal,
user_level=info[4][0],
ulevel_color=info[4][2],
ulevel_rank=info[4][3],
old_title=info[5][0],
title=info[5][1],
privilege_type=info[7],
)
@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
try:
return json.loads(self.emoticon_options)
except (json.JSONDecodeError, TypeError):
return {}
@property
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
try:
return json.loads(self.voice_config)
except (json.JSONDecodeError, TypeError):
return {}
@dataclasses.dataclass
class GiftMessage:
"""
礼物消息
"""
gift_name: str = ''
"""礼物名"""
num: int = 0
"""数量"""
uname: str = ''
"""用户名"""
face: str = ''
"""用户头像URL"""
guard_level: int = 0
"""舰队等级，0非舰队，1总督，2提督，3舰长"""
uid: int = 0
"""用户ID"""
timestamp: int = 0
"""时间戳"""
gift_id: int = 0
"""礼物ID"""
gift_type: int = 0
"""礼物类型（未知）"""
action: str = ''
"""目前遇到的有'喂食'、'赠送'"""
price: int = 0
"""礼物单价瓜子数"""
rnd: str = ''
"""随机数，可能是去重用的。有时是时间戳+去重ID，有时是UUID"""
coin_type: str = ''
"""瓜子类型，'silver'或'gold'，1000金瓜子 = 1元"""
total_coin: int = 0
"""总瓜子数"""
tid: str = ''
"""可能是事务ID，有时和rnd相同"""
@classmethod
def from_command(cls, data: dict):
return cls(
gift_name=data['giftName'],
num=data['num'],
uname=data['uname'],
face=data['face'],
guard_level=data['guard_level'],
uid=data['uid'],
timestamp=data['timestamp'],
gift_id=data['giftId'],
gift_type=data['giftType'],
action=data['action'],
price=data['price'],
rnd=data['rnd'],
coin_type=data['coin_type'],
total_coin=data['total_coin'],
tid=data['tid'],
)
@dataclasses.dataclass
class GuardBuyMessage:
"""
上舰消息
"""
uid: int = 0
"""用户ID"""
username: str = ''
"""用户名"""
guard_level: int = 0
"""舰队等级，0非舰队，1总督，2提督，3舰长"""
num: int = 0
"""数量"""
price: int = 0
"""单价金瓜子数"""
gift_id: int = 0
"""礼物ID"""
gift_name: str = ''
"""礼物名"""
start_time: int = 0
"""开始时间戳，和结束时间戳相同"""
end_time: int = 0
"""结束时间戳，和开始时间戳相同"""
@classmethod
def from_command(cls, data: dict):
return cls(
uid=data['uid'],
username=data['username'],
guard_level=data['guard_level'],
num=data['num'],
price=data['price'],
gift_id=data['gift_id'],
gift_name=data['gift_name'],
start_time=data['start_time'],
end_time=data['end_time'],
)
@dataclasses.dataclass
class SuperChatMessage:
"""
醒目留言消息
"""
price: int = 0
"""价格（人民币）"""
message: str = ''
"""消息"""
message_trans: str = ''
"""消息日文翻译（目前只出现在SUPER_CHAT_MESSAGE_JPN）"""
start_time: int = 0
"""开始时间戳"""
end_time: int = 0
"""结束时间戳"""
time: int = 0
"""剩余时间（约等于 结束时间戳 - 开始时间戳）"""
id: int = 0
"""醒目留言ID，删除时用"""
gift_id: int = 0
"""礼物ID"""
gift_name: str = ''
"""礼物名"""
uid: int = 0
"""用户ID"""
uname: str = ''
"""用户名"""
face: str = ''
"""用户头像URL"""
guard_level: int = 0
"""舰队等级，0非舰队，1总督，2提督，3舰长"""
user_level: int = 0
"""用户等级"""
background_bottom_color: str = ''
"""底部背景色，'#rrggbb'"""
background_color: str = ''
"""背景色，'#rrggbb'"""
background_icon: str = ''
"""背景图标"""
background_image: str = ''
"""背景图URL"""
background_price_color: str = ''
"""背景价格颜色，'#rrggbb'"""
@classmethod
def from_command(cls, data: dict):
return cls(
price=data['price'],
message=data['message'],
message_trans=data['message_trans'],
start_time=data['start_time'],
end_time=data['end_time'],
time=data['time'],
id=data['id'],
gift_id=data['gift']['gift_id'],
gift_name=data['gift']['gift_name'],
uid=data['uid'],
uname=data['user_info']['uname'],
face=data['user_info']['face'],
guard_level=data['user_info']['guard_level'],
user_level=data['user_info']['user_level'],
background_bottom_color=data['background_bottom_color'],
background_color=data['background_color'],
background_icon=data['background_icon'],
background_image=data['background_image'],
background_price_color=data['background_price_color'],
)
@dataclasses.dataclass
class SuperChatDeleteMessage:
"""
删除醒目留言消息
"""
ids: List[int] = dataclasses.field(default_factory=list)
"""醒目留言ID数组"""
@classmethod
def from_command(cls, data: dict):
return cls(
ids=data['ids'],
)

39
open_live_sample.py
View File

@ -2,10 +2,15 @@
import asyncio
import blivedm
import blivedm.models.open_live as open_models
import blivedm.models.web as web_models
# 在开放平台申请的开发者密钥
ACCESS_KEY = ''
ACCESS_SECRET = ''
# 在开放平台创建的项目ID
APP_ID = 0
# 主播身份码
ROOM_OWNER_AUTH_CODE = ''
@ -29,17 +34,41 @@ async def run_single_client():
client.start()
try:
# 演示70秒后停止
await asyncio.sleep(70)
client.stop()
# await asyncio.sleep(70)
# client.stop()
await client.join()
finally:
await client.stop_and_close()
class MyHandler(blivedm.HandlerInterface):
async def handle(self, client: blivedm.OpenLiveClient, command: dict):
print(command)
class MyHandler(blivedm.BaseHandler):
async def _on_heartbeat(self, client: blivedm.BLiveClient, message: web_models.HeartbeatMessage):
print(f'[{client.room_id}] 心跳')
async def _on_open_live_danmaku(self, client: blivedm.OpenLiveClient, message: open_models.DanmakuMessage):
print(f'[{message.room_id}] {message.uname}：{message.msg}')
async def _on_open_live_gift(self, client: blivedm.OpenLiveClient, message: open_models.GiftMessage):
coin_type = '金瓜子' if message.paid else '银瓜子'
print(f'[{message.room_id}] {message.uname} 赠送{message.gift_name}x{message.gift_num}'
f' （{coin_type}x{message.price}）')
async def _on_open_live_buy_guard(self, client: blivedm.OpenLiveClient, message: open_models.GuardBuyMessage):
print(f'[{message.room_id}] {message.user_info.uname} 购买 大航海等级={message.guard_level}')
async def _on_open_live_super_chat(
self, client: blivedm.OpenLiveClient, message: open_models.SuperChatMessage
):
print(f'[{message.room_id}] 醒目留言 ¥{message.rmb} {message.uname}：{message.message}')
async def _on_open_live_super_chat_delete(
self, client: blivedm.OpenLiveClient, message: open_models.SuperChatDeleteMessage
):
print(f'[{message.room_id}] 删除醒目留言 message_ids={message.message_ids}')
async def _on_open_live_like(self, client: blivedm.OpenLiveClient, message: open_models.LikeMessage):
print(f'[{message.room_id}] {message.uname} 点赞')
if __name__ == '__main__':

24
sample.py
View File

@ -2,10 +2,12 @@
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 = [
@ -16,14 +18,16 @@ TEST_ROOM_IDS = [
23105590,
]
session = None
session: Optional[aiohttp.ClientSession] = None
async def main():
init_session()
await run_single_client()
await run_multi_clients()
try:
await run_single_client()
await run_multi_clients()
finally:
await session.close()
def init_session():
@ -88,20 +92,20 @@ class MyHandler(blivedm.BaseHandler):
# f" uname={command['data']['uname']}")
# _CMD_CALLBACK_DICT['INTERACT_WORD'] = __interact_word_callback # noqa
async def _on_heartbeat(self, client: blivedm.BLiveClient, message: blivedm.HeartbeatMessage):
print(f'[{client.room_id}] 当前人气值：{message.popularity}')
async def _on_heartbeat(self, client: blivedm.BLiveClient, message: web_models.HeartbeatMessage):
print(f'[{client.room_id}] 心跳')
async def _on_danmaku(self, client: blivedm.BLiveClient, message: blivedm.DanmakuMessage):
async def _on_danmaku(self, client: blivedm.BLiveClient, message: web_models.DanmakuMessage):
print(f'[{client.room_id}] {message.uname}：{message.msg}')
async def _on_gift(self, client: blivedm.BLiveClient, message: blivedm.GiftMessage):
async 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}）')
async def _on_buy_guard(self, client: blivedm.BLiveClient, message: blivedm.GuardBuyMessage):
async def _on_buy_guard(self, client: blivedm.BLiveClient, message: web_models.GuardBuyMessage):
print(f'[{client.room_id}] {message.username} 购买{message.gift_name}')
async def _on_super_chat(self, client: blivedm.BLiveClient, message: blivedm.SuperChatMessage):
async def _on_super_chat(self, client: blivedm.BLiveClient, message: web_models.SuperChatMessage):
print(f'[{client.room_id}] 醒目留言 ¥{message.price} {message.uname}：{message.message}')