Python
11 分钟阅读M3U8 Python 下载脚本
基于 HLS 协议的流媒体分片抓取、AES-128 解密与多线程合并的 Python 实现
现代在线视频平台普遍采用 HLS（HTTP Live Streaming） 作为点播与直播的传输协议：服务端将视频切分为若干 .ts 分片，客户端通过 .m3u8 播放列表按序拉取并播放。对开发者而言，理解这套「清单 + 分片」的结构，是编写下载工具的前提。
本文介绍一个轻量 Python 脚本，完成 M3U8 解析 → 多线程分片下载 → AES-128 解密 → FFmpeg 合并 的完整流水线，适用于自有内容归档、内网测试流调试等场景。关于 M3U 播放列表与 IPTV 生态的背景，可参考 IPTV 入门。
一、HLS 协议要点
HLS 由 Apple 提出，现已成为 RFC 8216 标准。与一次性下载完整 MP4 不同，HLS 将媒体流拆成短时长分片（通常 2–10 秒），通过 M3U8 文本清单描述分片 URL、加密方式与码率变体。
| 概念 | 说明 |
|---|---|
| Master Playlist | 多码率入口，含 #EXT-X-STREAM-INF 与 BANDWIDTH 字段，指向各清晰度子清单 |
| Media Playlist | 实际分片列表，每行一个 .ts URL，可含 #EXT-X-KEY 加密声明 |
Segment（.ts） | MPEG-TS 容器分片，下载后需按序合并 |
#EXT-X-KEY | 声明加密方式（常见 METHOD=AES-128）及密钥 URI、IV |
脚本的处理逻辑与上述结构一一对应：先判断是否为 Master Playlist 并选取最高带宽变体，再解析 Media Playlist 中的分片 URL 与密钥信息，最后并发下载并合并。
二、脚本能力概览
| 模块 | 能力 |
|---|---|
| 清单解析 | 支持 Master / Media 两级 Playlist；自动选取 BANDWIDTH 最高的 variant |
| 并发下载 | ThreadPoolExecutor 多线程拉取分片，默认 16 线程，可配置 |
| 断点续传 | 已存在的分片文件自动跳过，适合中断后重试 |
| AES-128 解密 | 解析 #EXT-X-KEY，通过 pycryptodome 做 CBC 解密 |
| 合并输出 | 优先调用 ffmpeg -f concat -c copy 封装为 MP4；无 FFmpeg 时退化为二进制拼接 .ts |
| 进度反馈 | 集成 tqdm 显示分片下载进度 |
依赖安装
pip install requests tqdm pycryptodome
# 合并为 MP4 还需系统安装 ffmpeg
brew install ffmpeg # macOS三、核心流程
输入 m3u8 URL
│
├─ Master Playlist? ──是──▶ 选取最高 BANDWIDTH variant
│ │
└─ Media Playlist ◀───────┘
│
├─ 解析 segments[] + key_info (AES-128)
│
├─ ThreadPoolExecutor 并发下载 → tmp/{000000.ts, 000001.ts, ...}
│ └─ 已存在分片跳过；失败分片记录后中止
│
└─ 合并
├─ ffmpeg 可用 → concat demuxer → output.mp4
└─ 否则 → 二进制拼接 → output.tsVariant 选取：遍历 #EXT-X-STREAM-INF 行，解析 BANDWIDTH 数值，取最大值对应的子清单 URL。适用于自适应码率（ABR）流，确保下载最高清晰度版本。
AES-128 解密：从 #EXT-X-KEY 提取 URI 与 IV（支持 0x 前缀十六进制），首次解密时拉取密钥字节并缓存；若 Playlist 未声明 IV，脚本回退为零 IV（部分流可能因此解密异常，需结合具体源验证）。
原子写入：每个分片先写入 .part 临时文件，完成后 os.replace 落盘，避免中断产生损坏分片。
四、完整代码
import os
import re
import sys
import shutil
import argparse
import tempfile
import subprocess
from urllib.parse import urljoin, urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
try:
import requests
except ImportError:
print("请先安装依赖: pip install requests tqdm")
sys.exit(1)
try:
from tqdm import tqdm
except Exception:
tqdm = None
try:
from Crypto.Cipher import AES
have_crypto = True
except Exception:
have_crypto = False
HEADERS = {"User-Agent": "m3u8-downloader/1.0"}
def fetch_text(url, session=None, **kwargs):
s = session or requests
resp = s.get(url, headers=HEADERS, timeout=15, **kwargs)
resp.raise_for_status()
return resp.text
def pick_variant(m3u8_text, base_url):
"""若是 Master Playlist，选取 BANDWIDTH 最大的 variant。"""
lines = m3u8_text.strip().splitlines()
variant_infos = []
for i, line in enumerate(lines):
if line.startswith("#EXT-X-STREAM-INF"):
m = re.search(r"BANDWIDTH=(\d+)", line)
bw = int(m.group(1)) if m else 0
for j in range(i + 1, len(lines)):
u = lines[j].strip()
if u and not u.startswith("#"):
variant_infos.append((bw, urljoin(base_url, u)))
break
if not variant_infos:
return None
variant_infos.sort(key=lambda x: x[0], reverse=True)
return variant_infos[0][1]
def parse_segments(m3u8_text, base_url):
"""解析分片 URL 列表与 AES-128 密钥信息。"""
lines = m3u8_text.strip().splitlines()
segments = []
key = None
for line in lines:
line = line.strip()
if line.startswith("#EXT-X-KEY"):
m_method = re.search(r"METHOD=([^,]+)", line)
m_uri = re.search(r'URI="([^"]+)"', line)
m_iv = re.search(r"IV=([^,]+)", line)
method = m_method.group(1) if m_method else None
uri = urljoin(base_url, m_uri.group(1)) if m_uri else None
iv = None
if m_iv:
ivtxt = m_iv.group(1)
if ivtxt.startswith(("0x", "0X")):
iv = bytes.fromhex(ivtxt[2:])
else:
try:
iv = bytes.fromhex(ivtxt)
except Exception:
iv = None
key = {"method": method, "uri": uri, "iv": iv}
if line and not line.startswith("#"):
segments.append(urljoin(base_url, line))
return segments, key
def download_segment(idx, url, dest_path, session, retries=3, key=None):
"""下载单个分片，支持 AES-128 解密。"""
tmp_path = dest_path + ".part"
if os.path.exists(dest_path):
return True, dest_path
for attempt in range(1, retries + 1):
try:
r = session.get(url, headers=HEADERS, timeout=30, stream=True)
r.raise_for_status()
data = r.content
if key and key.get("method", "").upper() == "AES-128":
if not have_crypto:
raise RuntimeError("需要 pycryptodome 支持 AES 解密，但未安装。")
if not key.get("_key_bytes"):
kb = session.get(key["uri"], headers=HEADERS, timeout=15)
kb.raise_for_status()
key["_key_bytes"] = kb.content
iv = key.get("iv") or (b"\x00" * 16)
cipher = AES.new(key["_key_bytes"], AES.MODE_CBC, iv=iv)
data = cipher.decrypt(data)
with open(tmp_path, "wb") as f:
f.write(data)
os.replace(tmp_path, dest_path)
return True, dest_path
except Exception as e:
if attempt == retries:
return False, str(e)
return False, "unknown error"
def merge_with_ffmpeg(segment_files, output_file):
"""通过 ffmpeg concat demuxer 无损合并分片。"""
tmpdir = os.path.dirname(output_file) or "."
list_file = os.path.join(tmpdir, "ff_concat_list.txt")
with open(list_file, "w", encoding="utf-8") as f:
for seg in segment_files:
f.write(f"file '{os.path.abspath(seg)}'\n")
cmd = [
"ffmpeg", "-hide_banner", "-loglevel", "error",
"-f", "concat", "-safe", "0", "-i", list_file,
"-c", "copy", output_file,
]
try:
subprocess.check_call(cmd)
os.remove(list_file)
return True, None
except subprocess.CalledProcessError as e:
return False, str(e)
def download_m3u8(url, out, threads=16, tmp=None, keep_ts=True, session=None):
session = session or requests.Session()
text = fetch_text(url, session=session)
variant = pick_variant(text, url)
if variant:
print(f"检测到 Master Playlist，选择 variant: {variant}")
text = fetch_text(variant, session=session)
base_for_join = variant.rsplit("/", 1)[0] + "/"
else:
base_for_join = url.rsplit("/", 1)[0] + "/"
segments, key = parse_segments(text, base_for_join)
if not segments:
raise RuntimeError("未找到任何分片 —— 请确认 URL 为有效的 Media Playlist。")
print(f"共 {len(segments)} 个分片，并发线程数={threads}")
tmpdir = tmp or tempfile.mkdtemp(prefix="m3u8_dl_")
os.makedirs(tmpdir, exist_ok=True)
seg_files = [os.path.join(tmpdir, f"{i:06d}.ts") for i in range(len(segments))]
if key and key.get("method", "").upper() == "AES-128":
if not have_crypto:
print("警告：Playlist 使用 AES-128，但未安装 pycryptodome。")
else:
print(f"检测到 AES-128 加密，key URI: {key.get('uri')}")
failures = []
pbar = tqdm(total=len(segments), desc="下载", unit="seg") if tqdm else None
with ThreadPoolExecutor(max_workers=threads) as exe:
futures = {
exe.submit(download_segment, i, seg, seg_files[i], session, 5, key): i
for i, seg in enumerate(segments)
}
for fut in as_completed(futures):
ok, info = fut.result()
if not ok:
failures.append((futures[fut], info))
if pbar:
pbar.update(1)
if pbar:
pbar.close()
if failures:
print(f"{len(failures)} 个分片下载失败，中止合并。")
raise RuntimeError("存在下载失败的分片。")
if shutil.which("ffmpeg"):
print("使用 ffmpeg 合并...")
success, err = merge_with_ffmpeg(seg_files, out)
if success:
print(f"完成 → {out}")
if not keep_ts:
shutil.rmtree(tmpdir, ignore_errors=True)
return out
print(f"ffmpeg 合并失败: {err}，退化为二进制拼接。")
out_ts = out if out.lower().endswith(".ts") else out + ".ts"
with open(out_ts, "wb") as outf:
for seg in seg_files:
with open(seg, "rb") as f:
shutil.copyfileobj(f, outf)
print(f"拼接完成 → {out_ts}")
if not keep_ts:
shutil.rmtree(tmpdir, ignore_errors=True)
return out_ts
def cli(url: str, output_base: str, threads: int = 32):
parser = argparse.ArgumentParser(description="多线程下载 M3U8 (HLS) 视频并合并")
parser.add_argument("--keep-ts", action="store_true", help="保留临时 ts 文件")
args = parser.parse_args()
output = output_base + ".mp4"
temp = output_base + "_temp"
try:
out = download_m3u8(url, output, threads=threads, tmp=temp, keep_ts=args.keep_ts)
print("完成：", out)
except Exception as e:
print("出错：", e)
sys.exit(1)
if __name__ == "__main__":
videos = [
# {"url": "https://example.com/stream.m3u8", "name": "output-filename"},
]
if not videos:
print("请在 videos 列表中配置待下载资源。")
else:
for video in videos:
cli(url=video["url"], output_base=video["name"])五、使用方式
在 videos 列表中填入目标 URL 与输出文件名（不含扩展名），直接运行脚本；或通过命令行封装调用 cli()：
python m3u8_downloader.py
# 输出 → {name}.mp4，临时分片存放在 {name}_temp/批量下载时，在 videos 数组中追加多条 {url, name} 即可顺序执行。
六、已知局限
| 局限 | 说明 |
|---|---|
| 加密范围 | 仅支持 AES-128；SAMPLE-AES、DRM（Widevine / FairPlay）无法处理 |
| IV 回退 | Playlist 未提供 IV 时使用零 IV，部分 CDN 源可能解密失败 |
| 直播流 | #EXT-X-ENDLIST 缺失的直播清单会持续更新，本脚本面向点播（VOD）设计 |
| 合并兼容性 | 无 FFmpeg 时输出裸 .ts，需播放器支持 TS 容器 |
| 合规性 | 仅下载有权访问的内容；请遵守目标站点的服务条款与版权法规 |
七、与 IPTV / 爬虫场景的关系
- IPTV 场景：直播源通常为滚动更新的 Media Playlist，本脚本更适合点播回放或固定清单的归档下载。频道列表获取方式见 IPTV 部署指南。
- 爬虫场景：部分 Web 平台的视频地址以
.m3u8形式存在于页面接口或 DevTools 网络面板中，抓取逻辑可参考 Python 爬虫 先定位清单 URL，再交由本脚本完成分片层下载。
两者分工明确：爬虫负责「找到 m3u8 链接」，本脚本负责「把 HLS 流还原为本地文件」。