getm3u8.py

#!/usr/bin/env python
import os.path
import sys
import time
import urllib.parse
import urllib.request
from typing import Iterable

def get_segment_urls(m3u8_url: str) -> Iterable[str]:
    m3u8_text: str = urllib.request.urlopen(m3u8_url).read().decode('utf-8')
    for line in m3u8_text.splitlines():
        if line.startswith("#"):
            continue
        yield urllib.parse.urljoin(m3u8_url, line)

def download_segment(segment_url: str, destination: str) -> None:
    if os.path.exists(destination):
        print(f"Skip {destination}", file=sys.stderr)
    else:
        print(f"Download {segment_url}", file=sys.stderr)
        content: bytes = urllib.request.urlopen(segment_url, timeout=30).read()
        with open(destination, "wb") as f:
            f.write(content)

def download_segments(m3u8_url: str, output_dir: str) -> None:
    segment_urls = list(get_segment_urls(m3u8_url))
    for i in range(len(segment_urls)):
        segment_url = segment_urls[i]
        _, ext = os.path.splitext(segment_url)
        output_path = os.path.join(output_dir, "{:05d}{}".format(i, ext))
        progress: float = 100 * i / len(segment_urls)
        print("[{:04.2f}%] ".format(progress), file=sys.stderr, end="")
        download_segment(segment_url, output_path)

def main(m3u8_url: str, output_dir: str) -> None:
    success = False
    while not success:
        try:
            download_segments(m3u8_url, output_dir)
            success = True
        except KeyboardInterrupt:
            return
        except Exception as e:
            print(e, file=sys.stderr)
            print("Waiting before retrying...", file=sys.stderr)
            time.sleep(30)

if len(sys.argv) != 3:
    print(f"Usage: {sys.argv[0]} M3U8_URL OUTPUT_DIR")
    sys.exit(1)
else:
    main(sys.argv[1], sys.argv[2])