简介一套基于Python的m3u8下载器源码面向需要将在线m3u8视频批量下载到本地的用户也适合Python爬虫与流媒体处理学习者参考。资源共3个文件以Python脚本和Markdown说明文档为主整体仅3KB体量轻、结构清晰便于逐行阅读和二次修改。其中包含两个下载脚本分别覆盖m3u8解析、ts分片请求、本地写入与合并等关键流程可满足视频点播或直播内容的离线备份需求说明文档则对运行环境、依赖安装和基本用法作了交代。目前已有62人学习下载适合希望快速获得可运行脚本或想研究m3u8下载原理的开发者。通过学习该资源能够掌握基于requests/urllib发送网络请求、解析ts链接、保存文件等基础实现思路在此基础上还可自行扩展并发下载、断点续传、ffmpeg合并等功能把这套代码改造成更完整的小工具适配不同站点的m3u8地址。1. 从拿到 m3u8 到落地成文件差的就是这个下载器浏览器里能播的视频并不代表你能把它存下来。大部分在线视频站点把完整媒体切成了几秒一段的 TS 分片再用一个 m3u8 索引文件去描述播放顺序播放器拿到索引后按顺序拉取分片看起来就是一个连续的视频。这种协议本身不复杂但真实场景里常常会遇到m3u8视频转换失败、分片 404、加密流、防盗链 403 这类问题。自己做下载器的好处是能把控制权拿回来知道哪些分片失败、为什么失败、怎么重试、怎么解密、怎么合并。这篇文章我会用一个 Python 实现的下载器为主线从 m3u8 解析讲起覆盖并发下载、AES-128 解密、TS 转 MP4 合并以及分片校验最终让你能拿着代码处理大多数流媒体下载场景。适合写过一点 Python、想彻底搞懂 m3u8 下载链路的人。2. m3u8 的索引结构先搞懂格式再写解析2.1 一个最小 m3u8 文件长什么样m3u8 本质上是 UTF-8 编码的文本文件每一行都是一个指令或者一个 URL。典型的 VOD点播文件比直播简单前者是固定分片列表后者是滑动窗口。写下载器时首先要把这两类区分开因为直播流的#EXT-X-ENDLIST可能永远不出现。#EXTM3U #EXT-X-VERSION:3 #EXT-X-TARGETDURATION:10 #EXT-X-MEDIA-SEQUENCE:0 #EXTINF:10.0, https://example.com/seg1.ts #EXTINF:10.0, https://example.com/seg2.ts #EXT-X-ENDLIST这个文件的关键信息有#EXTM3U表示这是 m3u8 格式#EXT-X-TARGETDURATION表示单分片最大时长用来预估延迟#EXTINF后面跟的是分片时长下一行是对应的分片 URL#EXT-X-ENDLIST表示点播流没有这个标签说明是直播流下载器要特别处理。2.2 相对路径与绝对路径的坑分片 URL 不一定是完整地址很多 m3u8 索引里写的是相对路径。比如 m3u8 地址是https://cdn.example.com/hls/video/index.m3u8而分片写成../ts/seg1.ts直接用这个相对路径去请求就会 404。这里需要的是标准 URL 解析而不是字符串拼接。from urllib.parse import urljoin base_url https://cdn.example.com/hls/video/index.m3u8 seg_url ../ts/seg1.ts full_url urljoin(base_url, seg_url) print(full_url) # 输出https://cdn.example.com/hls/ts/seg1.tsurljoin的行为逻辑是按照 URL 的目录层级来解析遇到..就向上回退一级最后得到的结果是浏览器实际请求时会用的完整地址。很多人用字符串base_url.rstrip(/) / seg_url来处理这在分片是../开头的相对路径时直接出错所以解析 m3u8 时统一用urljoin不要自己拼。2.3 处理嵌套 m3u8主索引与子索引大型点播视频的文件列表经常是分层的外层是主 m3u8里面列的不直接是 TS 分片而是多个子 m3u8 地址每个子索引对应不同码率或不同段落。判断方式很简单看#EXTINF下面的 URL 后缀如果是.m3u8就说明还有一层。def extract_segments(m3u8_text, base_uri): segments [] current_duration None for line in m3u8_text.splitlines(): line line.strip() if line.startswith(#EXTINF:): duration float(line.split(:)[1].split(,)[0]) current_duration duration elif line and not line.startswith(#): if line.endswith(.m3u8): # 命中子索引递归解析 sub_url urljoin(base_uri, line) sub_text fetch_text(sub_url) segments.extend(extract_segments(sub_text, sub_url)) else: segments.append({ duration: current_duration, url: urljoin(base_uri, line) }) return segments这段逻辑里current_duration必须在遇到#EXTINF时更新然后等到下一行非注释行来消费它。递归解析.m3u8时要传入子索引自己的 URL 作为新的base_uri否则子索引里的相对路径又会解析错。这里没有处理#EXT-X-KEY加密标签加密流的处理留到第 4 章。#EXT-X-KEY标签出现的位置可能在#EXTINF之前也可能在整个文件头部。无论哪种情况它影响的是后续所有分片直到新的#EXT-X-KEY出现。解析的时候需要按顺序维护当前生效的密钥信息而不是只找第一个。3. 线程池并发下载 TS 分片速度与稳定性的平衡3.1 单线程下载的问题在哪里单线程逐个下载分片在本地文件只有几十个分片时问题不大但一个 2 小时的电影可能切成 1800 个分片每个分片 200KB 到 1MB 不等。单线程下载 1800 个分片的耗时大约是累加每个请求的 RTT 加传输时间千兆宽带上依然可能花上 10 多分钟。引入并发时真正的瓶颈不是带宽而是线程切换和内存占用。常见做法是用concurrent.futures.ThreadPoolExecutor因为下载任务是 IO 密集型的等待响应的过程不占 CPU多线程可以把等待时间利用起来。要注意的是不要把线程数盲目调到 50 以上大部分 CDN 对单 IP 并发连接数有限制超了会直接返回 403 或者 429。3.2 带重试的分片下载函数设计import requests import time from concurrent.futures import ThreadPoolExecutor, as_completed HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://example.com/ } def download_segment(segment_info): url segment_info[url] max_retries 3 for attempt in range(max_retries): try: resp requests.get(url, headersHEADERS, timeout10) if resp.status_code 200: return segment_info[index], resp.content elif resp.status_code in (403, 429): time.sleep(1.5 * (attempt 1)) else: time.sleep(0.5) except requests.exceptions.RequestException: time.sleep(1.0 * (attempt 1)) return segment_info[index], None这里HEADERS里的Referer很关键。很多 CDN 会做防盗链校验请求 TS 分片时如果Referer和播放页不一致直接 403。User-Agent也要伪装成浏览器部分站点会拦截非浏览器 UA。重试策略是遇到网络异常时指数退避遇到 403 或 429 时等待更长时间因为这是服务端主动限流不能快速重试。3.3 按序写盘还是乱序写盘分片下载完成之后顺序可能已经乱了。不能每下载一个分片就往最终文件追加写入因为分片大小不一致乱序追加会导致最终视频花屏或无法播放。两种方案一是全部下载完成后按索引排序再合并二是预分配固定大小的文件用seek按索引定位写入。def download_all(segments, max_workers8, output_pathoutput.ts): total len(segments) downloaded {} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_index {} for idx, seg in enumerate(segments): seg[index] idx future executor.submit(download_segment, seg) future_to_index[future] idx for future in as_completed(future_to_index): idx, data future.result() if data is not None: downloaded[idx] data else: print(f分片 {idx} 下载失败) # 按索引排序写入 with open(output_path, wb) as f: for idx in range(total): if idx in downloaded: f.write(downloaded[idx])with open(output.ts, wb)按顺序追加写入内存里只保留分片数据写完即释放。但这种方式在分片多的时候内存峰值会很高因为as_completed完成后数据全部堆积在downloaded字典里。大量分片时更稳妥的做法是改用磁盘缓存每个分片落盘到临时目录最后按索引用文件流合并。3.4 临时目录与磁盘缓存import os import shutil def download_to_temp(segments, temp_dirtemp_segments, max_workers8): os.makedirs(temp_dir, exist_okTrue) tasks [] with ThreadPoolExecutor(max_workersmax_workers) as executor: for idx, seg in enumerate(segments): seg[index] idx tasks.append(executor.submit(save_segment_to_file, seg, temp_dir)) for future in as_completed(tasks): result future.result() if result is not None: idx, path result else: print(f分片 {idx} 最终失败) def save_segment_to_file(segment_info, temp_dir): idx, data download_segment(segment_info) if data is None: return idx, None tmp_path os.path.join(temp_dir, f{idx:05d}.ts) with open(tmp_path, wb) as f: f.write(data) return idx, tmp_path{idx:05d}的作用是把索引格式化为最少 5 位数字这样按文件名排序时顺序正确不会出现2.ts排在10.ts后面的问题。最后合并时用sorted(os.listdir(temp_dir))读取顺序是字典序恰好对应数字顺序。这组参数值得根据实际调整max_workers8适合大多数 CDNtimeout10针对弱网要适当加大max_retries3对部分不稳定存储源不够建议做成可配置项。4. 加密流的解密与合并AES-128 和 EXT-X-KEY4.1 识别加密方式METHOD 与 URI部分 m3u8 文件头部会多出这么几行#EXT-X-KEY:METHODAES-128,URIkey.key,IV0x7b2e1f9a3c8d4e6f这表示后续分片用 AES-128 加密密钥文件在key.keyIV 由标签显式指定。还有一种写法是不带IV字段这种情况默认从#EXT-X-MEDIA-SEQUENCE开始按分片序号推导 IV具体规则是用 16 字节大端表示的序列号。实际项目中绝大多数只会遇到METHODAES-128和METHODNONE两种后者等于不加密。有些站点的密钥文件本身还需要请求头才能访问比如key.key的 URL 和服务器的分片不在同一个域名下此时构造 requests 请求时要传和分片一样的Referer。密钥文件通常只有 16 字节直接读取。import re from Crypto.Cipher import AES # 需要 pycryptodome def parse_key_tag(m3u8_text, base_uri): key_match re.search(r#EXT-X-KEY:METHOD([^,]),URI([^])(?:,IV0x([0-9a-fA-F]))?, m3u8_text) if not key_match: return None method key_match.group(1) if method ! AES-128: return None key_url urljoin(base_uri, key_match.group(2)) iv_hex key_match.group(3) resp requests.get(key_url, headersHEADERS, timeout10) key resp.content # 16字节 iv None if iv_hex: iv bytes.fromhex(iv_hex) # 32位hex转16字节 return {key: key, iv: iv}正则里必须同时捕获METHOD、URI和可选的IV。bytes.fromhex(iv_hex)要求输入的 hex 是偶数长度IV0x...的0x前缀已经被正则排除所以这里直接转换没问题。4.2 逐分片解密CBC 模式的坑AES-128 在 HLS 协议里固定使用 CBC 模式每个分片独立加密但分片长度未必是 16 字节的倍数。标准做法是最后一个分片解密后要 PKCS7 去填充非最后一个分片解密后原样保留。from Crypto.Cipher import AES def decrypt_segment(data, key, iv): cipher AES.new(key, AES.MODE_CBC, iv) decrypted cipher.decrypt(data) # 手动去除 PKCS7 填充最后一个分片才做 padding_len decrypted[-1] if 1 padding_len 16: return decrypted[:-padding_len] return decrypted如果每个分片单独解密iv参数直接用标签里的值。但如果 m3u8 里没写IV则需要按规则生成iv (media_sequence segment_index).to_bytes(16, big)。这里segment_index不是列表绝对下标而是相对于#EXT-X-MEDIA-SEQUENCE的偏移写代码时先用全局计数器维护。4.3 合并转 MP4合并 TS 后调用 ffmpeg合并 TS 分片时可以二进制直接拼接TS 本身就是为拼接设计的容器格式。拼接完成后得到的是.ts文件大多数播放器能播但部分设备或剪辑软件不认。要转成 mp4最常见的是调用系统里的 ffmpeg 命令。ffmpeg -i output.ts -c copy -bsf:a aac_adtstoasc output.mp4这里-c copy是流复制不重新编码速度很快几秒内完成一个 2GB 文件。-bsf:a aac_adtstoasc是音频位流过滤器TS 容器里的 AAC 是 ADTS 封装MP4 需要的是 ASC 封装不加这个参数转出来的 mp4 在手机上经常没有声音。如果视频编码是 H.265需要加上-tag:v hvc1否则部分播放器不认。m3u8视频转换失败的场景多半就是漏了这个过滤器或者原始 TS 里音频是 MP3那么aac_adtstoasc反而会报错此时去掉-bsf:a参数直接转换即可。代码调用可以写成import subprocess def convert_ts_to_mp4(ts_path, mp4_path): cmd [ ffmpeg, -y, -i, ts_path, -c, copy, -bsf:a, aac_adtstoasc, mp4_path ] result subprocess.run(cmd, capture_outputTrue, textTrue) if result.returncode ! 0: raise RuntimeError(result.stderr[-1000:])capture_outputTrue会捕获标准输出和错误失败时把 stderr 的末尾 1000 字打出来多数情况能看到具体原因比如找不到某个流或编码不支持。4.4 加密流的完整下载流程加密流和普通流的下载流程差异只在写盘前。def process_encrypted_stream(segments, key_info, temp_dir): os.makedirs(temp_dir, exist_okTrue) media_sequence 0 # 从 m3u8 中解析出的 #EXT-X-MEDIA-SEQUENCE for idx, seg in enumerate(segments): raw download_segment(seg)[1] if raw is None: continue iv key_info[iv] if iv is None: iv (media_sequence idx).to_bytes(16, big) decrypted decrypt_segment(raw, key_info[key], iv) with open(os.path.join(temp_dir, f{idx:05d}.ts), wb) as f: f.write(decrypted)每个分片解密后立刻落盘避免解密后的数据积压在内存里。idx是全局分片下标media_sequence idx只有在分片列表连续时才正确。如果有的分片#EXTINF中间穿插了#EXT-X-DISCONTINUITY说明视频流有拼接片段直接按序号推 IV 会出错这种情况需要为每个分片记录它自己的media_sequence。5. 下载完成后的分片校验与修复不从零开始的断点续传5.1 分片大小校验和自定义校验合并完成后最终文件是不是完整的可以在合并前校验统计每个分片实际下载的字节数再和#EXTINF的时长做粗略估算或者直接依赖 HTTP 响应的Content-Length。更可靠的做法是解析 TS 分片头部的0x47同步字节或者校验文件大小是否和请求时的Content-Length一致。def validate_segment_file(path): size os.path.getsize(path) if size 0: return False with open(path, rb) as f: head f.read(4) if len(head) 4: return False # TS 分片第一个字节必须是 0x47 if head[0] ! 0x47: return False # 4字节为 0x47 G 传输错误标志(1bit) 起始标志(1bit) 优先级(1bit) PID(13bit) pid ((head[1] 0x1F) 8) | head[2] return pid ! 0x1FFF # 0x1FFF 是空包TS 分片第一字节恒为0x47这是 MPEG-TS 的同步字节。如果解密失败或者下载不完整头部通常不是0x47可以直接判定这次校验失败并重新下载该分片。5.2 断点续传记录已下载分片清单下载到一半中断重新跑一次全部下载是浪费时间。常见做法是维护一个downloaded.txt每成功写完一个分片就追加一行索引重新下载时跳过已有的。DONE_FILE os.path.join(temp_dir, download_progress.txt) def load_done_set(): if not os.path.exists(DONE_FILE): return set() with open(DONE_FILE, r) as f: return {int(line.strip()) for line in f if line.strip().isdigit()} def mark_done(idx): with open(DONE_FILE, a) as f: f.write(f{idx}\n) def download_with_resume(segments, temp_dir): done load_done_set() pending [seg for idx, seg in enumerate(segments) if idx not in done] for seg in pending: idx segments.index(seg) data download_segment(seg) if data is None: continue write_to_file(idx, data, temp_dir) mark_done(idx)downloaded.txt追加写入是原子操作不会因为程序崩溃损坏。断点续传的粒度是分片级不需要对文件内偏移做记录。5.3 用二分定位损坏分片合并后的 mp4 播放到某处花屏或者中断这时要快速定位是哪个分片坏了。常规做法是把每个已下载分片按顺序拼接后逐个尝试解码但成本太高。一个我觉得实用的技巧是采用二分定位先合并前半部分用 ffmpeg 探测是否能正常解析再决定坏分片在哪一半。ffmpeg -v error -i temp_segments/first_half.ts -f null - 21 | head -20-v error只输出错误级别日志-f null表示只解码不输出文件。如果有报错说明前半段里有坏分片继续二分如果没报错就查后半段。这个操作在本地走一遍很快错误日志行里会包含具体的分片文件名。正常情况下一轮二分能把 1000 个分片定位到 1 个的排查量从全量降到 10 次探测以内。定位后单独重新下载该分片替换临时目录里的旧文件再跑一次合并就行。这套流程比我见过的完整重试策略省时间也更可控它就是下载器是否「可维护」的直观体现。本文还有配套的精品资源点击获取 从零实现 Python m3u8 下载器：并发下载、AES-128 解密与 TS 转 MP4
简介一套基于Python的m3u8下载器源码面向需要将在线m3u8视频批量下载到本地的用户也适合Python爬虫与流媒体处理学习者参考。资源共3个文件以Python脚本和Markdown说明文档为主整体仅3KB体量轻、结构清晰便于逐行阅读和二次修改。其中包含两个下载脚本分别覆盖m3u8解析、ts分片请求、本地写入与合并等关键流程可满足视频点播或直播内容的离线备份需求说明文档则对运行环境、依赖安装和基本用法作了交代。目前已有62人学习下载适合希望快速获得可运行脚本或想研究m3u8下载原理的开发者。通过学习该资源能够掌握基于requests/urllib发送网络请求、解析ts链接、保存文件等基础实现思路在此基础上还可自行扩展并发下载、断点续传、ffmpeg合并等功能把这套代码改造成更完整的小工具适配不同站点的m3u8地址。1. 从拿到 m3u8 到落地成文件差的就是这个下载器浏览器里能播的视频并不代表你能把它存下来。大部分在线视频站点把完整媒体切成了几秒一段的 TS 分片再用一个 m3u8 索引文件去描述播放顺序播放器拿到索引后按顺序拉取分片看起来就是一个连续的视频。这种协议本身不复杂但真实场景里常常会遇到m3u8视频转换失败、分片 404、加密流、防盗链 403 这类问题。自己做下载器的好处是能把控制权拿回来知道哪些分片失败、为什么失败、怎么重试、怎么解密、怎么合并。这篇文章我会用一个 Python 实现的下载器为主线从 m3u8 解析讲起覆盖并发下载、AES-128 解密、TS 转 MP4 合并以及分片校验最终让你能拿着代码处理大多数流媒体下载场景。适合写过一点 Python、想彻底搞懂 m3u8 下载链路的人。2. m3u8 的索引结构先搞懂格式再写解析2.1 一个最小 m3u8 文件长什么样m3u8 本质上是 UTF-8 编码的文本文件每一行都是一个指令或者一个 URL。典型的 VOD点播文件比直播简单前者是固定分片列表后者是滑动窗口。写下载器时首先要把这两类区分开因为直播流的#EXT-X-ENDLIST可能永远不出现。#EXTM3U #EXT-X-VERSION:3 #EXT-X-TARGETDURATION:10 #EXT-X-MEDIA-SEQUENCE:0 #EXTINF:10.0, https://example.com/seg1.ts #EXTINF:10.0, https://example.com/seg2.ts #EXT-X-ENDLIST这个文件的关键信息有#EXTM3U表示这是 m3u8 格式#EXT-X-TARGETDURATION表示单分片最大时长用来预估延迟#EXTINF后面跟的是分片时长下一行是对应的分片 URL#EXT-X-ENDLIST表示点播流没有这个标签说明是直播流下载器要特别处理。2.2 相对路径与绝对路径的坑分片 URL 不一定是完整地址很多 m3u8 索引里写的是相对路径。比如 m3u8 地址是https://cdn.example.com/hls/video/index.m3u8而分片写成../ts/seg1.ts直接用这个相对路径去请求就会 404。这里需要的是标准 URL 解析而不是字符串拼接。from urllib.parse import urljoin base_url https://cdn.example.com/hls/video/index.m3u8 seg_url ../ts/seg1.ts full_url urljoin(base_url, seg_url) print(full_url) # 输出https://cdn.example.com/hls/ts/seg1.tsurljoin的行为逻辑是按照 URL 的目录层级来解析遇到..就向上回退一级最后得到的结果是浏览器实际请求时会用的完整地址。很多人用字符串base_url.rstrip(/) / seg_url来处理这在分片是../开头的相对路径时直接出错所以解析 m3u8 时统一用urljoin不要自己拼。2.3 处理嵌套 m3u8主索引与子索引大型点播视频的文件列表经常是分层的外层是主 m3u8里面列的不直接是 TS 分片而是多个子 m3u8 地址每个子索引对应不同码率或不同段落。判断方式很简单看#EXTINF下面的 URL 后缀如果是.m3u8就说明还有一层。def extract_segments(m3u8_text, base_uri): segments [] current_duration None for line in m3u8_text.splitlines(): line line.strip() if line.startswith(#EXTINF:): duration float(line.split(:)[1].split(,)[0]) current_duration duration elif line and not line.startswith(#): if line.endswith(.m3u8): # 命中子索引递归解析 sub_url urljoin(base_uri, line) sub_text fetch_text(sub_url) segments.extend(extract_segments(sub_text, sub_url)) else: segments.append({ duration: current_duration, url: urljoin(base_uri, line) }) return segments这段逻辑里current_duration必须在遇到#EXTINF时更新然后等到下一行非注释行来消费它。递归解析.m3u8时要传入子索引自己的 URL 作为新的base_uri否则子索引里的相对路径又会解析错。这里没有处理#EXT-X-KEY加密标签加密流的处理留到第 4 章。#EXT-X-KEY标签出现的位置可能在#EXTINF之前也可能在整个文件头部。无论哪种情况它影响的是后续所有分片直到新的#EXT-X-KEY出现。解析的时候需要按顺序维护当前生效的密钥信息而不是只找第一个。3. 线程池并发下载 TS 分片速度与稳定性的平衡3.1 单线程下载的问题在哪里单线程逐个下载分片在本地文件只有几十个分片时问题不大但一个 2 小时的电影可能切成 1800 个分片每个分片 200KB 到 1MB 不等。单线程下载 1800 个分片的耗时大约是累加每个请求的 RTT 加传输时间千兆宽带上依然可能花上 10 多分钟。引入并发时真正的瓶颈不是带宽而是线程切换和内存占用。常见做法是用concurrent.futures.ThreadPoolExecutor因为下载任务是 IO 密集型的等待响应的过程不占 CPU多线程可以把等待时间利用起来。要注意的是不要把线程数盲目调到 50 以上大部分 CDN 对单 IP 并发连接数有限制超了会直接返回 403 或者 429。3.2 带重试的分片下载函数设计import requests import time from concurrent.futures import ThreadPoolExecutor, as_completed HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://example.com/ } def download_segment(segment_info): url segment_info[url] max_retries 3 for attempt in range(max_retries): try: resp requests.get(url, headersHEADERS, timeout10) if resp.status_code 200: return segment_info[index], resp.content elif resp.status_code in (403, 429): time.sleep(1.5 * (attempt 1)) else: time.sleep(0.5) except requests.exceptions.RequestException: time.sleep(1.0 * (attempt 1)) return segment_info[index], None这里HEADERS里的Referer很关键。很多 CDN 会做防盗链校验请求 TS 分片时如果Referer和播放页不一致直接 403。User-Agent也要伪装成浏览器部分站点会拦截非浏览器 UA。重试策略是遇到网络异常时指数退避遇到 403 或 429 时等待更长时间因为这是服务端主动限流不能快速重试。3.3 按序写盘还是乱序写盘分片下载完成之后顺序可能已经乱了。不能每下载一个分片就往最终文件追加写入因为分片大小不一致乱序追加会导致最终视频花屏或无法播放。两种方案一是全部下载完成后按索引排序再合并二是预分配固定大小的文件用seek按索引定位写入。def download_all(segments, max_workers8, output_pathoutput.ts): total len(segments) downloaded {} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_index {} for idx, seg in enumerate(segments): seg[index] idx future executor.submit(download_segment, seg) future_to_index[future] idx for future in as_completed(future_to_index): idx, data future.result() if data is not None: downloaded[idx] data else: print(f分片 {idx} 下载失败) # 按索引排序写入 with open(output_path, wb) as f: for idx in range(total): if idx in downloaded: f.write(downloaded[idx])with open(output.ts, wb)按顺序追加写入内存里只保留分片数据写完即释放。但这种方式在分片多的时候内存峰值会很高因为as_completed完成后数据全部堆积在downloaded字典里。大量分片时更稳妥的做法是改用磁盘缓存每个分片落盘到临时目录最后按索引用文件流合并。3.4 临时目录与磁盘缓存import os import shutil def download_to_temp(segments, temp_dirtemp_segments, max_workers8): os.makedirs(temp_dir, exist_okTrue) tasks [] with ThreadPoolExecutor(max_workersmax_workers) as executor: for idx, seg in enumerate(segments): seg[index] idx tasks.append(executor.submit(save_segment_to_file, seg, temp_dir)) for future in as_completed(tasks): result future.result() if result is not None: idx, path result else: print(f分片 {idx} 最终失败) def save_segment_to_file(segment_info, temp_dir): idx, data download_segment(segment_info) if data is None: return idx, None tmp_path os.path.join(temp_dir, f{idx:05d}.ts) with open(tmp_path, wb) as f: f.write(data) return idx, tmp_path{idx:05d}的作用是把索引格式化为最少 5 位数字这样按文件名排序时顺序正确不会出现2.ts排在10.ts后面的问题。最后合并时用sorted(os.listdir(temp_dir))读取顺序是字典序恰好对应数字顺序。这组参数值得根据实际调整max_workers8适合大多数 CDNtimeout10针对弱网要适当加大max_retries3对部分不稳定存储源不够建议做成可配置项。4. 加密流的解密与合并AES-128 和 EXT-X-KEY4.1 识别加密方式METHOD 与 URI部分 m3u8 文件头部会多出这么几行#EXT-X-KEY:METHODAES-128,URIkey.key,IV0x7b2e1f9a3c8d4e6f这表示后续分片用 AES-128 加密密钥文件在key.keyIV 由标签显式指定。还有一种写法是不带IV字段这种情况默认从#EXT-X-MEDIA-SEQUENCE开始按分片序号推导 IV具体规则是用 16 字节大端表示的序列号。实际项目中绝大多数只会遇到METHODAES-128和METHODNONE两种后者等于不加密。有些站点的密钥文件本身还需要请求头才能访问比如key.key的 URL 和服务器的分片不在同一个域名下此时构造 requests 请求时要传和分片一样的Referer。密钥文件通常只有 16 字节直接读取。import re from Crypto.Cipher import AES # 需要 pycryptodome def parse_key_tag(m3u8_text, base_uri): key_match re.search(r#EXT-X-KEY:METHOD([^,]),URI([^])(?:,IV0x([0-9a-fA-F]))?, m3u8_text) if not key_match: return None method key_match.group(1) if method ! AES-128: return None key_url urljoin(base_uri, key_match.group(2)) iv_hex key_match.group(3) resp requests.get(key_url, headersHEADERS, timeout10) key resp.content # 16字节 iv None if iv_hex: iv bytes.fromhex(iv_hex) # 32位hex转16字节 return {key: key, iv: iv}正则里必须同时捕获METHOD、URI和可选的IV。bytes.fromhex(iv_hex)要求输入的 hex 是偶数长度IV0x...的0x前缀已经被正则排除所以这里直接转换没问题。4.2 逐分片解密CBC 模式的坑AES-128 在 HLS 协议里固定使用 CBC 模式每个分片独立加密但分片长度未必是 16 字节的倍数。标准做法是最后一个分片解密后要 PKCS7 去填充非最后一个分片解密后原样保留。from Crypto.Cipher import AES def decrypt_segment(data, key, iv): cipher AES.new(key, AES.MODE_CBC, iv) decrypted cipher.decrypt(data) # 手动去除 PKCS7 填充最后一个分片才做 padding_len decrypted[-1] if 1 padding_len 16: return decrypted[:-padding_len] return decrypted如果每个分片单独解密iv参数直接用标签里的值。但如果 m3u8 里没写IV则需要按规则生成iv (media_sequence segment_index).to_bytes(16, big)。这里segment_index不是列表绝对下标而是相对于#EXT-X-MEDIA-SEQUENCE的偏移写代码时先用全局计数器维护。4.3 合并转 MP4合并 TS 后调用 ffmpeg合并 TS 分片时可以二进制直接拼接TS 本身就是为拼接设计的容器格式。拼接完成后得到的是.ts文件大多数播放器能播但部分设备或剪辑软件不认。要转成 mp4最常见的是调用系统里的 ffmpeg 命令。ffmpeg -i output.ts -c copy -bsf:a aac_adtstoasc output.mp4这里-c copy是流复制不重新编码速度很快几秒内完成一个 2GB 文件。-bsf:a aac_adtstoasc是音频位流过滤器TS 容器里的 AAC 是 ADTS 封装MP4 需要的是 ASC 封装不加这个参数转出来的 mp4 在手机上经常没有声音。如果视频编码是 H.265需要加上-tag:v hvc1否则部分播放器不认。m3u8视频转换失败的场景多半就是漏了这个过滤器或者原始 TS 里音频是 MP3那么aac_adtstoasc反而会报错此时去掉-bsf:a参数直接转换即可。代码调用可以写成import subprocess def convert_ts_to_mp4(ts_path, mp4_path): cmd [ ffmpeg, -y, -i, ts_path, -c, copy, -bsf:a, aac_adtstoasc, mp4_path ] result subprocess.run(cmd, capture_outputTrue, textTrue) if result.returncode ! 0: raise RuntimeError(result.stderr[-1000:])capture_outputTrue会捕获标准输出和错误失败时把 stderr 的末尾 1000 字打出来多数情况能看到具体原因比如找不到某个流或编码不支持。4.4 加密流的完整下载流程加密流和普通流的下载流程差异只在写盘前。def process_encrypted_stream(segments, key_info, temp_dir): os.makedirs(temp_dir, exist_okTrue) media_sequence 0 # 从 m3u8 中解析出的 #EXT-X-MEDIA-SEQUENCE for idx, seg in enumerate(segments): raw download_segment(seg)[1] if raw is None: continue iv key_info[iv] if iv is None: iv (media_sequence idx).to_bytes(16, big) decrypted decrypt_segment(raw, key_info[key], iv) with open(os.path.join(temp_dir, f{idx:05d}.ts), wb) as f: f.write(decrypted)每个分片解密后立刻落盘避免解密后的数据积压在内存里。idx是全局分片下标media_sequence idx只有在分片列表连续时才正确。如果有的分片#EXTINF中间穿插了#EXT-X-DISCONTINUITY说明视频流有拼接片段直接按序号推 IV 会出错这种情况需要为每个分片记录它自己的media_sequence。5. 下载完成后的分片校验与修复不从零开始的断点续传5.1 分片大小校验和自定义校验合并完成后最终文件是不是完整的可以在合并前校验统计每个分片实际下载的字节数再和#EXTINF的时长做粗略估算或者直接依赖 HTTP 响应的Content-Length。更可靠的做法是解析 TS 分片头部的0x47同步字节或者校验文件大小是否和请求时的Content-Length一致。def validate_segment_file(path): size os.path.getsize(path) if size 0: return False with open(path, rb) as f: head f.read(4) if len(head) 4: return False # TS 分片第一个字节必须是 0x47 if head[0] ! 0x47: return False # 4字节为 0x47 G 传输错误标志(1bit) 起始标志(1bit) 优先级(1bit) PID(13bit) pid ((head[1] 0x1F) 8) | head[2] return pid ! 0x1FFF # 0x1FFF 是空包TS 分片第一字节恒为0x47这是 MPEG-TS 的同步字节。如果解密失败或者下载不完整头部通常不是0x47可以直接判定这次校验失败并重新下载该分片。5.2 断点续传记录已下载分片清单下载到一半中断重新跑一次全部下载是浪费时间。常见做法是维护一个downloaded.txt每成功写完一个分片就追加一行索引重新下载时跳过已有的。DONE_FILE os.path.join(temp_dir, download_progress.txt) def load_done_set(): if not os.path.exists(DONE_FILE): return set() with open(DONE_FILE, r) as f: return {int(line.strip()) for line in f if line.strip().isdigit()} def mark_done(idx): with open(DONE_FILE, a) as f: f.write(f{idx}\n) def download_with_resume(segments, temp_dir): done load_done_set() pending [seg for idx, seg in enumerate(segments) if idx not in done] for seg in pending: idx segments.index(seg) data download_segment(seg) if data is None: continue write_to_file(idx, data, temp_dir) mark_done(idx)downloaded.txt追加写入是原子操作不会因为程序崩溃损坏。断点续传的粒度是分片级不需要对文件内偏移做记录。5.3 用二分定位损坏分片合并后的 mp4 播放到某处花屏或者中断这时要快速定位是哪个分片坏了。常规做法是把每个已下载分片按顺序拼接后逐个尝试解码但成本太高。一个我觉得实用的技巧是采用二分定位先合并前半部分用 ffmpeg 探测是否能正常解析再决定坏分片在哪一半。ffmpeg -v error -i temp_segments/first_half.ts -f null - 21 | head -20-v error只输出错误级别日志-f null表示只解码不输出文件。如果有报错说明前半段里有坏分片继续二分如果没报错就查后半段。这个操作在本地走一遍很快错误日志行里会包含具体的分片文件名。正常情况下一轮二分能把 1000 个分片定位到 1 个的排查量从全量降到 10 次探测以内。定位后单独重新下载该分片替换临时目录里的旧文件再跑一次合并就行。这套流程比我见过的完整重试策略省时间也更可控它就是下载器是否「可维护」的直观体现。本文还有配套的精品资源点击获取