简介基于 Python 的 m3u8 下载器是一款面向视频下载需求者的轻量开源工具用于解析 m3u8 播放列表并批量下载其中链接的 ts 视频片段最终合并为完整视频。它适合需要离线观看、备份在线视频或研究流媒体协议的学习者与开发者使用只需提供 m3u8 文件地址即可自动完成下载流程。压缩包共 3 个文件包含 2 个 Python 脚本与 1 个 Markdown 说明文档整体体积仅 3 KB结构精简目录清晰主程序脚本与适配 Python3 的版本并存便于对照阅读和直接运行。已有 62 人学习下载资源虽小但代码逻辑完整涵盖 m3u8 解析、网络请求、异常处理等关键环节可作为理解流媒体下载原理的入门示例也可以在此基础上扩展多线程下载、进度显示、断点续传或调用 ffmpeg 转封装等高级功能。1. 为什么要自己写一个 m3u8 下载器直接在浏览器里看 m3u8 视频很容易但想“保存”下来就没那么顺手。播放器只按需拉取当前要播的几秒分片不会帮你把整段节目落盘DevTools 里能看到一串 .ts 请求但要手动逐个保存几百个分片再拼起来纯属折磨。用 Python 写一个下载器核心就三件事解析播放列表、并发抓取 ts 分片、用 ffmpeg 合并转封装。这篇文章不打算做“玩具脚本”而是把工程细节拆开包括加密流的 key 处理、断点续传、直播流追加录制以及 m3u8 视频转换失败时怎么定位根因。适合已经跑通 Python 基础、想认真搞定流媒体下载的开发者你不需要太深的视频编解码知识但对网络请求和文件 IO 要有点感觉。如果还没装 Python 环境按标准的 python 安装教程配好 3.8 就行下文默认你已有可用的解释器和 pip。2. m3u8 协议与 ts 片段解析从播放列表到真实地址m3u8 本质上是一个 UTF-8 文本清单每行是一个 URI 或一个#开头的标签。理解它的结构是写下载器的第一步。下面的示例是一个点播VOD媒体列表#EXTM3U #EXT-X-VERSION:3 #EXT-X-TARGETDURATION:10 #EXT-X-MEDIA-SEQUENCE:0 #EXTINF:10.0, https://cdn.example.com/hls/vod/seg0.ts #EXTINF:10.0, https://cdn.example.com/hls/vod/seg1.ts #EXT-X-ENDLIST#EXTINF标签后面是本分片的时长和逗号下一行才是真正的 ts 地址。注意地址可能是相对路径比如seg0.ts或../asset/seg0.ts此时必须基于 m3u8 文件自身的 URL 做拼接。最常见的坑是播放列表在https://cdn.example.com/hls/vod/playlist.m3u8里边的seg0.ts实际对应https://cdn.example.com/hls/vod/seg0.ts但如果写的是/hls/vod/seg0.ts前面的域名就得从原 URL 的 scheme netloc 推导。用urllib.parse.urljoin能统一处理这两种情况。from urllib.parse import urljoin def parse_m3u8(text: str, base_url: str): segments [] for line in text.splitlines(): line line.strip() if line and not line.startswith(#): segments.append(urljoin(base_url, line)) return segments这个函数把所有非#开头且非空的行都当作 ts 地址urljoin会自动处理绝对/相对路径包括../回退。但真实场景里有两种变体需要额外处理。第一种是#EXTINF和 ts 地址写在同一行某些低质量源会输出#EXTINF:10.0,seg0.ts上面的逐行判断会把整行误判为地址需要用正则拆出逗号后面的部分。第二种是主列表Master Playlist里面没有直接的 ts而是按码率给了多个子播放列表#EXTM3U #EXT-X-STREAM-INF:BANDWIDTH1280000,RESOLUTION1280x720 720p/index.m3u8 #EXT-X-STREAM-INF:BANDWIDTH512000,RESOLUTION640x360 360p/index.m3u8在拿到这种响应时直接按上面的函数解析得到的是子列表 URL而不是 ts。你需要递归解析找到#EXT-X-STREAM-INF后请求其后的子列表直到某个列表里出现#EXTINF。选码率的逻辑我一般写成一个可配置函数默认取BANDWIDTH最大值对应的档位同时让用户能通过命令行参数手动指定清晰度。BANDWIDTH的单位是 bps用正则rBANDWIDTH(\d)提取后转整数比较即可。下面整理一个解析时需要关注的标签速查表。这不是 HLS 协议的完整规范只覆盖下载器最常用到的部分。标签含义下载器处理策略#EXTM3U文件头忽略#EXT-X-VERSION协议版本忽略兼容性影响小#EXT-X-STREAM-INF子流信息递归解析子列表#EXTINF分片时长后跟 ts 地址取下一行作为分片 URL#EXT-X-BYTERANGE分片是某个大文件的字节区间需要 HTTP Range 请求#EXT-X-KEY分片加密方式记录 key 和 IV用于第 5 章解密#EXT-X-ENDLIST点播结束标记无该标签代表直播流#EXT-X-BYTERANGE不常见但遇到时值得留意。它表示多个“逻辑分片”其实存放在同一个物理文件里每个分片只占用一段字节区间比如#EXT-X-BYTERANGE:2000000表示从文件开头偏移 0 处取 200000 字节。你需要在下载时拼接Range: bytesstart-end请求头然后单独保存成独立的 ts。我遇到过一个冷门视频站这么干当时没处理合并出来的视频全部花屏排查了半天才发现是分片切割方式不对。如果你的目标是兼容尽可能多的源这个标签需要支持。解析完的分片列表不要只留在内存里。下载几百个分片通常要几分钟中途 Python 进程一崩前面的解析和下载记录全丢。我会把解析结果落盘到segments.txt每行一个 URL同时把#EXTINF的时长和#EXT-X-KEY信息单独写到meta.json。这样断点续传时可以直接读取文件不必重新请求 m3u8。另外点播流的列表是固定的直播流则没有#EXT-X-ENDLIST需要定期重新拉取列表追加新出现的分片这个机制放到第 5 章再展开。3. 下载核心requests 与并发抓取 ts 片段的工程细节拿到分片 URL 列表后最直接的方式是 for 循环挨个下载。但一个 1080p 的视频经常有三四百个分片每个分片几百 KB串行下载往往要几分钟到十几分钟速度完全取决于网络延迟。常见做法是用线程池控制并发数在 816 之间既能跑满带宽又不至于被源站限流或触发封禁。下面这个下载函数加了超时、重试、断点跳过和写入临时文件几个关键特性import requests from pathlib import Path import time HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36 } def download_ts(url: str, save_path: str, timeout10, retries3, min_size1024): save_file Path(save_path) if save_file.exists() and save_file.stat().st_size min_size: return True for attempt in range(retries): try: with requests.get(url, headersHEADERS, timeouttimeout, streamTrue) as r: r.raise_for_status() with open(save_file, wb) as f: for chunk in r.iter_content(chunk_size64 * 1024): f.write(chunk) if save_file.stat().st_size min_size: raise IOError(f文件过小: {save_file.stat().st_size} bytes) return True except (requests.RequestException, IOError) as e: print(f第 {attempt1} 次失败: {url} - {e}) time.sleep(2 * (attempt 1)) return False参数说明timeout设为 10避免某个分片因为连接重置而卡死整个队列retries3是重试次数退避时间用2 * (attempt 1)控制也就是第 1 次失败等 2 秒、第 2 次等 4 秒、第 3 次等 6 秒让网络抖动有时间恢复。min_size1024用来过滤源站返回的错误页——很多 CDN 在鉴权失败时会返回一个几 KB 的 HTML 错误页如果不检查会把它当 ts 存下来合并时直接 decode 失败。streamTrue是必须的它让响应体按 64KB 迭代写入磁盘避免大文件一次性载入内存。并发部分用concurrent.futures.ThreadPoolExecutor已经足够不需要引入复杂的异步框架。下面的代码把分片列表包装成 future每完成一个就更新进度并把失败的分片单独收集from concurrent.futures import ThreadPoolExecutor, as_completed def batch_download(segments, output_dir, max_workers8): output_dir Path(output_dir) output_dir.mkdir(parentsTrue, exist_okTrue) done, failed 0, [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map {} for idx, url in enumerate(segments): # 用 5 位零填充保证字典序和播放顺序一致 save_path output_dir / fsegment_{idx:05d}.ts future_map[executor.submit(download_ts, url, save_path)] (idx, url, save_path) for future in as_completed(future_map): idx, url, save_path future_map[future] try: if future.result(): done 1 else: failed.append((idx, url)) except Exception as exc: failed.append((idx, url)) print(f\r已下载 {done}/{len(segments)}失败 {len(failed)}, end, flushTrue) print() return failedmax_workers建议从 8 开始调。线程过多时本机上下文切换开销增大而且许多源站会根据来源 IP 做并发限制超过阈值后开始大量返回 503反而更慢。我一般先用 8 测一段观察失败率和平均速度如果失败率接近 0 且 CPU 没跑满再试着调到 16。这里保存文件名用segment_{idx:05d}.ts当分片数超过 99999 时自动变成 6 位不会破坏字典序。有一部分下载器喜欢用原来的 ts 名字但不同源的命名规则差异很大甚至可能重名所以我更推荐用本地序号重命名。requests的线程安全性主要体现在 session 的使用上。同一个requests.Session可以被多个线程共享因为它内部有连接池比每次requests.get创建新连接更高效。改造方式是把download_ts接收一个 session 参数在线程池外创建一次然后传入每个 worker。另外防盗链是绕不开的话题。如果下载时收到403 Forbidden多半是源站要求Referer头必须匹配播放页地址。在HEADERS里补上Referer: base_url一般能解决。如果还是被拦截把浏览器 DevTools 里那个 ts 请求的完整请求头原封不动复制过来通常就能过。如果你在做这类下载器时对性能有更高要求可以换成asyncio aiohttp的协程版本。它和线程池在并发模型上不同但工程上的边界条件是一样的超时、重试、并发数限制、UA 伪装。协程的优势是单线程内更高的连接复用率适合几千个分片的大视频线程池的优点是代码更贴近同步逻辑出错时堆栈更容易定位。对于入门级别的下载器线程池已经足够等遇到“并发数一高就被断连”的问题再考虑协程迁移也不迟。下载完所有分片后别急着合并。先检查failed列表如果非空把失败的 URL 重新丢进线程池跑第二轮——第一轮因为网络抖动失败的分片第二轮成功概率很高。还是失败的把索引和 URL 写入failures.txt方便你手动用wget或迅雷补下也可以作为后续重试的凭据。这一步虽然简单却决定了合并阶段能不能一次通过。4. 合并与转封装ffmpeg 调用和 m3u8 视频转换失败排查ts 分片都齐了之后合并的方式有两种直接拼接二进制文件或者交给 ffmpeg 处理。直接拼接对纯 ts 流有时能成但遇到编码参数不一致、时间戳断裂的情况出来的视频会卡顿或花屏。所以实际上都用 ffmpeg。先拿 Python 把分片文件列表生成为 ffmpeg 需要的 concat 文件from pathlib import Path def write_filelist(ts_dir: Path, filelist_path: Path): ts_files sorted(ts_dir.glob(segment_*.ts)) with open(filelist_path, w, encodingutf-8) as f: for ts in ts_files: f.write(ffile {ts}\n) return filelist_path这里用glob加sorted因为前面保存时已经做了零填充所以排序结果就是播放顺序。接着调用 ffmpeg 合并ffmpeg -f concat -safe 0 -i filelist.txt -c copy output.mp4-f concat告诉 ffmpeg 输入文件不是媒体文件而是一个文件列表-safe 0允许列表里的路径包含相对路径或特殊字符-c copy表示不重新编码直接复制所有流这样合并一个 1GB 的视频只需要几十秒。如果合并后没有声音或者画面在某一片段开始花屏说明某个 ts 分片损坏或编码格式不完整。这时需要把-c copy换回-c:v h264 -c:a aac重新编码代价是耗时翻好几倍。所以更聪明的做法是在合并前先校验文件大小和完整性把异常分片重新下载。还有一种省事的思路既然 ffmpeg 本身支持读取远程 m3u8那你甚至可以跳过大部分下载逻辑直接用这条命令ffmpeg -i https://example.com/playlist.m3u8 -c copy output.mp4ffmpeg 会自己拉取 ts、合并、转封装。但这条命令的问题也很明显任何一处分片拉取失败整个进程直接退出没有断点续传加密流的 key 处理依赖 ffmpeg 编译时是否带 openssl而且对防盗链的响应头控制不如 Python 灵活。所以我的取舍是用 Python 做精细的分片下载和失败重试用 ffmpeg 只做最后的合并。两者各干各擅长的事。下面把最常见的 m3u8 视频转换失败情况整理成表。如果你的 ffmpeg 报错能对上号可以直接按列出的方向处理。错误信息根因解决方案Invalid data found when processing input某个 ts 文件为 0 字节或损坏删除该文件重新下载对应分片Server returned 403 ForbiddenUA/Referer 被源站拦截加-user_agent和-headers参数Non-monotonous DTS in output stream分片时间戳不连续常见于直播录播合并命令加-fflags genptsCould not find codec parameters加密流未解密或 ts 缺头先解密分片再合并Cannot allocate memory文件列表过大ffmpeg 内存不足分批合并先合一部分为临时文件muxer does not support non-seekable output输出到管道或特殊文件指定真实可写文件路径Non-monotonous DTS出现频率不低尤其是在直播源录制的 ts 里。时间戳不连续可能导致播放器快进时卡死。加上-fflags genpts让 ffmpeg 重新生成 PTS能规避大部分这种情况。完整命令是ffmpeg -f concat -safe 0 -i filelist.txt -c copy -fflags genpts output.mp4如果你下载的是加密流还没解密就合并报错往往是Could not find codec parameters或直接Invalid data found。因为 ts 里的数据是被 AES-128 加密后的密文ffmpeg 无法识别其编码格式。遇到这种情况回到第 5 章的解密步骤处理不要试图用-c copy硬拼。合并完成后建议顺手校验一下输出文件的时长是否和源视频一致。可以用 ffprobeffprobe -v error -show_entries formatduration -of defaultnoprint_wrappers1:nokey1 output.mp4把输出的时长和#EXT-X-TARGETDURATION乘以分片总数做个粗略对比误差在几秒内都算正常。如果偏差过大说明中间有分片被静默跳过或内容缺失需要回头检查失败列表。5. 进阶AES-128 加密流的 key 处理和断点续传技巧不少视频站的分片并不是裸 ts而是经过 AES-128 加密的。m3u8 里会出现这样的标签#EXT-X-KEY:METHODAES-128,URIhttps://cdn.example.com/key.key,IV0x1234567890abcdef1234567890abcdefMETHODAES-128表示后续分片用 AES-128-CBC 加密URI指向一个 key 文件IV是初始向量。IV不一定会写缺省时使用分片的 Media Sequence 作为 16 字节大端整数。key 文件本身是 32 字节的二进制不是文本别用读取字符串的方式去读它。加密分片在合并前必须解密。用pycryptodome库可以这样处理from Crypto.Cipher import AES from pathlib import Path def decrypt_ts(src: Path, dst: Path, key: bytes, iv: bytes): aes AES.new(key, AES.MODE_CBC, iv) data src.read_bytes() plain aes.decrypt(data) pad_len plain[-1] # PKCS7 填充字节 dst.write_bytes(plain[:-pad_len])这段代码把整个 ts 读入内存AES-128-CBC 的块大小是 16 字节解密后末尾有一个字节的填充标记需要去掉。对于几百 MB 的单个 ts 不适用但实际每个分片也就几 MB内存开销可以接受。解密后的文件用原来的命名保存到decrypted/子目录filelist 指向解密后的路径再走第 4 章的合并命令。注意如果 m3u8 里出现多个#EXT-X-KEY比如不同时间段换了密钥需要按分片逐个切换 key不能全局用一个。断点续传的细节比想象中更重要。第一版下载器在download_ts里已经判断了“文件已存在且大小足够就跳过”这能应付简单中断但如果源站 URL 带了时效性签名重新运行后 URL 已经过期跳过的文件还在没完成的分片却拿不到新地址了。更稳的方案是维护download_status.json字段里记录分片序号、URL、文件名、文件大小、下载时间。重启时先比较 URL 签名和文件大小决定是跳过还是重新下载。签名失效时重新解析 m3u8用分片序号映射到新 URL再续下未完成的部分。直播流的断点续传逻辑又不一样。直播 m3u8 没有#EXT-X-ENDLIST播放列表会不断追加新的分片。我一般起一个定时任务每 10 秒重新拉一次 m3u8对比已下载序号把新增分片加入下载队列。录制结束后再按第 4 章的流程合并。需要注意直播源的 ts 文件名经常不带递增序号甚至重复使用相同名字所以归档时一定用本地递增序号重命名否则后下载的分片会覆盖先前的最终视频会在某个时间点开始无限重复。本文还有配套的精品资源点击获取