youtube_hls_dl.py

import asyncio
import os
from urllib.parse import urlparse, parse_qs
from subprocess import Popen

import itertools
import requests
import ffmpeg
import m3u8
from natsort import natsorted


def _get_file_name(base, part):
    return os.path.join('output', f'{base}.part{part}.ts')


async def dl_file(url, path):
    print(f'downloading {url}')
    with open(path, 'wb') as f:
        f.write(requests.get(url).content)


async def load_stream(video_id):
    r = requests.get('https://www.youtube.com/get_video_info', {'video_id': video_id})
    data = parse_qs(r.text)
    meta_stream_url = data['hlsvp'][0]
    meta = m3u8.load(meta_stream_url)
    hls_stream_url = meta.playlists[-1].uri
    part_num = 0
    last_file = None
    hls = m3u8.load(hls_stream_url)
    media_sequence = hls.media_sequence
    for i in itertools.count():
        if not os.path.exists(_get_file_name(video_id, i)):
            part_num = i
            break
    while True:
        try:
            idx = hls.files.index(last_file) + 1
        except ValueError:
            if media_sequence != hls.media_sequence:
                idx = 0
            else:
                idx = -1
        last_file = hls.files[-1]
        for file in hls.files[idx:]:
            file_name = _get_file_name(video_id, part_num)
            ioloop.create_task(dl_file(file, file_name))
            part_num += 1
        await asyncio.sleep(hls.target_duration)
        hls = m3u8.load(hls_stream_url)


if __name__ == '__main__':
    ioloop = asyncio.get_event_loop()
    ioloop.create_task(load_stream('ueupsBPNkSc'))
    ioloop.run_forever()