资讯动态

M3U8批量下载实战:从索引解析到ffmpeg合并的完整流水线

发布时间:2026/10/20 12:18:04 来源:尧图企业网站定制
简介M3U8批量视频流下载工具面向需要离线保存网页视频的普通用户与运维人员解决m3u8流媒体地址解析困难、逐条下载效率低的问题。压缩包共13个文件约23.6MB以exe可执行程序、json配置、log日志、conf参数文件及dat数据文件为主其中主程序与ffmpeg、aria2c等组件配合可完成地址解析、真实链接获取、多任务下载与下载后转码日志与配置文件则便于调整并发、缓存等参数并排查异常。目前已有3481人学习下载说明该工具在同类场景中具备一定实用参考价值。通过这份资源读者可获得一套可直接运行的绿色下载方案理解m3u8解析与下载链路的基本构成并借助内置转码能力将下载结果转为可本地播放的格式适合需要批量归档在线视频、研究流媒体下载流程的读者参考使用。1. 批量下载 M3U8从手动点播到脚本流水线的效率跃迁如果你手头攒了几十个甚至上百个 M3U8 链接还在浏览器里一个个点开、等它加载、再右键另存那基本等于把时间扔进碎纸机。M3U8 批量视频流下载要解决的就是这个场景把分散的索引文件、TS 分片、密钥获取和合并转码串成一条可重复执行的流水线让机器替你跑完重复劳动。它适合手里有合法授权内容、需要归档或离线回看的运维、测试和内容整理人员也适合想搞懂 HLS 协议落地细节的开发者。核心不是“下载”这个动作而是把索引解析、并发调度、失败重试和格式封装这几件事拆清楚后面无论换什么工具你都能自己搭出稳定流程。2. 拆解 M3U8 与 TS 分片先看懂索引再动手2.1 M3U8 文件里到底写了什么M3U8 本质是 UTF-8 编码的播放列表常见标签有#EXTM3U、#EXT-X-VERSION、#EXT-X-TARGETDURATION、#EXTINF、#EXT-X-KEY和#EXT-X-ENDLIST。主播放列表master playlist里放的是多码率变体每个变体指向一个子播放列表子播放列表里才是真正的 TS 分片地址。批量下载的第一步不是写代码而是用文本编辑器打开一个 M3U8确认它属于哪种结构。如果只有#EXTINF和.ts路径那就是媒体播放列表可以直接进入分片下载如果出现#EXT-X-STREAM-INF说明需要先选码率再拿子列表地址。常见做法是先用curl或requests把 M3U8 内容拉下来解析出所有分片 URL再统一调度。这里有个容易翻车的点分片地址可能是相对路径必须基于 M3U8 的最终 URL 做拼接不能只看域名。比如索引在https://example.com/hls/1080p/index.m3u8分片写的是seg-1.ts真实地址就是https://example.com/hls/1080p/seg-1.ts。用urljoin处理最稳妥。import requests from urllib.parse import urljoin def parse_m3u8(m3u8_url, headersNone): resp requests.get(m3u8_url, headersheaders, timeout15) resp.raise_for_status() lines resp.text.splitlines() segments [] key_info None for line in lines: line line.strip() if line.startswith(#EXT-X-KEY): # 提取加密方式和密钥地址常见为 AES-128 key_info line elif line and not line.startswith(#): # 相对路径转绝对路径 segments.append(urljoin(m3u8_url, line)) return segments, key_info这段代码只做两件事拉取索引文本、把非注释行转成绝对 URL。headers参数用于携带 Referer 或 User-Agent很多站点会校验这两个头缺失时返回 403。timeout建议设 10 到 20 秒太短容易误判网络抖动太长会拖慢批量任务。key_info先原样保留后面解密时再解析。2.2 分片下载的并发模型与参数选择拿到分片列表后串行下载 500 个 TS 文件可能要十几分钟并发能把时间压到一两分钟。但并发不是越高越好线程数超过目标服务器限制会触发 429 或直接封 IP。我一般先用 4 到 8 个线程试水观察响应状态和耗时再决定是否上调。Python 里用concurrent.futures.ThreadPoolExecutor就够不需要上异步框架因为瓶颈通常在网络 IO 和服务器限速不在本地 CPU。import os from concurrent.futures import ThreadPoolExecutor, as_completed def download_segment(url, save_dir, index, headersNone, retries3): os.makedirs(save_dir, exist_okTrue) filepath os.path.join(save_dir, f{index:05d}.ts) if os.path.exists(filepath) and os.path.getsize(filepath) 0: return filepath # 断点续传已存在且非空则跳过 for attempt in range(retries): try: r requests.get(url, headersheaders, timeout20, streamTrue) r.raise_for_status() with open(filepath, wb) as f: for chunk in r.iter_content(chunk_size8192): f.write(chunk) return filepath except Exception as e: if attempt retries - 1: raise RuntimeError(f分片 {index} 下载失败: {e}) return None def batch_download(segments, save_dir, max_workers6, headersNone): results [None] * len(segments) with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map { executor.submit(download_segment, url, save_dir, i, headers): i for i, url in enumerate(segments) } for future in as_completed(future_map): idx future_map[future] results[idx] future.result() return resultsmax_workers控制并发上限默认 6 是保守值。retries3给每个分片三次机会失败后抛异常而不是静默跳过避免最后合并出残缺视频。streamTrue配合iter_content防止大分片一次性读进内存。断点续传逻辑用文件是否存在且大小大于零来判断简单但有效适合批量任务中途中断后重跑。2.3 解密、合并与转封装如果 M3U8 里出现#EXT-X-KEY:METHODAES-128,URIkey.key说明分片被 AES-128 加密。你需要先下载密钥文件再对每个 TS 分片做解密。密钥通常是 16 字节二进制用requests.get拿content即可。解密用cryptography库的 Cipher 或pycryptodome的 AES模式是 CBCIV 可能写在#EXT-X-KEY的IV属性里也可能是分片序号。没有 IV 时常见做法是用分片序号作为 IV但不同封装工具行为不一致需要实际验证。合并阶段最简单的方式是二进制拼接把所有 TS 按顺序写入一个文件再交给 ffmpeg 转封装成 MP4。直接拼接得到的.ts文件多数播放器能放但拖动进度条可能不流畅转成 MP4 后兼容性更好。# 先按序号拼接 TS 分片 cat $(ls -v seg_*.ts) merged.ts # 用 ffmpeg 转封装为 MP4-c copy 表示不重新编码 ffmpeg -i merged.ts -c copy -bsf:a aac_adtstoasc output.mp4-c copy避免重新编码速度快且不损失画质。-bsf:a aac_adtstoasc用于修正 AAC 音频在 TS 和 MP4 容器间的格式差异不加这个参数有时会出现音频无声或杂音。如果源分片是 H.265部分旧设备不支持需要重新编码那就去掉-c copy改用-c:v libx264但耗时会显著增加。3. 批量任务编排从单链接到链接池3.1 链接池的读取与去重批量下载的核心输入是一个链接列表。常见来源是文本文件、CSV 或数据库导出。我一般用纯文本一行一个 M3U8 URL方便用sort -u去重。读取时过滤空行和以#开头的注释行避免把说明文字当链接请求。def load_urls(filepath): urls [] with open(filepath, r, encodingutf-8) as f: for line in f: line line.strip() if line and not line.startswith(#): urls.append(line) # 保持顺序去重 seen set() unique [] for u in urls: if u not in seen: seen.add(u) unique.append(u) return unique去重逻辑用set记录已见 URL同时保留原始顺序。如果链接量很大比如上万条可以考虑用sqlite做持久化队列避免内存占用过高。但对大多数批量下载场景几千条以内用列表足够。3.2 每个链接独立目录与命名规范批量任务最容易乱的地方是文件覆盖。不同 M3U8 的分片都叫seg-1.ts如果全放在一个目录后下载的会覆盖前面的。正确做法是为每个链接生成独立子目录目录名可以用 URL 的哈希或时间戳加序号。我习惯用hashlib.md5(url.encode()).hexdigest()[:8]作为目录后缀既避免非法字符又保证唯一性。import hashlib def make_task_dir(base_dir, url): tag hashlib.md5(url.encode()).hexdigest()[:8] task_dir os.path.join(base_dir, ftask_{tag}) os.makedirs(task_dir, exist_okTrue) return task_dir最终输出文件名可以保留原始标题或序号但目录必须隔离。如果链接列表里有重复 URL去重后只处理一次。任务目录里再分segments/和output/两个子目录分片和成品分开存放排查问题时一目了然。3.3 失败重试与日志记录批量任务跑几个小时很正常中途某个链接失败不能影响整体。每个链接独立 try-except失败后记录到failed.log包含 URL、错误类型和时间戳。重试策略分两层分片级别重试 3 次链接级别重试 1 次。链接级别重试前先清理该任务目录下的空文件避免断点续传逻辑误判。import logging import time logging.basicConfig( filenamebatch_download.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) def process_one(url, base_dir, headersNone): task_dir make_task_dir(base_dir, url) seg_dir os.path.join(task_dir, segments) try: segments, key_info parse_m3u8(url, headers) if not segments: raise ValueError(未解析到分片) batch_download(segments, seg_dir, max_workers6, headersheaders) # 合并与转码步骤略按 2.3 执行 logging.info(f成功: {url}) return True except Exception as e: logging.error(f失败: {url} | {e}) return False def run_batch(url_file, base_dir, headersNone): urls load_urls(url_file) for i, url in enumerate(urls, 1): print(f[{i}/{len(urls)}] 处理: {url}) ok process_one(url, base_dir, headers) if not ok: time.sleep(2) # 失败后稍作停顿避免连续触发限流日志用标准logging模块输出到文件同时保留控制台进度。time.sleep(2)在失败后暂停两秒给服务器和本地网络一个缓冲。如果连续多个链接失败可能是 IP 被限或请求头失效需要人工介入检查。4. 避坑与排查批量下载中最容易翻车的五个点4.1 现象分片全部 403但浏览器能播原因目标服务器校验Referer或User-Agent脚本请求头缺失或与浏览器不一致。解决用浏览器开发者工具抓一个实际分片请求把Referer、User-Agent、Origin原样复制到脚本 headers 里。注意Referer有时是 M3U8 所在页面地址不是 M3U8 本身。4.2 现象合并后视频时长对但画面花屏或音画不同步原因分片下载顺序错乱或部分分片下载不完整但文件大小大于零断点续传逻辑误判为已完成。解决下载时用临时文件名写完再重命名合并前校验每个分片大小是否接近预期异常分片强制重新下载。另外确认cat拼接时按数字序号排序ls -v比默认ls可靠。4.3 现象AES-128 解密后仍然无法播放原因IV 取值错误。部分封装工具在#EXT-X-KEY里不写IV默认用分片序号作为 IV但序号从 0 还是 1 开始因工具而异。解决先用单个分片做实验分别用序号 0 和 1 作为 IV 解密用 ffprobe 检查哪个能正常识别。确认后再批量处理。4.4 现象并发调到 16 后大量超时降到 4 就正常原因目标服务器对单 IP 并发连接数有限制或本地出口带宽被占满。解决并发数从 4 开始逐步上调观察成功率。同时给每个请求加随机延迟比如time.sleep(random.uniform(0.1, 0.3))模拟人工节奏。如果服务器返回 429立即降低并发并增加重试间隔。4.5 现象ffmpeg 转 MP4 报 “Non-monotonous DTS” 警告原因TS 分片的时间戳不连续常见于直播录制或分片来自不同编码会话。解决转封装时加-fflags genpts让 ffmpeg 重新生成时间戳或直接重新编码。如果只是警告且输出能播可以忽略如果音画不同步必须重新编码。5. 进阶技巧用 ffmpeg 直接拉流与校验成品5.1 让 ffmpeg 替你完成解析和合并如果不想自己写解析和合并逻辑ffmpeg 可以直接读取 M3U8 并输出 MP4。命令简洁适合单个或少量链接。批量场景下用 Python 调用 ffmpeg 子进程比自己拼分片更省心但可控性略低。ffmpeg -headers Referer: https://example.com/page \ -user_agent Mozilla/5.0 \ -i https://example.com/hls/1080p/index.m3u8 \ -c copy -bsf:a aac_adtstoasc \ output.mp4-headers和-user_agent用于绕过基础校验。-c copy不重新编码。如果 M3U8 是加密的ffmpeg 需要能访问密钥地址必要时把密钥下载到本地并修改 M3U8 里的 URI 为本地路径。这个方法的边界是遇到需要复杂鉴权或动态密钥的源ffmpeg 可能直接报错还是得回到脚本方案。5.2 成品校验别等归档后才发现文件损坏下载完成后用ffprobe检查每个 MP4 的时长、码率和流信息。批量校验可以写成循环把异常文件输出到报告里。for f in output/*.mp4; do duration$(ffprobe -v error -show_entries formatduration -of defaultnoprint_wrappers1:nokey1 $f) if [ -z $duration ] || [ $duration N/A ]; then echo 损坏: $f else echo 正常: $f | 时长: ${duration}s fi doneffprobe返回空或N/A说明容器无法解析文件大概率损坏。时长明显短于预期说明分片缺失。我一般会把校验结果和原始链接列表做交叉比对确保每个链接都有对应成品。从那以后我每次批量任务跑完都强制走一遍 ffprobe 校验宁可多花几分钟也不想到用的时候才发现文件打不开。希望帮到你。本文还有配套的精品资源点击获取

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑