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，做完了第一步，下一步就是用消息隊(duì)列將彈幕發(fā)送出來。開啟單獨(dú)的消費(fèi)者接收彈幕。

為了實(shí)現(xiàn)上盡量簡單，就不上那些專業(yè)的消息隊(duì)列了，這里用了redis的list作為隊(duì)列，將彈幕內(nèi)容放進(jìn)去。

發(fā)送者核心代碼如下：

# 鏈接Redis
def init_redis():
    r = redis.Redis(host='localhost', port=6379, decode_responses=True)
    return r

# 消息發(fā)送者
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('推送隊(duì)列：', char.lower())
                    redis.rpush(list_name, char.lower())

完成了彈幕內(nèi)容的發(fā)送后，需要寫一個(gè)消費(fèi)者，消費(fèi)這些彈幕，把里面的指令都提取出來。

并且，在消費(fèi)者收到彈幕后，如何消費(fèi)呢？我們需要一個(gè)能夠用代碼指令控制電腦的辦法。

咱繼續(xù)本著不造輪子的原則，找到了一個(gè)Python的自動(dòng)化控制庫PyAutoGUI

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

安裝上這個(gè)庫，在代碼中引入，便可以通過他的API控制電腦鼠標(biāo)和鍵盤執(zhí)行對應(yīng)的操作。簡直是完美?。?/p>

消費(fèi)者（控制電腦）核心Python代碼如下：

# 鏈接Redis
def init_redis():
    r = redis.Redis(host='localhost', port=6379, decode_responses=True)
    return r

# 消費(fèi)者
def control(key_name):
    print("key_name =", key_name)
    if key_name == None:
        print("本次無指令發(fā)出")
        return
    key_name = key_name.lower()
    # 控制電腦指令
    if key_name in key_list:
        print("發(fā)出指令", key_name)
        pyautogui.keyDown(key_name)
        time.sleep(press_sec)
        pyautogui.keyUp(key_name)
        print("結(jié)束指令", key_name)

if __name__ == '__main__':
    r = init_redis()
    print("開始監(jiān)聽彈幕消息, loop_sec =", loop_sec)
    while True:
        key_name = r.lpop(list_name)
        control(key_name)
        time.sleep(loop_sec)

ok，大功告成，我們打開彈幕發(fā)送隊(duì)列和消費(fèi)者，這個(gè)不斷循環(huán)消費(fèi)的隊(duì)列就開始運(yùn)行了。一旦彈幕中有wsad這種控制游戲常用的按鍵，電腦就會(huì)自己給自己發(fā)出指令。

初版運(yùn)行中的問題

我興沖沖的打開自己的B站直播間，開始調(diào)試，結(jié)果發(fā)現(xiàn)我還是太天真了。這個(gè)初版代碼暴露了非常多的問題。我們一個(gè)個(gè)來說下是什么問題，我是如何解決的。

指令不人性化

水友們其實(shí)很喜歡發(fā)送類似www dddd這類重復(fù)單詞（疊詞），但初版的實(shí)現(xiàn)只支持單個(gè)字幕，水友們發(fā)現(xiàn)不得勁，沒有作用后，就從直播間走了。

這點(diǎn)很容易解決，把彈幕內(nèi)容拆分成每個(gè)單詞，然后再推送給隊(duì)列。

解決方法：拆解彈幕，把DDD，拆成D,D,D，發(fā)送個(gè)消費(fèi)者。

危險(xiǎn)指令

首先是玩家的指令超出了應(yīng)該有的范圍。

在我把賽博朋克游戲打開，讓彈幕觀眾控制游戲里的開車時(shí)，有個(gè)神秘觀眾進(jìn)了直播間，默默發(fā)了個(gè)“F”，然后。。。

然后游戲里的V（主角名）就從車?yán)锵聛砹耍?，我是讓你們開車的，不是讓你們下來和警察斗毆的。。。

解決方法：添加彈幕過濾器。

# 將彈幕進(jìn)行拆分，只發(fā)送指定的指令給消費(fèi)者
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('推送隊(duì)列：', char.lower())
                    redis.rpush(list_name, char.lower())

上面兩個(gè)問題解決后，發(fā)送者就像下面這樣運(yùn)行了：

彈幕指令堆積

這是個(gè)很大的問題，如果處理所有水友發(fā)送的全部彈幕指令，一定會(huì)存在消費(fèi)不過來的問題。

解決方法：需要固定時(shí)間處理彈幕，其他拋棄。

if __name__ == '__main__':
    r = init_redis()
    print("開始監(jiān)聽彈幕消息, loop_sec =", loop_sec)
    while True:
        key_name = r.lpop(list_name)
        # 每次只取出一個(gè)指令，然后把list清空，也就是這個(gè)時(shí)間窗口內(nèi)其他彈幕都扔掉！
        r.delete(list_name)
        control(key_name)
        time.sleep(loop_sec)

彈幕從發(fā)出到觀眾看到結(jié)果有延遲

在最開始的視頻里，你們也能感受到了，從觀眾的指令發(fā)出，到最終被觀眾看到，大概要經(jīng)歷5秒的延遲。其中，起碼有3秒，都是網(wǎng)絡(luò)直播流的延遲，這一點(diǎn)，很難去優(yōu)化。

回爐重造后的版本

經(jīng)過一系列調(diào)優(yōu)和涉及，我們的版本也算是從V0.1到了V0.2了。猛虎落淚。

下面是重構(gòu)后的結(jié)構(gòu)圖：

后記

在寫完這個(gè)項(xiàng)目后，我在直播間試了很多次，體驗(yàn)已經(jīng)無限接近UP主當(dāng)時(shí)的視頻了。我開播掛在那邊好久，但是，人氣最高的時(shí)候，也只有20幾個(gè)人，寥寥十幾條彈幕，還有很多是我發(fā)的。我還期望著觀眾能夠拉更多人進(jìn)來一起玩呢，事與愿違啊。

由此可得出結(jié)論，我，先得有粉絲，才能玩得起來啊，嗚嗚嗚嗚。大家要是不介意，可以關(guān)注下我的B站賬號(hào)，也叫：蠻三刀醬。我會(huì)偶爾抽風(fēng)發(fā)點(diǎn)有趣的技術(shù)視頻的。

本文實(shí)現(xiàn)的全部代碼已經(jīng)開源在了Github上，大家可以在自己的直播間里試試呀：

https://github.com/qqxx6661/live_comment_control_stream

到此這篇關(guān)于用Python實(shí)現(xiàn)一個(gè)模仿UP主彈幕控制的直播間功能的文章就介紹到這了,更多相關(guān)python彈幕控制的直播間內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: