test for unavailable on this app video

unavailabletest.py

# depends: tqdm, requests, brotli
import json
from datetime import datetime
from pathlib import Path
from tqdm import tqdm
import requests.cookies
import multiprocessing
video_id = "7-BqJKE14Yo"
proxies = {}

# Location to save bad requests
DUMP_LOCATION = '/tmp'


newpipe_payload = {
    "contentCheckOk": True,
    "context": {
        "client": {
            "androidSdkVersion": 31,
            "clientName": "ANDROID",
            "clientVersion": "17.31.35",
            "gl": "US",
            "hl": "en-GB",
            "osName": "Android",
            "osVersion": "12",
            "platform": "MOBILE"
        },
        "user": {
            "lockedSafetyMode": False
        }
    },
    "racyCheckOk": True,
    "videoId": video_id
}

newpipe_headers = {
    'User-agent': 'com.google.android.youtube/17.31.35 (Linux; U; Android 12; GB) gzip',
    'x-goog-api-format-version': '2',
    'content-type': 'application/json',
    'accept-language': 'en-GB, en;q=0.9',
    'Accept-Encoding': 'gzip',
}

yt_dlp_payload = {
    "contentCheckOk": True,
    "context": {
        "client": {
            "androidSdkVersion": 30,
            "clientName": "ANDROID",
            "clientVersion": "17.29.34",
            "hl": "en",
            "timeZone": "UTC",
            "utcOffsetMinutes": 0
        }
    },
    "params": "8AEB",
    "playbackContext": {
        "contentPlaybackContext": {
            "html5Preference": "HTML5_PREF_WANTS"
        }
    },
    "racyCheckOk": True,
    "videoId": video_id
}

yt_dlp_headers = {
    'User-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.106 Safari/537.36',
    'X-Youtube-Client-Name': '3',
    'X-Youtube-Client-Version': '17.29.34',
    'Origin': 'https://www.youtube.com',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
    'content-type': 'application/json',
    'accept-language': 'en-us,en;q=0.5',
    'Sec-Fetch-Mode': 'navigate',
    'Accept-Encoding': 'gzip, deflate, br',
}

tests = {'newpipe': (newpipe_payload, newpipe_headers), 'yt-dlp': (yt_dlp_payload, yt_dlp_headers)}


def dump_request(response, video_id, timestamp, reason):
    filename = f'{timestamp}_{video_id}_{reason}.json'
    with open(Path(DUMP_LOCATION) / filename, 'w') as f:
        json.dump(response.json(), f, indent=4)


def listener(q):
    bars = {}
    for t_name in tests:
        bars[t_name] = tqdm(desc=t_name)
        bars[t_name+'_bad'] = tqdm(desc=t_name + " (bad)")

    for bar_name in iter(q.get, None):
        bars[bar_name].update(1)


session = requests.Session()
ip = None


def do_test(tests, q):
    while True:
        for t_name, data in tests.items():
            payload = data[0]
            headers = data[1]

            unix_timestamp = datetime.now().timestamp()
            session.cookies = requests.cookies.cookiejar_from_dict({})
            res = session.post(
                f'https://youtubei.googleapis.com/youtubei/v1/player?key=AIzaSyA8eiZmM1FaDVjRy-df2KTyQ_vz_yYM39w&prettyPrint=false',
                json=payload, headers=headers, proxies=proxies)
            data = res.json()

            pr_video_id = data['videoDetails']['videoId']
            if video_id != pr_video_id:
                dump_request(res, video_id, unix_timestamp, 'video_id')
                q.put(t_name+'_bad')
            else:
                q.put(t_name)


if __name__ == '__main__':
    q = multiprocessing.Queue()
    listener_p = multiprocessing.Process(target=listener, args=(q,))
    listener_p.start()
    processes = []
    for n in range(8):
        p = multiprocessing.Process(target=do_test, args=(tests, q))
        p.start()
        processes.append(p)
    listener_p.join()