How to write a Python script to download M3U8 streams

While tools like FFmpeg or youtube-dl are excellent for standard HLS downloads, they can sometimes fall short when dealing with highly customized, authenticated, or uniquely encrypted streams. By building your own Python M3U8 downloader, you can inject custom headers for every segment request, manage proxy rotation, execute concurrent downloads, and programmatically select specific renditions from master playlists.

In this guide, we will engineer a robust Python script that parses an M3U8 file, downloads the segments concurrently, decrypts them if necessary, and merges them into a single video file.

Prerequisites and Libraries

To build a resilient downloader, we should avoid writing a raw regex parser for M3U8 syntax. The HLS specification (RFC 8216) is complex, with many edge cases. Instead, we rely on the m3u8 Python library, alongside requests for networking.

pip install m3u8 requests pycryptodome

• m3u8: Parses the playlist strings into manageable Python objects.
• requests: Handles HTTP communication, allowing custom headers, cookies, and sessions.
• pycryptodome: Provides AES decryption capabilities for encrypted streams.

Parsing the Master and Media Playlists

An M3U8 URL often points to a master playlist containing various quality levels. Your script must detect this, select a media playlist, and parse its segments.

import m3u8
import requests

def get_best_playlist(url, headers):
    playlist = m3u8.load(url, headers=headers)
    
    if playlist.is_variant:
        # Sort by bandwidth, highest first
        variants = sorted(playlist.playlists, key=lambda p: p.stream_info.bandwidth, reverse=True)
        best_uri = variants[0].absolute_uri
        print(f"Selected highest quality variant: {best_uri}")
        return m3u8.load(best_uri, headers=headers)
    
    return playlist

url = "https://example.com/master.m3u8"
headers = {"User-Agent": "Mozilla/5.0"}
media_playlist = get_best_playlist(url, headers)

The m3u8.load() function handles relative paths internally and populates the absolute_uri for subsequent requests, preventing a massive source of bugs in custom downloaders.

Handling AES-128 Encryption

If the stream is encrypted, the media playlist will contain an #EXT-X-KEY tag. For METHOD=AES-128, the entire segment is encrypted using AES in CBC (Cipher Block Chaining) mode. You must fetch the key, extract the Initialization Vector (IV), and decrypt the binary data.

from Crypto.Cipher import AES

def decrypt_segment(encrypted_data, key_uri, iv_hex, seq_num, headers):
    # Fetch the 16-byte key
    key = requests.get(key_uri, headers=headers).content
    
    # Determine the IV
    if iv_hex:
        iv = bytes.fromhex(iv_hex.replace("0x", ""))
    else:
        # Fallback to segment sequence number
        iv = seq_num.to_bytes(16, byteorder='big')
        
    cipher = AES.new(key, AES.MODE_CBC, iv)
    return cipher.decrypt(encrypted_data)

Downloading Segments Concurrently

Downloading segments sequentially is incredibly slow. To saturate your bandwidth, we will use Python's concurrent.futures.ThreadPoolExecutor. We must also ensure that the downloaded segments are merged in the correct sequential order.

import concurrent.futures
import os

def download_segment(segment, index, headers):
    response = requests.get(segment.absolute_uri, headers=headers)
    data = response.content
    
    # Check for encryption
    if segment.key and segment.key.method == "AES-128":
        data = decrypt_segment(data, segment.key.absolute_uri, segment.key.iv, segment.media_sequence, headers)
        
    file_path = f"seg_{index:04d}.ts"
    with open(file_path, "wb") as f:
        f.write(data)
    
    print(f"Downloaded {file_path}")
    return file_path

# Execute downloads concurrently
downloaded_files = []
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
    futures = {executor.submit(download_segment, seg, i, headers): i for i, seg in enumerate(media_playlist.segments)}
    
    # Collect files in order
    results = [None] * len(media_playlist.segments)
    for future in concurrent.futures.as_completed(futures):
        idx = futures[future]
        results[idx] = future.result()
        
downloaded_files = results

Important: Be mindful of the max_workers parameter. Setting it too high might trigger anti-DDoS measures or rate limits on the origin server, resulting in 429 Too Many Requests or 403 Forbidden errors.

Merging Segments into a Single Video

M3U8 segments are typically MPEG-TS (.ts) files. These can technically be concatenated at the binary level (e.g., using cat on Linux or binary writes in Python). However, to fix potential timestamp discontinuities, it is vastly safer to pass the merged file through FFmpeg.

import subprocess

def merge_segments(file_list, output_file="output.mp4"):
    # Create a concatenation text file for FFmpeg
    with open("concat_list.txt", "w") as f:
        for file in file_list:
            f.write(f"file '{file}'\n")
            
    # Run FFmpeg to merge without re-encoding
    command = [
        "ffmpeg", "-f", "concat", "-safe", "0", 
        "-i", "concat_list.txt", "-c", "copy", output_file
    ]
    subprocess.run(command, check=True)
    
    # Cleanup temporary files
    os.remove("concat_list.txt")
    for file in file_list:
        os.remove(file)
        
    print(f"Merge complete: {output_file}")

merge_segments(downloaded_files)

Advanced Troubleshooting Tips

• Stale Tokens: Many CDNs append time-sensitive query parameters (tokens) to the master M3U8 URL. If you parse the master playlist, ensure those token parameters are appended to the segment requests if the m3u8 library doesn't inherit them automatically.
• Handling EXT-X-MAP: Fragmented MP4 (fMP4) streams use an initialization segment declared by #EXT-X-MAP. Your script must download this initialization file and prepend it to the final concatenated file, otherwise the video will not play.
• Retry Logic: Network connections drop. Wrap the requests.get() calls inside a while loop with exponential backoff and a try-except block to catch requests.exceptions.RequestException.

Conclusion

Building a custom Python downloader for M3U8 streams allows you to bypass restrictions that standard tools cannot. By leveraging the m3u8 parser library, Python's concurrency models, and AES-128 decryption utilities, you can engineer a highly resilient and automated media pipeline.