纯净、安全、绿色的下载网站

首页|软件分类|下载排行|最新软件|IT学院

当前位置:首页IT学院IT技术

python弹幕直播间 用Python实现一个模仿UP主弹幕控制的直播间功能

蛮三刀酱   2021-12-03 我要评论
想了解用Python实现一个模仿UP主弹幕控制的直播间功能的相关内容吗蛮三刀酱在本文为您仔细讲解python弹幕直播间的相关知识和一些Code实例欢迎阅读和指正我们先划重点:python弹幕直播间,python直播间,pythonUP主弹幕下面大家一起来学习吧。

灵感来源

之前在B站看到一个有意思的视频:

【B站】【亦】终极云游戏!五千人同开一辆车复现经典群体智慧实验

大家可以看看很有意思。

up主通过代码实现了实时读取直播间里的弹幕内容进而控制自己的电脑把弹幕翻译成指令操控《赛博朋克2077》游戏。

观众也越来越多最后甚至还把直接间搞崩了(当然其实是因为那天B站全站崩了)。

我十分好奇到底是怎么做到的。

外行看热闹内行看门道作为半个内行我们就模仿UP主的想法自己做一个。

所以今天我的目标就是复刻一个 通过弹幕控制直播间 的代码并且最终在自己的直播间开播。

先给大家看看最终我的成品小视频:

【B站】模仿UP主做一个弹幕控制的直播间!

看起来是不是很像样了。

初版设计思路

首先在脑海里规划一个大致的思路如下图:

这个思路看起来很简单不过还是得解释一下首先我们要搞清楚弹幕的内容是怎么抓到的。

大部分我们常见的直播平台在浏览器端弹幕都是通过WebSocket来推送给观众的。在手机平板等客户端(非Web端)可能会有一些更加复杂的TCP进行弹幕的推送。

关于TCP的消息投递有个很好的文章就是美团的这个:美团终端消息投递服务Pike的演进之路

归根结底,这些弹幕都是通过在客户端和服务端建立长链接来实现的。

所以我们需要做的就是用代码作为客户端与直播平台进行长链接。这样就能拿到弹幕。

我们只是需要实现整个弹幕控制的流程所以弹幕的抓取也不是本文的重点我们来淘一个现成的轮子!在Github上一顿找找到了一个非常不错的开源库里面能够获取很多直播平台的弹幕:

https://github.com/wbt5/real-url

获取斗鱼&虎牙&哔哩哔哩&抖音&快手等 58 个直播平台的真实流媒体地址(直播源)和弹幕直播源可在 PotPlayer、flv.js 等播放器中播放。

我们把代码clone下来运行main函数随便输入一个Bilibili直播间地址就能拿到直播间实时的弹幕流:

代码里把获取到的一条条弹幕(包括用户名)直接打印在了控制台。

他是如何做到的呢?核心的Python代码如下(不熟悉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

它连上了Bilibili的直播弹幕WSS地址也就是WebSocket地址然后伪装成客户端接受弹幕推送。

OK做完了第一步下一步就是用消息队列将弹幕发送出来。开启单独的消费者接收弹幕。

为了实现上尽量简单就不上那些专业的消息队列了这里用了redis的list作为队列将弹幕内容放进去。

发送者核心代码如下:

# 链接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())

完成了弹幕内容的发送后需要写一个消费者消费这些弹幕把里面的指令都提取出来。

并且在消费者收到弹幕后如何消费呢?我们需要一个能够用代码指令控制电脑的办法。

咱继续本着不造轮子的原则找到了一个Python的自动化控制库PyAutoGUI

PyAutoGUI is a cross-platform GUI automation Python module for human beings. Used to programmatically control the mouse & keyboard.

安装上这个库在代码中引入便可以通过他的API控制电脑鼠标和键盘执行对应的操作。简直是完美啊!

消费者(控制电脑)核心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)

ok大功告成我们打开弹幕发送队列和消费者这个不断循环消费的队列就开始运行了。一旦弹幕中有wsad这种控制游戏常用的按键电脑就会自己给自己发出指令。

初版运行中的问题

我兴冲冲的打开自己的B站直播间开始调试结果发现我还是太天真了。这个初版代码暴露了非常多的问题。我们一个个来说下是什么问题我是如何解决的。

指令不人性化

水友们其实很喜欢发送类似www dddd这类重复单词(叠词)但初版的实现只支持单个字幕水友们发现不得劲没有作用后就从直播间走了。

这点很容易解决把弹幕内容拆分成每个单词然后再推送给队列。

解决方法:拆解弹幕把DDD拆成D,D,D发送个消费者。

危险指令

首先是玩家的指令超出了应该有的范围。

在我把赛博朋克游戏打开让弹幕观众控制游戏里的开车时有个神秘观众进了直播间默默发了个“F”然后。。。

然后游戏里的V(主角名)就从车里下来了淦我是让你们开车的不是让你们下来和警察斗殴的。。。

解决方法:添加弹幕过滤器。

# 将弹幕进行拆分只发送指定的指令给消费者
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())

上面两个问题解决后发送者就像下面这样运行了:

弹幕指令堆积

这是个很大的问题如果处理所有水友发送的全部弹幕指令一定会存在消费不过来的问题。

解决方法:需要固定时间处理弹幕其他抛弃。

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)

弹幕从发出到观众看到结果有延迟

在最开始的视频里你们也能感受到了从观众的指令发出到最终被观众看到大概要经历5秒的延迟。其中起码有3秒都是网络直播流的延迟这一点很难去优化。

回炉重造后的版本

经过一系列调优和涉及我们的版本也算是从V0.1到了V0.2了。猛虎落泪。

下面是重构后的结构图:

后记

在写完这个项目后我在直播间试了很多次体验已经无限接近UP主当时的视频了。我开播挂在那边好久但是人气最高的时候也只有20几个人寥寥十几条弹幕还有很多是我发的。我还期望着观众能够拉更多人进来一起玩呢事与愿违啊。

由此可得出结论我先得有粉丝才能玩得起来啊呜呜呜呜。大家要是不介意可以关注下我的B站账号也叫:蛮三刀酱。我会偶尔抽风发点有趣的技术视频的。

本文实现的全部代码已经开源在了Github上大家可以在自己的直播间里试试呀:

https://github.com/qqxx6661/live_comment_control_stream


相关文章

猜您喜欢

  • pygame实现方块动画 pygame实现方块动画实例讲解

    想了解pygame实现方块动画实例讲解的相关内容吗我的天才女友在本文为您仔细讲解pygame实现方块动画的相关知识和一些Code实例欢迎阅读和指正我们先划重点:pygame方块动画下面大家一起来学习吧。..
  • javascript函数进阶 javascript之函数进阶详解

    想了解javascript之函数进阶详解的相关内容吗维生素D没有用在本文为您仔细讲解javascript函数进阶的相关知识和一些Code实例欢迎阅读和指正我们先划重点:javascript函数,javascript函数进阶下面大家一起来学习吧。..

网友评论

Copyright 2020 www.ducttapegames.com 【环球游戏网】 版权所有 软件发布

声明:所有软件和文章来自软件开发商或者作者 如有异议 请与本站联系 点此查看联系方式