一个 m3u8 文件里藏了多少坑：HLS 解析、AES 解密与无损合并实战

#EXTM3U
#EXT-X-STREAM-INF:BANDWIDTH=4500000,RESOLUTION=1920x1080,CODECS="avc1.640028"
1080p/index.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=2500000,RESOLUTION=1280x720,CODECS="avc1.64001f"
720p/index.m3u8

#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:10
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-KEY:METHOD=AES-128,URI="key.bin",IV=0x00000000000000000000000000000001
#EXTINF:10.0,
seg0.ts
#EXTINF:10.0,
seg1.ts
#EXT-X-ENDLIST

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed

def download_segment(url, path, retries=3):
    for attempt in range(retries):
        try:
            # Range 请求实现断点续传：已有多少字节就从哪继续
            downloaded = 0
            if os.path.exists(path):
                downloaded = os.path.getsize(path)
            headers = {"Range": f"bytes={downloaded}-"} if downloaded else {}
            with requests.get(url, headers=headers, stream=True, timeout=30) as r:
                if r.status_code in (200, 206):
                    mode = "ab" if downloaded else "wb"
                    with open(path, mode) as f:
                        for chunk in r.iter_content(65536):
                            f.write(chunk)
                    return path
                r.raise_for_status()
        except Exception:
            if attempt == retries - 1:
                raise
            time.sleep(1.5 ** attempt)  # 指数退避

def download_all(segments, out_dir, workers=8):
    os.makedirs(out_dir, exist_ok=True)
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {
            ex.submit(download_segment, seg["url"], os.path.join(out_dir, seg["name"])): seg
            for seg in segments
        }
        for fut in as_completed(futures):
            fut.result()  # 有异常会在这里抛出来

from Crypto.Cipher import AES

def decrypt_segment(data, key, iv):
    cipher = AES.new(key, AES.MODE_CBC, iv)
    decrypted = cipher.decrypt(data)
    # 去掉 PKCS#7 填充
    pad_len = decrypted[-1]
    return decrypted[:-pad_len]

#EXT-X-MAP:URI="init.mp4"

# 生成列表文件 list.txt，内容形如：
#   file 'seg0.ts'
#   file 'seg1.ts'
#   ...
ffmpeg -f concat -safe 0 -i list.txt -c copy output.mp4

file 'init.mp4'
file 'seg0.m4s'
file 'seg1.m4s'

import os, re, time, subprocess
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urljoin
from Crypto.Cipher import AES

def fetch(url):
    r = requests.get(url, timeout=30)
    r.raise_for_status()
    return r.text.lstrip('\ufeff')

def get_base(url):
    return url.rsplit('/', 1)[0] + '/'

def parse_media(text, base):
    """解析 media playlist，返回 segments 和加密信息"""
    segments, key_info, seq = [], None, 0
    seq_match = re.search(r'#EXT-X-MEDIA-SEQUENCE:(\d+)', text)
    if seq_match:
        seq = int(seq_match.group(1))
    key_match = re.search(r'#EXT-X-KEY:METHOD=([^,]+)(?:,URI="([^"]+)")?(?:,IV=0x([0-9A-Fa-f]+))?', text)
    if key_match and key_match.group(1) != 'NONE':
        key_info = {
            'uri': urljoin(base, key_match.group(2)),
            'iv': bytes.fromhex(key_match.group(3)) if key_match.group(3) else None,
        }
    lines = [l for l in text.splitlines() if l.strip() and not l.strip().startswith('#')]
    for i, uri in enumerate(lines):
        segments.append({'url': urljoin(base, uri), 'name': uri.split('/')[-1], 'index': seq + i})
    return segments, key_info

def decrypt_file(path, key, iv):
    with open(path, 'rb') as f:
        data = f.read()
    dec = AES.new(key, AES.MODE_CBC, iv).decrypt(data)
    with open(path, 'wb') as f:
        f.write(dec[:-dec[-1]])  # 去 PKCS#7 填充

def main(m3u8_url, out='output.mp4', workers=8):
    text = fetch(m3u8_url)
    base = get_base(m3u8_url)
    # 是 master 就取第一档（这里简化，实际按第二节的选流策略挑）
    if '#EXT-X-STREAM-INF' in text:
        uri = [l for l in text.splitlines() if l.strip() and not l.startswith('#')][0]
        m3u8_url = urljoin(base, uri)
        base = get_base(m3u8_url)
        text = fetch(m3u8_url)
    segments, key_info = parse_media(text, base)

    tmp = 'segments'
    os.makedirs(tmp, exist_ok=True)
    with ThreadPoolExecutor(max_workers=workers) as ex:
        list(ex.map(lambda s: download(s['url'], os.path.join(tmp, s['name'])), segments))

    if key_info:  # 解密
        key = requests.get(key_info['uri'], timeout=30).content
        for s in segments:
            iv = key_info['iv'] or s['index'].to_bytes(16, 'big')
            decrypt_file(os.path.join(tmp, s['name']), key, iv)

    # 合并
    with open('list.txt', 'w') as f:
        for s in segments:
            f.write(f"file '{os.path.join(tmp, s['name'])}'\n")
    subprocess.run(['ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', 'list.txt', '-c', 'copy', out], check=True)