模仿UP主，用Python实现一个弹幕控制的直播间！

平凡随和 發表於 2021-12-2 21:01:00

模仿UP主，用Python实现一个弹幕控制的直播间！

<h2 id="灵感来源">灵感来源</h2>
之前在B站看到一个有意思的视频：
【B站】【亦】终极云游戏！五千人同开一辆车，复现经典群体智慧实验
<img src="https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/e1eb9fda15cc400d951eea9b5f05bbea~tplv-k3u1fbpfcp-zoom-1.image">
大家可以看看，很有意思。
up主通过代码实现了实时读取直播间里的弹幕内容，进而控制自己的电脑，把弹幕翻译成指令操控《赛博朋克2077》游戏。
观众也越来越多，最后甚至还把直接间搞崩了（当然，其实是因为那天B站全站崩了）。
我十分好奇到底是怎么做到的。
外行看热闹，内行看门道，作为半个内行，我们就模仿UP主的想法，自己做一个。
所以今天我的目标就是复刻一个 通过弹幕控制直播间 的代码，并且最终在自己的直播间开播。
先给大家看看最终我的成品小视频：
【B站】模仿UP主，做一个弹幕控制的直播间！
看起来是不是很像样了。
<h2 id="初版设计思路">初版设计思路</h2>
首先在脑海里规划一个大致的思路，如下图：
<img src="https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/9ac731afb53b4548a0bcb30c8822af1f~tplv-k3u1fbpfcp-zoom-1.image">
这个思路看起来很简单，不过还是得解释一下，首先我们要搞清楚，弹幕的内容是怎么抓到的。
大部分我们常见的直播平台，在浏览器端，弹幕都是通过WebSocket来推送给观众的。在手机平板等客户端（非Web端），可能会有一些更加复杂的TCP进行弹幕的推送。
关于TCP的消息投递，有个很好的文章，就是美团的这个：美团终端消息投递服务Pike的演进之路
归根结底,这些弹幕都是通过在客户端和服务端建立长链接来实现的。
所以，我们需要做的就是用代码作为客户端，与直播平台进行长链接。这样就能拿到弹幕。
我们只是需要实现整个弹幕控制的流程，所以弹幕的抓取也不是本文的重点，我们来淘一个现成的轮子！在Github上一顿找，找到了一个非常不错的开源库，里面能够获取很多直播平台的弹幕：
https://github.com/wbt5/real-url
<blockquote>
获取斗鱼&虎牙&哔哩哔哩&抖音&快手等 58 个直播平台的真实流媒体地址(直播源)和弹幕，直播源可在 PotPlayer、flv.js 等播放器中播放。
</blockquote>
我们把代码clone下来，运行main函数，随便输入一个Bilibili直播间地址，就能拿到直播间实时的弹幕流：
<img src="https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/e6a9c0e715a54a0fb515eca1382dcaca~tplv-k3u1fbpfcp-zoom-1.image">
代码里把获取到的一条条弹幕（包括用户名）直接打印在了控制台。
他是如何做到的呢？核心的Python代码如下（不熟悉Python？不要紧，就当做伪代码，很容易看懂）：
<pre><code class="language-python">wss_url = 'wss://broadcastlv.chat.bilibili.com/sub'
heartbeat = b'\x00\x00\x00\x1f\x00\x10\x00\x01\x00\x00\x00\x02\x00\x00\x00\x01\x5b\x6f\x62\x6a\x65\x63\x74\x20' \
 b'\x4f\x62\x6a\x65\x63\x74\x5d '
heartbeatInterval = 60

@staticmethod
async def get_ws_info(url):
url = 'https://api.live.bilibili.com/room/v1/Room/room_init?id=' + url.split('/')[-1]
reg_datas = []
async with aiohttp.ClientSession() as session:
 async with session.get(url) as resp:
 room_json = json.loads(await resp.text())
 room_id = room_json['data']['room_id']
 data = json.dumps({
 'roomid': room_id,
 'uid': int(1e14 + 2e14 * random.random()),
 'protover': 1
 }, separators=(',', ':')).encode('ascii')
 data = (pack('>i', len(data) + 16) + b'\x00\x10\x00\x01' +
 pack('>i', 7) + pack('>i', 1) + data)
 reg_datas.append(data)

return Bilibili.wss_url, reg_datas
</code></pre>
它连上了Bilibili的直播弹幕WSS地址，也就是WebSocket地址，然后伪装成客户端，接受弹幕推送。
OK，做完了第一步，下一步就是用消息队列将弹幕发送出来。开启单独的消费者接收弹幕。
为了实现上尽量简单，就不上那些专业的消息队列了，这里用了redis的list作为队列，将弹幕内容放进去。
发送者核心代码如下：
<pre><code class="language-Python"># 链接Redis
def init_redis():
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
return r

# 消息发送者
async def printer(q, redis):
while True:
 m = await q.get()
 if m['msg_type'] == 'danmaku':
 print(f'{m["name"]}：{m["content"]}')
 list_str = list(m["content"])
 print("弹幕拆分:", list_str)
 for char in list_str:
 if char.lower() in key_list:
 print('推送队列：', char.lower())
 redis.rpush(list_name, char.lower())
</code></pre>
完成了弹幕内容的发送后，需要写一个消费者，消费这些弹幕，把里面的指令都提取出来。
并且，在消费者收到弹幕后，如何消费呢？我们需要一个能够用代码指令控制电脑的办法。
咱继续本着不造轮子的原则，找到了一个Python的自动化控制库PyAutoGUI
<blockquote>
PyAutoGUI is a cross-platform GUI automation Python module for human beings. Used to programmatically control the mouse & keyboard.
</blockquote>
安装上这个库，在代码中引入，便可以通过他的API控制电脑鼠标和键盘执行对应的操作。简直是完美啊！
消费者（控制电脑）核心Python代码如下：
<pre><code class="language-python"># 链接Redis
def init_redis():
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
return r

# 消费者
def control(key_name):
print("key_name =", key_name)
if key_name == None:
 print("本次无指令发出")
 return
key_name = key_name.lower()
# 控制电脑指令
if key_name in key_list:
 print("发出指令", key_name)
 pyautogui.keyDown(key_name)
 time.sleep(press_sec)
 pyautogui.keyUp(key_name)
 print("结束指令", key_name)

if __name__ == '__main__':
r = init_redis()
print("开始监听弹幕消息, loop_sec =", loop_sec)
while True:
 key_name = r.lpop(list_name)
 control(key_name)
 time.sleep(loop_sec)
</code></pre>
ok，大功告成，我们打开弹幕发送队列和消费者，这个不断循环消费的队列就开始运行了。一旦弹幕中有wsad这种控制游戏常用的按键，电脑就会自己给自己发出指令。
<img src="https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/06f4c92b3c914ee2b9483480d68767c8~tplv-k3u1fbpfcp-zoom-1.image">
<h2 id="初版运行中的问题">初版运行中的问题</h2>
我兴冲冲的打开自己的B站直播间，开始调试，结果发现我还是太天真了。这个初版代码暴露了非常多的问题。我们一个个来说下是什么问题，我是如何解决的。
<h3 id="指令不人性化">指令不人性化</h3>
水友们其实很喜欢发送类似www dddd这类重复单词（叠词），但初版的实现只支持单个字幕，水友们发现不得劲，没有作用后，就从直播间走了。
这点很容易解决，把弹幕内容拆分成每个单词，然后再推送给队列。
解决方法：拆解弹幕，把DDD，拆成D,D,D，发送个消费者。
<h3 id="危险指令">危险指令</h3>
首先是玩家的指令超出了应该有的范围。
在我把赛博朋克游戏打开，让弹幕观众控制游戏里的开车时，有个神秘观众进了直播间，默默发了个“F”，然后。。。
然后游戏里的V（主角名）就从车里下来了，淦，我是让你们开车的，不是让你们下来和警察斗殴的。。。
解决方法：添加弹幕过滤器。
<pre><code class="language-python"># 将弹幕进行拆分，只发送指定的指令给消费者
key_list = ('w', 's', 'a', 'd', 'j', 'k', 'u', 'i', 'z', 'x', 'f', 'enter', 'shift', 'backspace')
list_str = list(m["content"])
 print("弹幕拆分:", list_str)
 for char in list_str:
 if char.lower() in key_list:
 print('推送队列：', char.lower())
 redis.rpush(list_name, char.lower())
</code></pre>
上面两个问题解决后，发送者就像下面这样运行了：
<img src="https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/666d333f672b4bba92d90ee65e5e1c6a~tplv-k3u1fbpfcp-zoom-1.image">
<h3 id="弹幕指令堆积">弹幕指令堆积</h3>
这是个很大的问题，如果处理所有水友发送的全部弹幕指令，一定会存在消费不过来的问题。
解决方法：需要固定时间处理弹幕，其他抛弃。
<pre><code class="language-python">if __name__ == '__main__':
r = init_redis()
print("开始监听弹幕消息, loop_sec =", loop_sec)
while True:
 key_name = r.lpop(list_name)
 # 每次只取出一个指令，然后把list清空，也就是这个时间窗口内其他弹幕都扔掉！
 r.delete(list_name)
 control(key_name)
 time.sleep(loop_sec)
</code></pre>
<h3 id="弹幕从发出到观众看到结果有延迟">弹幕从发出到观众看到结果有延迟</h3>
在最开始的视频里，你们也能感受到了，从观众的指令发出，到最终被观众看到，大概要经历5秒的延迟。其中，起码有3秒，都是网络直播流的延迟，这一点，很难去优化。
<h2 id="回炉重造后的版本">回炉重造后的版本</h2>
经过一系列调优和涉及，我们的版本也算是从V0.1到了V0.2了。猛虎落泪。
下面是重构后的结构图：
<img src="https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/597e6823075e438e81a0fdb1b59f488d~tplv-k3u1fbpfcp-zoom-1.image">
<h2 id="后记">后记</h2>
在写完这个项目后，我在直播间试了很多次，体验已经无限接近UP主当时的视频了。我开播挂在那边好久，但是，人气最高的时候，也只有20几个人，寥寥十几条弹幕，还有很多是我发的。我还期望着观众能够拉更多人进来一起玩呢，事与愿违啊。
由此可得出结论，我，先得有粉丝，才能玩得起来啊，呜呜呜呜。大家要是不介意，可以关注下我的B站账号，也叫：蛮三刀酱。我会偶尔抽风发点有趣的技术视频的。
本文实现的全部代码已经开源在了Github上，大家可以在自己的直播间里试试呀：
https://github.com/qqxx6661/live_comment_control_stream
我是在阿里搬砖的工程师 @蛮三刀酱
持续的更新优质文章，离不开你的点赞，转发和分享！
全网唯一技术公众号：后端技术漫谈 
来源：https://www.cnblogs.com/rude3knife/p/15635306.html

頁: [1]

琼殿技术论坛's Archiver

模仿UP主，用Python实现一个弹幕控制的直播间！