简介：M3U8批量视频流下载工具面向需要离线保存网页视频的普通用户与运维人员，解决m3u8流媒体地址解析困难、逐条下载繁琐的问题。压缩包共13个文件，约23.6MB，以exe可执行程序、json配置、log日志、conf参数文件及dat数据文件为主，其中主程序与ffmpeg、aria2c等组件配合，可完成地址解析、真实链接获取与下载后转码，日志与配置文件则便于调整并发、缓存等参数并排查异常。目前已有3481人学习下载。使用者可借助该工具批量抓取m3u8视频流，下载完成后直接转码为常见格式在电脑播放，同时通过日志与配置了解下载流程与排错思路，适合需要批量收藏在线视频、研究流媒体下载机制的人群参考使用。
1. 批量下载 M3U8 这件事，比你想的更值得认真对待
如果你手头攒了几百个课程视频链接，每个都是.m3u8结尾，点开浏览器能播，但想存到本地就犯难——右键没有“另存为”，录屏又掉画质，这时候一套能跑通的批量下载方案就是刚需。M3U8 本身不是视频文件，它是一份文本索引，里面按时间顺序列着一堆.ts分片地址，播放器拿到索引后逐片拉取再拼接，才成了你看到的完整视频。批量下载的核心工作就两步：解析索引拿到分片清单，并发拉取后按顺序合并成 MP4。这套流程适合做课程归档、素材备份、离线学习的从业者，也适合想搞懂流媒体落地链路的开发者。下面我从索引结构讲到并发控制，再到合并与排错，把这条链路拆开。
2. M3U8 索引结构与批量任务拆解：先看懂再动手
2.1 索引文件里到底写了什么
一个典型的 M3U8 文件长这样：
#EXTM3U #EXT-X-VERSION:3 #EXT-X-TARGETDURATION:10 #EXT-X-MEDIA-SEQUENCE:0 #EXTINF:10.000000, segment_000.ts #EXTINF:10.000000, segment_001.ts #EXTINF:9.960000, segment_002.ts #EXT-X-ENDLIST#EXTM3U是固定头，#EXT-X-TARGETDURATION告诉播放器单个分片最大时长，#EXTINF后面跟的是该分片实际时长，下一行就是分片文件名。#EXT-X-ENDLIST表示这是点播流，分片已经全部就绪；如果没有这一行，说明是直播流，分片会持续追加，批量下载就得另做处理。分片地址可能是相对路径，也可能是完整 URL，解析时要拿索引文件自身的 URL 做基准去拼绝对地址，这一步漏了后面全盘拉不到。
批量任务拆解的逻辑是：先遍历所有 M3U8 链接，逐个下载索引文件并解析出分片列表，然后对每个视频的分片做并发下载，最后合并。这里有个关键决策——并发粒度放在哪一层。常见做法是外层视频串行、内层分片并发，因为同时下载几十个视频会把带宽打满，反而拖慢整体进度，而且合并阶段需要等单个视频的全部分片就位，串行外层能让合并尽早开始。
2.2 用 Python 解析索引并生成任务清单
import re import os from urllib.parse import urljoin def parse_m3u8(content, base_url): """ 解析 M3U8 文本，返回分片绝对地址列表和是否为点播流 base_url: 索引文件自身的 URL，用于拼接相对路径 """ lines = content.strip().splitlines() segments = [] is_vod = False for line in lines: line = line.strip() if line.startswith('#EXT-X-ENDLIST'): is_vod = True elif line and not line.startswith('#'): # 相对路径拼成绝对地址 segments.append(urljoin(base_url, line)) return segments, is_vod def build_task_list(m3u8_urls, output_dir): """ 为每个 M3U8 链接生成下载任务 返回 [{url, name, segments, is_vod}, ...] """ tasks = [] for idx, url in enumerate(m3u8_urls): # 用链接最后一段做文件名，去掉查询参数 name = url.split('/')[-1].split('?')[0].replace('.m3u8', '') if not name: name = f'video_{idx:04d}' tasks.append({ 'url': url, 'name': name, 'output': os.path.join(output_dir, f'{name}.mp4'), 'segments': [], 'is_vod': False }) return tasksparse_m3u8里用urljoin而不是字符串拼接，是因为分片路径可能是../seg/001.ts这种带相对层级的写法，手动拼容易出错。is_vod标志用来区分点播和直播，直播流没有#EXT-X-ENDLIST，分片列表不完整，直接下载会得到一个残缺文件，遇到这种情况要么跳过要么等直播结束再处理。build_task_list里对文件名做了去查询参数处理，因为很多 M3U8 链接后面挂着 token，直接当文件名会包含非法字符。
2.3 批量场景下的并发模型选择
单视频下载用requests逐片拉也行，但几百个视频加起来分片数可能上万，串行拉取耗时不可接受。常见做法是用concurrent.futures.ThreadPoolExecutor做分片级并发，线程数控制在 8 到 16 之间。为什么不是越高越好？因为大多数流媒体服务器对单 IP 的并发连接有限制，开到 32 以上容易触发 403 或连接重置，反而要花时间重试。我一般会先跑一个视频做探测，看服务器能承受多少并发不断流，再把这个值用到批量任务里。
另一个决策点是重试策略。分片下载失败是常态，尤其是长视频的后半段，网络抖动或服务端限速都会导致个别分片拉取失败。必须给每个分片配重试，重试次数设 3 次，每次间隔递增（1s、2s、4s），超过次数就把该分片标记为失败，整个视频标记为不完整，避免合并出一个静默损坏的文件。
3. 并发下载与合并落地：从分片到可播放 MP4
3.1 分片下载器的实现与参数说明
import requests import time from concurrent.futures import ThreadPoolExecutor, as_completed def download_segment(seg_url, save_path, retries=3, timeout=15): """ 下载单个分片，带重试 retries: 最大重试次数 timeout: 单次请求超时秒数 """ headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)', 'Referer': seg_url.rsplit('/', 1)[0] + '/' } for attempt in range(retries): try: resp = requests.get(seg_url, headers=headers, timeout=timeout, stream=True) if resp.status_code == 200: with open(save_path, 'wb') as f: for chunk in resp.iter_content(chunk_size=8192): f.write(chunk) return True # 403/404 不重试，直接失败 if resp.status_code in (403, 404): return False except requests.RequestException: pass time.sleep(2 ** attempt) # 指数退避 return False def download_video(task, max_workers=12): """ 下载单个视频的全部分片 max_workers: 分片并发数，建议 8-16 """ seg_dir = task['output'] + '_segments' os.makedirs(seg_dir, exist_ok=True) seg_paths = [os.path.join(seg_dir, f'{i:05d}.ts') for i in range(len(task['segments']))] failed = [] with ThreadPoolExecutor(max_workers=max_workers) as pool: future_map = { pool.submit(download_segment, url, path): i for i, (url, path) in enumerate(zip(task['segments'], seg_paths)) } for future in as_completed(future_map): idx = future_map[future] if not future.result(): failed.append(idx) return seg_paths, failedheaders里带Referer是血泪经验——不少流媒体服务器会校验来源页，不带 Referer 直接返回 403，而浏览器播放时自动带了，所以手动下载必须补上。stream=True配合iter_content是为了避免大分片一次性读进内存，虽然单个 ts 通常只有几百 KB 到几 MB，但批量场景下内存占用还是要控制。max_workers默认给 12，这个值在多数服务器上能跑满带宽又不触发限流，如果遇到大量 403 就往下调到 6 或 8。重试逻辑里对 403 和 404 直接返回失败不重试，因为这两类错误重试也不会变好，只会浪费时间。
3.2 分片合并：为什么不能直接 cat
最直觉的合并方式是cat *.ts > output.mp4，但这里有个坑：分片文件名如果按字符串排序，10.ts会排在2.ts前面，导致顺序错乱。所以合并前必须按数字索引排序，而不是按文件名默认排序。另一个坑是部分流的分片带独立的时间戳基准，直接拼接可能出现音画不同步，这种情况需要用 ffmpeg 的 concat 协议处理。
import subprocess def merge_segments(seg_paths, output_path): """ 用 ffmpeg concat 协议合并分片 先生成文件列表，再调用 ffmpeg """ list_file = output_path + '.txt' with open(list_file, 'w', encoding='utf-8') as f: for path in seg_paths: # ffmpeg concat 要求路径用单引号包裹，且转义单引号 safe = path.replace("'", "'\\''") f.write(f"file '{safe}'\n") cmd = [ 'ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', list_file, '-c', 'copy', output_path ] result = subprocess.run(cmd, capture_output=True, text=True) os.remove(list_file) return result.returncode == 0-c copy表示直接复制流不重新编码，速度极快，几分钟的视频几秒就能合并完。-safe 0是允许文件列表里出现绝对路径，不加这个参数 ffmpeg 会拒绝处理。用 ffmpeg concat 而不是直接二进制拼接的好处是它能处理分片间的时间戳衔接，减少音画不同步的概率。如果合并后发现播放器拖进度条卡顿，说明分片时间戳有断裂，这时候得去掉-c copy改成重新编码，但耗时会大幅增加，属于不得已的手段。
3.3 批量调度：把单视频流程串起来
def batch_download(m3u8_urls, output_dir, video_workers=1, seg_workers=12): """ 批量下载入口 video_workers: 同时下载的视频数，建议 1-2 seg_workers: 单视频内分片并发数 """ os.makedirs(output_dir, exist_ok=True) tasks = build_task_list(m3u8_urls, output_dir) report = [] for task in tasks: # 拉取索引 resp = requests.get(task['url'], timeout=20) segments, is_vod = parse_m3u8(resp.text, task['url']) if not is_vod: report.append((task['name'], 'skip', '直播流，跳过')) continue task['segments'] = segments seg_paths, failed = download_video(task, max_workers=seg_workers) if failed: report.append((task['name'], 'partial', f'{len(failed)} 个分片失败')) continue ok = merge_segments(seg_paths, task['output']) report.append((task['name'], 'ok' if ok else 'merge_fail', '')) return reportvideo_workers默认给 1，也就是视频之间串行。前面提过原因：并行下载多个视频会让分片请求交错，带宽被摊薄，单个视频的完成时间反而拉长，而且合并要等全部分片就位，串行能让每个视频尽快进入合并阶段。如果带宽充裕且视频都很短，可以调到 2，但再高就容易触发服务端限流。report里区分了skip、partial、ok、merge_fail四种状态，批量跑完先看报告，partial的视频需要单独重跑失败分片，merge_fail通常是 ffmpeg 没装或路径有空格。
4. 避坑与排查：批量下载最容易翻车的五个点
4.1 现象：所有分片都返回 403，但浏览器能播
原因：服务端校验了Referer或User-Agent，浏览器播放时自动带了这些头，脚本裸请求被拒。解决：在请求头里补上Referer为索引文件所在目录，User-Agent伪装成常见浏览器。如果补了还不行，检查是否有Cookie校验，从浏览器开发者工具里把请求头完整复制过来。
4.2 现象：合并后的 MP4 能播但时长不对，拖进度条跳回开头
原因：分片顺序错了，通常是文件名排序把10.ts排到了2.ts前面。解决：合并前按数字索引排序，不要依赖文件系统的默认排序。如果已经合并了，用ffprobe看时长，和索引里#EXTINF累加值对比，差得多就是顺序问题。
4.3 现象：批量跑到一半大量超时，重试也失败
原因：并发数开太高，触发了服务端的连接数限制或限速。解决：把seg_workers从 12 降到 6 甚至 4，重试间隔从指数退避改成固定 3 秒。如果还不行，加一个全局信号量控制总并发，确保同一时刻的活跃连接数不超过阈值。
4.4 现象：部分视频下载完提示partial，但重跑还是失败
原因：这些分片可能本身在服务端就是 404，或者索引里的地址已经过期（带 token 的链接有时效）。解决：重新拉取一次索引文件，看分片地址是否变化。如果地址没变但依然 404，说明源站分片缺失，这个视频无法完整下载，只能标记跳过。
4.5 现象：ffmpeg 合并报Unsafe file name或直接卡住
原因：文件列表里有绝对路径或特殊字符，-safe 0没加或者路径转义不对。解决：确保命令里带了-safe 0，文件列表里每行路径用单引号包裹，路径中的单引号做转义。如果路径含中文，确认 ffmpeg 版本支持 UTF-8，老版本可能需要把路径转成短路径或英文路径。
5. 进阶技巧：断点续传与完整性校验
批量下载最怕跑到 90% 断网，重头再来。我一般会给每个视频维护一个.done标记文件，记录已完成的分片索引，重跑时先读标记跳过已下载的分片。实现上很简单：每下载成功一个分片就往标记文件追加一行索引，启动时读进一个集合，download_video里判断索引在集合里就直接跳过。这样即使中断，重跑也只补缺失的部分。
def load_done(seg_dir): """读取已完成分片索引""" done_file = os.path.join(seg_dir, '.done') if not os.path.exists(done_file): return set() with open(done_file) as f: return {int(line.strip()) for line in f if line.strip().isdigit()} def mark_done(seg_dir, idx): """追加完成标记""" with open(os.path.join(seg_dir, '.done'), 'a') as f: f.write(f'{idx}\n')完整性校验分两层：第一层是分片数量，下载完的分片数必须等于索引里的分片数；第二层是合并后的时长，用ffprobe读出实际时长，和索引里#EXTINF累加值对比，误差超过 2 秒就说明有分片缺失或损坏。我习惯在批量跑完后统一跑一遍校验，把不通过的视频列出来单独处理，而不是边下边校验，那样会拖慢整体速度。
另一个实用技巧是给输出文件按内容哈希命名，避免重复下载。如果两个 M3U8 链接指向同一份内容，分片地址可能不同但最终视频一样，这时候用索引内容的哈希做文件名，第二次遇到直接跳过。这个做法在整理来源混乱的课程包时特别省事，能自动去重。
从那以后我每次跑批量任务前，都会先用一个视频做全流程探测——确认请求头、并发数、合并参数都对了，再放开跑全量。这个习惯帮我省下了无数次重跑的时间。希望帮到你。
本文还有配套的精品资源，点击获取
