another m3u8 downloader

am3u8-dl.py

#!/usr/bin/env python3
import sys
import os
import aiohttp
import asyncio
from tqdm.asyncio import tqdm
from urllib.parse import urljoin

async def verify_noop(filename):
    return True

async def download_ts(url, verify=verify_noop, chunk_size=512 * 1024, speedbar=None):
    filename = os.path.basename(url)
    filename_dl = filename + '.part'
    if os.path.exists(filename):
        return filename
    async with aiohttp.ClientSession() as session:
        async with session.get(url, ssl=False) as resp:
            if speedbar:
                speedbar.total += int(resp.headers['content-length'])
            with open(filename_dl, 'wb') as fd:
                while True:
                    chunk = await resp.content.read(chunk_size)
                    if not chunk:
                        break
                    if speedbar:
                        speedbar.update(len(chunk))
                    fd.write(chunk)
    if await verify_noop(filename_dl):
        os.rename(filename_dl, filename)
        return filename

async def sem_run(sem, coro):
    async with sem:
        return await coro

async def download_m3u8(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url, ssl=False) as resp:
            async for line in resp.content:
                line = line.decode()
                if line.startswith('#'):
                    continue
                yield urljoin(url, line.strip())

async def merge_ts(files, out, keep=True):
    proc = await asyncio.create_subprocess_exec(
            'ffmpeg', '-hide_banner', '-i', '-', '-c', 'copy', '-y', out,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE)
    for filename in files:
        with open(filename, 'rb') as fd:
            while True:
                chunk = fd.read(512 * 1024)
                if not chunk:
                    break
                proc.stdin.write(chunk)
                await proc.stdin.drain()
        if not keep:
            os.unlink(filename)
    proc.stdin.close()
    # await proc.stdin.wait_closed()
    await proc.wait()

async def download(url, out, jobs=1):
    links = []
    async for link in download_m3u8(url):
        links.append(link)
    sem = asyncio.Semaphore(jobs)
    with tqdm(total=1, desc='xfer', unit='B', unit_scale=True, unit_divisor=1024) as speedbar:
        tasks = [asyncio.ensure_future(sem_run(sem, download_ts(link, speedbar=speedbar))) for link in links]
        files = await tqdm.gather(*tasks, desc='chunk', unit='ts')
    sys.stderr.flush()
    await merge_ts(files, out, keep=False)

async def download_all(args):
    url = args.urls[0]
    name = os.path.basename(url) + '.mp4'
    return await download(url, name, jobs=args.jobs)

def cli():
    import argparse
    parser = argparse.ArgumentParser(description='Another m3u8 downloader')
    parser.add_argument('urls', metavar='URL', nargs='+', help='urls to download')
    parser.add_argument('-j', dest='jobs', type=int, default=2, help='concurrent downloads')
    args = parser.parse_args()

    loop = asyncio.get_event_loop()
    loop.run_until_complete(download_all(args))
    # https://docs.aiohttp.org/en/stable/client_advanced.html#graceful-shutdown
    loop.run_until_complete(asyncio.sleep(0.250))
    loop.close()

if __name__ == '__main__':
    cli()