如何在youtube直播中获得实时反应？

Question

我需要仅使用 api 来获得实时反应和喜欢/不喜欢，而不需要任何库。使用 Node.js。

我正在尝试从聊天接收短信的终点获取实时反应和喜欢/不喜欢。端点仅获取短信。

Answer 1

以下网络抓取 Python 脚本使用

CHANNEL_ID

和

VIDEO_ID

参数化，打印当前反应，例如：

[
    {
        "totalReactions": 2,
        "duration": {
            "seconds": "1"
        },
        "intensityScore": 0.75,
        "reactionsData": [
            {
                "unicodeEmojiId": "\ud83c\udf89",
                "reactionCount": 1
            },
            {
                "unicodeEmojiId": "\ud83d\ude04",
                "reactionCount": 1
            }
        ]
    },
    {
        "totalReactions": 1,
        "duration": {
            "seconds": "1"
        },
        "intensityScore": 0.75,
        "reactionsData": [
            {
                "unicodeEmojiId": "\u2764",
                "reactionCount": 1
            }
        ]
    },
    {
        "totalReactions": 0,
        "duration": {
            "seconds": "1"
        },
        "intensityScore": 1
    },
    {
        "totalReactions": 0,
        "duration": {
            "seconds": "1"
        },
        "intensityScore": 1
    },
    {
        "totalReactions": 0,
        "duration": {
            "seconds": "1"
        },
        "intensityScore": 1
    }
]

import requests
import json
import blackboxprotobuf
import base64
import time
from datetime import datetime

CHANNEL_ID = 'THE_CHANNEL_ID'
VIDEO_ID = 'THE_VIDEO_ID'

url = 'https://www.youtube.com/youtubei/v1/live_chat/get_live_chat'

headers = {
    'Content-Type': 'application/json',
}

def getBase64Protobuf(message, typedef):
    data = blackboxprotobuf.encode_message(message, typedef)
    return base64.b64encode(data, altchars = b'-_')

message = {
    '1': {
        '5': {
            '1': CHANNEL_ID,
            '2': VIDEO_ID
        }
    },
    '4': 2,
}

typedef = {
    '1': {
        'type': 'message',
        'message_typedef': {
            '5': {
                'type': 'message',
                'message_typedef': {
                    '1': {
                        'type': 'string'
                    },
                    '2': {
                        'type': 'string'
                    }
                },
                'field_order': [
                    '1',
                    '2'
                ]
            }
        },
        'field_order': [
            '5'
        ]
    },
    '4': {
        'type': 'int'
    },
}

three = getBase64Protobuf(message, typedef)

message = {
    '119693434': {
        '3': three,
    }
}

typedef = {
    '119693434': {
        'type': 'message',
        'message_typedef': {
            '3': {
                'type': 'string'
            },
            '20': {
                'type': 'int'
            },
        },
        'field_order': [
            '3',
            '20'
        ]
    }
}

json_ = {
    'context': {
        'client': {
            'clientName': 'WEB',
            'clientVersion': '2.20240509.00.00'
        }
    },
}
    
initialDatetime = datetime.now()
print(initialDatetime)
    
while True:
    message['119693434']['20'] = int((time.time() + 400) * 1_000_000)
    continuation = getBase64Protobuf(message, typedef)
    json_['continuation'] = continuation
    
    data = requests.post(url, headers = headers, json = json_).json()
    dataStr = json.dumps(data, indent = 4)
    if 'frameworkUpdates' in data:
        print(json.dumps(data['frameworkUpdates']['entityBatchUpdate']['mutations'][0]['payload']['emojiFountainDataEntity']['reactionBuckets'], indent = 4))
    isNeedleInData = 'invalidationContinuationData' in dataStr
    print(isNeedleInData, datetime.now() - initialDatetime)
    if not isNeedleInData:
        break
    time.sleep(10)