You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
blivechat/blcsdk/models.py

420 lines
10 KiB
Python

12 months ago
# -*- coding: utf-8 -*-
import dataclasses
12 months ago
import enum
from typing import *
__all__ = (
'RoomKeyType',
'RoomKey',
'Command',
'ExtraData',
'AddRoomMsg',
'RoomInitMsg',
'DelRoomMsg',
'OpenPluginAdminUiMsg',
'AuthorType',
'GuardLevel',
'ContentType',
'AddTextMsg',
'AddGiftMsg',
'AddMemberMsg',
'AddSuperChatMsg',
'DelSuperChatMsg',
'UpdateTranslationMsg',
)
class RoomKeyType(enum.IntEnum):
ROOM_ID = 1
AUTH_CODE = 2
class RoomKey(NamedTuple):
"""用来标识一个房间"""
type: RoomKeyType
value: Union[int, str]
def __str__(self):
res = str(self.value)
if self.type == RoomKeyType.AUTH_CODE:
# 身份码要脱敏
res = '***' + res[-3:]
return res
__repr__ = __str__
12 months ago
@classmethod
def from_dict(cls, data: dict):
type_ = RoomKeyType(data['type'])
value = data['value']
if type_ == RoomKeyType.ROOM_ID:
if not isinstance(value, int):
raise TypeError(f'Type of value is {type(value)}, value={value}')
elif type_ == RoomKeyType.AUTH_CODE:
if not isinstance(value, str):
raise TypeError(f'Type of value is {type(value)}, value={value}')
return cls(type=type_, value=value)
def to_dict(self):
return {'type': self.type, 'value': self.value}
12 months ago
class Command(enum.IntEnum):
HEARTBEAT = 0
BLC_INIT = 1
ADD_ROOM = 2
ROOM_INIT = 3
DEL_ROOM = 4
OPEN_PLUGIN_ADMIN_UI = 5
# 从插件发送到blivechat的请求
LOG_REQ = 30
ADD_TEXT_REQ = 31
# 房间内消息
ADD_TEXT = 50
ADD_GIFT = 51
ADD_MEMBER = 52
ADD_SUPER_CHAT = 53
DEL_SUPER_CHAT = 54
UPDATE_TRANSLATION = 55
@dataclasses.dataclass
class ExtraData:
"""一些消息共用的附加信息"""
room_id: Optional[int] = None
"""房间ID"""
room_key: Optional[RoomKey] = None
"""blivechat用来标识一个房间的key"""
is_from_plugin: bool = False
"""
消息是插件生成的
如果你的插件既要监听消息又要生成消息注意判断这个以避免死循环
"""
@classmethod
def from_dict(cls, data: dict):
room_key_dict = data.get('roomKey', None)
if room_key_dict is not None:
room_key = RoomKey.from_dict(room_key_dict)
else:
room_key = None
return cls(
room_id=data.get('roomId', None),
room_key=room_key,
is_from_plugin=data.get('isFromPlugin', False),
)
@dataclasses.dataclass
class _EmptyMsg:
@classmethod
def from_command(cls, _data: dict):
return cls()
@dataclasses.dataclass
class AddRoomMsg(_EmptyMsg):
"""
添加房间消息房间信息在extra里
此时room_id是None因为还没有初始化
"""
@dataclasses.dataclass
class RoomInitMsg:
"""
房间初始化消息房间信息在extra里
一个房间创建后可能被多次初始化处理时注意去重
"""
is_success: bool = False
@classmethod
def from_command(cls, data: dict):
return cls(
is_success=data['isSuccess'],
)
@dataclasses.dataclass
class DelRoomMsg(_EmptyMsg):
"""
删除房间消息房间信息在extra里
注意此时room_id可能是None
"""
@dataclasses.dataclass
class OpenPluginAdminUiMsg(_EmptyMsg):
"""用户请求打开当前插件的管理界面消息"""
class AuthorType(enum.IntEnum):
NORMAL = 0
GUARD = 1
"""舰队"""
ADMIN = 2
"""房管"""
ROOM_OWNER = 3
"""主播"""
class GuardLevel(enum.IntEnum):
"""舰队等级"""
NONE = 0
LV3 = 1
"""总督"""
LV2 = 2
"""提督"""
LV1 = 3
"""舰长"""
class ContentType(enum.IntEnum):
TEXT = 0
EMOTICON = 1
@dataclasses.dataclass
class AddTextMsg:
"""弹幕消息"""
avatar_url: str = ''
"""用户头像URL"""
timestamp: int = 0
"""时间戳(秒)"""
author_name: str = ''
"""用户名"""
author_type: int = AuthorType.NORMAL.value
"""用户类型见AuthorType"""
content: str = ''
"""弹幕内容"""
privilege_type: int = GuardLevel.NONE.value
"""舰队等级见GuardLevel"""
is_gift_danmaku: bool = False
"""是否礼物弹幕"""
author_level: int = 1
"""用户等级"""
is_newbie: bool = False
"""是否正式会员"""
is_mobile_verified: bool = True
"""是否绑定手机"""
medal_level: int = 0
"""勋章等级如果没戴当前房间勋章则为0"""
id: str = ''
"""消息ID"""
translation: str = ''
"""弹幕内容翻译"""
content_type: int = ContentType.TEXT.value
"""内容类型见ContentType"""
content_type_params: Union[dict, list] = dataclasses.field(default_factory=dict)
"""跟内容类型相关的参数"""
uid: int = 0
"""用户ID"""
medal_name: str = ''
"""勋章名"""
@classmethod
def from_command(cls, data: list):
content_type = data[13]
content_type_params = data[14]
if content_type == ContentType.EMOTICON:
content_type_params = {'url': content_type_params[0]}
return cls(
avatar_url=data[0],
timestamp=data[1],
author_name=data[2],
author_type=data[3],
content=data[4],
privilege_type=data[5],
is_gift_danmaku=bool(data[6]),
author_level=data[7],
is_newbie=bool(data[8]),
is_mobile_verified=bool(data[9]),
medal_level=data[10],
id=data[11],
translation=data[12],
content_type=content_type,
content_type_params=content_type_params,
uid=data[16],
medal_name=data[17],
)
@dataclasses.dataclass
class AddGiftMsg:
"""礼物消息"""
id: str = ''
"""消息ID"""
avatar_url: str = ''
"""用户头像URL"""
timestamp: int = 0
"""时间戳(秒)"""
author_name: str = ''
"""用户名"""
total_coin: int = 0
"""总价付费瓜子数1000金瓜子 = 1元"""
total_free_coin: int = 0
"""总价免费瓜子数"""
gift_name: str = ''
"""礼物名"""
num: int = 0
"""数量"""
gift_id: int = 0
"""礼物ID"""
gift_icon_url: str = ''
"""礼物图标URL"""
uid: int = 0
"""用户ID"""
privilege_type: int = GuardLevel.NONE.value
"""舰队等级见GuardLevel"""
medal_level: int = 0
"""勋章等级如果没戴当前房间勋章则为0"""
medal_name: str = ''
"""勋章名"""
@classmethod
def from_command(cls, data: dict):
return cls(
id=data['id'],
avatar_url=data['avatarUrl'],
timestamp=data['timestamp'],
author_name=data['authorName'],
total_coin=data['totalCoin'],
total_free_coin=data['totalFreeCoin'],
gift_name=data['giftName'],
num=data['num'],
gift_id=data['giftId'],
gift_icon_url=data['giftIconUrl'],
uid=data['uid'],
privilege_type=data['privilegeType'],
medal_level=data['medalLevel'],
medal_name=data['medalName'],
)
@dataclasses.dataclass
class AddMemberMsg:
"""上舰消息"""
id: str = ''
"""消息ID"""
avatar_url: str = ''
"""用户头像URL"""
timestamp: int = 0
"""时间戳(秒)"""
author_name: str = ''
"""用户名"""
privilege_type: int = GuardLevel.NONE.value
"""舰队等级见GuardLevel"""
num: int = 0
"""数量"""
unit: str = ''
"""单位(月)"""
uid: int = 0
"""用户ID"""
medal_level: int = 0
"""勋章等级如果没戴当前房间勋章则为0"""
medal_name: str = ''
"""勋章名"""
@classmethod
def from_command(cls, data: dict):
return cls(
id=data['id'],
avatar_url=data['avatarUrl'],
timestamp=data['timestamp'],
author_name=data['authorName'],
privilege_type=data['privilegeType'],
num=data['num'],
unit=data['unit'],
uid=data['uid'],
medal_level=data['medalLevel'],
medal_name=data['medalName'],
)
@dataclasses.dataclass
class AddSuperChatMsg:
"""醒目留言消息"""
id: str = ''
"""消息ID"""
avatar_url: str = ''
"""用户头像URL"""
timestamp: int = 0
"""时间戳(秒)"""
author_name: str = ''
"""用户名"""
price: int = 0
"""价格(元)"""
content: str = ''
"""内容"""
translation: str = ''
"""内容翻译"""
uid: int = 0
"""用户ID"""
privilege_type: int = GuardLevel.NONE.value
"""舰队等级见GuardLevel"""
medal_level: int = 0
"""勋章等级如果没戴当前房间勋章则为0"""
medal_name: str = ''
"""勋章名"""
@classmethod
def from_command(cls, data: dict):
return cls(
id=data['id'],
avatar_url=data['avatarUrl'],
timestamp=data['timestamp'],
author_name=data['authorName'],
price=data['price'],
content=data['content'],
translation=data['translation'],
uid=data['uid'],
privilege_type=data['privilegeType'],
medal_level=data['medalLevel'],
medal_name=data['medalName'],
)
@dataclasses.dataclass
class DelSuperChatMsg:
"""删除醒目留言消息"""
ids: List[str] = dataclasses.field(default_factory=list)
"""醒目留言ID数组"""
@classmethod
def from_command(cls, data: dict):
return cls(
ids=data['ids'],
)
@dataclasses.dataclass
class UpdateTranslationMsg:
"""更新内容翻译消息"""
id: str = ''
"""消息ID"""
translation: str = ''
"""内容翻译"""
@classmethod
def from_command(cls, data: list):
return cls(
id=data[0],
translation=data[1],
)