From d8e6cfcbced1368ee2509d977afc50c9f5f3fcbc Mon Sep 17 00:00:00 2001
From: Cam <cam@Cams-MBP.lan>
Date: Thu, 6 Jan 2022 12:18:29 +0800
Subject: [PATCH] add new msg operator class and fix some bug
---
app.py | 42 ++++++++++++++---
blive/core.py | 21 +++++----
blive/framework.py | 25 +++++-----
blive/msg.py | 113 +++++++++++++++++++++++++++++++++++++++++++--
4 files changed, 171 insertions(+), 30 deletions(-)
diff --git a/app.py b/app.py
index 5e4888a..1f2b5ca 100644
--- a/app.py
+++ b/app.py
@@ -1,14 +1,44 @@
-from blive import BLiver, Events, BLiverCtx, DanMuMsg
+from blive import BLiver, Events, BLiverCtx
+from blive.msg import DanMuMsg, HotRankChangeV2Msg, InteractWordMsg, SendGiftMsg
-app = BLiver(510)
+app = BLiver(605)
-@app.handler(Events.DANMU_MSG)
+@app.on(Events.DANMU_MSG)
async def listen(ctx: BLiverCtx):
danmu = DanMuMsg(ctx.body)
- print(danmu.content())
- print(danmu.sender())
- print(danmu.timestamp())
+ print(
+ f"\n{danmu.sender['name']}({danmu.sender['medal']['medal_name']}:{danmu.sender['medal']['medal_level']}): \"{danmu.content}\"\n "
+ )
+
+
+@app.on(Events.INTERACT_WORD)
+async def listen_join(ctx: BLiverCtx):
+ join = InteractWordMsg(ctx.body)
+ print(
+ "欢迎",
+ f"{join.user['name']} ({join.user['medal']['medal_name']}:{join.user['medal']['medal_level']})",
+ "进入直播间",
+ )
+
+
+@app.on(Events.SUPER_CHAT_MESSAGE)
+async def listen_sc(ctx: BLiverCtx):
+ print(ctx.body)
+
+
+@app.on(Events.SEND_GIFT)
+async def listen_gift(ctx: BLiverCtx):
+ msg = SendGiftMsg(ctx.body)
+ print(f"{msg.sender['name']} 送出 {msg.gift['gift_name']}")
+
+
+@app.on(Events.HOT_RANK_CHANGED_V2)
+async def hot(ctx: BLiverCtx):
+ msg = HotRankChangeV2Msg(ctx.body)
+ print(
+ f"恭喜 {ctx.bliver.uname} 在 {msg.area_name} 区 的 {msg.rank_desc} 榜单中获得第 {msg.rank} 名"
+ )
app.run()
diff --git a/blive/core.py b/blive/core.py
index bea73c3..b393258 100644
--- a/blive/core.py
+++ b/blive/core.py
@@ -21,7 +21,7 @@ def get_blive_ws_url(roomid, ssl=False):
return url, data["data"]["token"]
-def get_blive_room_id(roomid):
+def get_blive_room_info(roomid):
"""
得到b站直播间id,（短id不是真实的id）
@@ -34,8 +34,7 @@ def get_blive_room_id(roomid):
data = resp.json()
return (
data["data"]["room_info"]["room_id"],
- data["data"]["room_info"]["short_id"],
- data["data"]["room_info"]["uid"],
+ data["data"]["anchor_info"]["base_info"]["uname"],
)
@@ -81,7 +80,7 @@ class AuthReplyCode(enum.IntEnum):
class ProtocolVersion(enum.IntEnum):
NORMAL = 0 # 未压缩
HEARTBEAT = 1 # 心跳
- INFLATE = 2 # zlib压缩
+ DEFLATE = 2 # zlib压缩
BROTLI = 3 # brotil 压缩
@@ -152,7 +151,6 @@ class B_MsgPackage:
if header.operation == Operation.HEARTBEAT_REPLY:
# 心跳不会粘包
packages.append((header, data[4:].decode("utf-8")))
-
# 通知包处理
elif header.operation == Operation.NOTIFY:
@@ -182,7 +180,7 @@ class B_MsgPackage:
packages.append((header, data[16:].decode("utf-8")))
# NOTIFY 消息可能会粘包
- if header.version == ProtocolVersion.INFLATE:
+ if header.version == ProtocolVersion.DEFLATE:
# 先zlib解码
data = zlib.decompress(data)
notify_pk_process(data)
@@ -213,7 +211,7 @@ class Events(str, enum.Enum):
ROOM_CHANGE = "ROOM_CHANGE" # 房间信息改变
ROOM_RANK = "ROOM_RANK" # 排名改变
DANMU_MSG = "DANMU_MSG" # 接收到弹幕【自动回复】
- SEND_GIFT = "ROOM_RANK" # 有人送礼【答谢送礼】
+ SEND_GIFT = "SEND_GIFT" # 有人送礼【答谢送礼】
WELCOME_GUARD = "WELCOME_GUARD" # 舰长进入（不会触发）
ENTRY_EFFECT = "ENTRY_EFFECT" # 舰长、高能榜、老爷进入【欢迎舰长】
WELCOME = "WELCOME" # 老爷进入
@@ -222,7 +220,7 @@ class Events(str, enum.Enum):
SHARE = "SHARE" # 用户分享直播间
SPECIAL_ATTENTION = "SPECIAL_ATTENTION" # 特别关注直播间，可用%special%判断
ROOM_REAL_TIME_MESSAGE_UPDATE = "ROOM_REAL_TIME_MESSAGE_UPDATE" # 粉丝数量改变
- SUPER_CHAT_MESSAGE = "ROOM_REAL_TIME_MESSAGE_UPDATE" # 醒目留言
+ SUPER_CHAT_MESSAGE = "SUPER_CHAT_MESSAGE" # 醒目留言
SUPER_CHAT_MESSAGE_JPN = "SUPER_CHAT_MESSAGE_JPN" # 醒目留言日文翻译
SUPER_CHAT_MESSAGE_DELETE = "SUPER_CHAT_MESSAGE_DELETE" # 删除醒目留言
ROOM_BLOCK_MSG = "ROOM_BLOCK_MSG" # 用户被禁言，%uname%昵称
@@ -254,3 +252,10 @@ class Events(str, enum.Enum):
# 勋章升级，仅送礼物后触发，需设置中开启“监听勋章升级”。%medal_level%获取新等级（但用户当前勋章不一定是本直播间）
MEDAL_UPGRADE = "MEDAL_UPGRADE"
STOP_LIVE_ROOM_LIST = "STOP_LIVE_ROOM_LIST" # 停止直播的房间
+ WIDGET_BANNER = "WIDGET_BANNER" # 小部件横幅
+ PK_BATTLE_PROCESS_NEW = "PK_BATTLE_PROCESS_NEW" # 开始pk
+ PK_BATTLE_PROCESS = "PK_BATTLE_PROCESS" # pk
+ COMMON_NOTICE_DANMAKU = "COMMON_NOTICE_DANMAKU" # 弹幕通知
+ HOT_RANK_CHANGED_V2 = "HOT_RANK_CHANGED_V2" # 热门榜改变v2
+ PK_BATTLE_SETTLE = "PK_BATTLE_SETTLE" # pk结果
+ PK_BATTLE_PRE_NEW = "PK_BATTLE_PRE_NEW" # pk预创建
diff --git a/blive/framework.py b/blive/framework.py
index e1c3d3c..42ea4a2 100644
--- a/blive/framework.py
+++ b/blive/framework.py
@@ -1,7 +1,7 @@
import sys
import json
import asyncio
-from typing import Dict, Tuple
+from typing import Dict, List, Tuple, Union
import aiohttp
from aiohttp.client_ws import ClientWebSocketResponse
from aiohttp.http_websocket import WSMessage
@@ -13,7 +13,7 @@ from .core import (
Operation,
PackageHeader,
packman,
- get_blive_room_id,
+ get_blive_room_info,
get_blive_ws_url,
certification,
heartbeat,
@@ -55,20 +55,25 @@ class Processor:
class BLiver:
def __init__(self, roomid, logger=None, log_level="INFO"):
self.roomid = roomid
+ self.real_roomid, self.uname = get_blive_room_info(roomid)
if not logger:
self.logger = loguru.logger
self.logger.remove()
+ self.logger.add(sys.stderr, level=log_level)
else:
self.logger = logger
- self.logger.add(sys.stderr, level=log_level)
self._ws: ClientWebSocketResponse = None
self.scheduler = AsyncIOScheduler(timezone="Asia/ShangHai")
self.processor = Processor(logger=self.logger)
- def handler(self, event: Events):
+ def on(self, event: Union[Events, List[Events]]):
def f_wrapper(func):
- self.logger.debug("handler added")
- self.processor.register(event, func)
+ self.logger.debug("handler added,{}", func)
+ if isinstance(event, list):
+ for e in event:
+ self.processor.register(e, func)
+ else:
+ self.processor.register(event, func)
return func
return f_wrapper
@@ -121,22 +126,20 @@ class BLiver:
self.logger.debug("heartbeat sended")
async def listen(self):
- rommid, _, _ = get_blive_room_id(self.roomid) # 如果是短id,就得到直播间真实id
- url, token = get_blive_ws_url(rommid)
+
+ url, token = get_blive_ws_url(self.real_roomid)
async with aiohttp.ClientSession().ws_connect(url) as ws:
self._ws = ws
await ws.send_bytes(
- packman.pack(certification(rommid, token), Operation.AUTH)
+ packman.pack(certification(self.real_roomid, token), Operation.AUTH)
)
self.scheduler.add_job(self.heartbeat, trigger="interval", seconds=30)
self.scheduler.start()
# 开始监听
while True:
msg: WSMessage = await ws.receive()
- # print(msg)
if msg.type != aiohttp.WSMsgType.BINARY:
continue
- # print(msg.data)
mq = packman.unpack(msg.data)
self.logger.debug("received msg:\n{}", mq)
tasks = [self.processor.process(BLiverCtx(self, m)) for m in mq]
diff --git a/blive/msg.py b/blive/msg.py
index 6aba14d..76393b2 100644
--- a/blive/msg.py
+++ b/blive/msg.py
@@ -11,12 +11,10 @@ class BaseMsg(ABC):
super().__init__()
self.body = body
+ @property
def cmd(self):
return self.body["cmd"]
- def info(self):
- return self.body["info"]
-
def __repr__(self) -> str:
return json.dumps(self.body)
@@ -25,11 +23,116 @@ class DanMuMsg(BaseMsg):
def __init__(self, body) -> None:
super(DanMuMsg, self).__init__(body)
+ @property
def content(self):
- return self.info()[1]
+ return self.body["info"][1]
+ @property
def sender(self):
- return {"id": self.body["info"][2][0], "name": self.body["info"][2][1]}
+ return {
+ "id": self.body["info"][2][0],
+ "name": self.body["info"][2][1],
+ "medal": {
+ "medal_name": self.body["info"][3][1] if self.body["info"][3] else "",
+ "medal_level": self.body["info"][3][0] if self.body["info"][3] else 0,
+ },
+ }
+ @property
def timestamp(self):
return self.body["info"][9]
+
+
+class InteractWordMsg(BaseMsg):
+ def __init__(self, body) -> None:
+ super().__init__(body)
+
+ @property
+ def user(self):
+ return {
+ "id": self.body["data"]["uid"],
+ "name": self.body["data"]["uname"],
+ "medal": {
+ "medal_name": self.body["data"]["fans_medal"]["medal_name"],
+ "medal_level": self.body["data"]["fans_medal"]["medal_level"],
+ },
+ }
+
+ @property
+ def timestamp(self):
+ return self.body["data"]["timestamp"]
+
+
+class StopLiveRoomListMsg(BaseMsg):
+ def __init__(self, body) -> None:
+ super().__init__(body)
+
+ @property
+ def room_id_list(self):
+ return self.body["data"]["room_id_list"]
+
+
+class HotRankChangeV2Msg(BaseMsg):
+ def __init__(self, body) -> None:
+ super().__init__(body)
+
+ @property
+ def area_name(self):
+ return self.body["data"]["area_name"]
+
+ @property
+ def rank_desc(self):
+ return self.body["data"]["rank_desc"]
+
+ @property
+ def rank(self):
+ return self.body["data"]["rank"]
+
+ @property
+ def trend(self):
+ return self.body["data"]["trend"]
+
+ @property
+ def timestamp(self):
+ return self.body["data"]["timestamp"]
+
+
+class SendGiftMsg(BaseMsg):
+ # TODO 礼物逻辑复杂, 考虑更复杂的封装类
+ def __init__(self, body) -> None:
+
+ super().__init__(body)
+
+ @property
+ def sender(self):
+ return {
+ "id": self.body["data"]["uid"],
+ "name": self.body["data"]["uname"],
+ "medal": {
+ "medal_name": self.body["data"]["medal_info"]["medal_name"],
+ "medal_level": self.body["data"]["medal_info"]["medal_level"],
+ },
+ }
+
+ @property
+ def action(self):
+ return self.body["data"]["action"]
+
+ @property
+ def gift(self):
+ return {
+ "gift_id": self.body["data"]["giftId"],
+ "gift_name": self.body["data"]["giftName"],
+ "gift_type": self.body["data"]["giftType"],
+ }
+
+ @property
+ def combo(self):
+ return {
+ "batch_combo_id": self.body["data"]["batch_combo_id"],
+ "batch_combo_send": self.body["data"]["batch_combo_send"],
+ "combo_resources_id": self.body["data"]["combo_resources_id"],
+ "combo_send": self.body["data"]["combo_send"],
+ "combo_stay_time": self.body["data"]["combo_stay_time"],
+ "combo_total_coin": self.body["data"]["combo_total_coin"],
+ }