别再手动扒视频了！用Python解析m3u8文件，5分钟批量获取所有.ts片段下载地址
别再手动扒视频了！用Python解析m3u8文件，5分钟批量获取所有.ts片段下载地址
每次手动复制粘贴.ts地址时，我都觉得自己像个流水线上的工人——机械、低效且容易出错。直到发现Python可以自动化这个过程，才意识到原来技术真的能解放生产力。本文将带你从零构建一个健壮的m3u8解析器，不仅能处理普通列表，还能应对多码率、加密流等复杂场景，最终生成可直接用于批量下载的地址清单。
1. 认识m3u8：不只是文本文件
m3u8本质是HLS（HTTP Live Streaming）协议的核心播放列表，其结构远比表面看到的复杂。一个典型的未加密文件如下：
#EXTM3U #EXT-X-VERSION:3 #EXT-X-TARGETDURATION:10 #EXTINF:9.009, http://example.com/segment1.ts #EXTINF:9.009, http://example.com/segment2.ts关键标签解析：
| 标签 | 作用 | 是否必需 |
|---|---|---|
| #EXTM3U | 文件类型声明 | 是 |
| #EXT-X-VERSION | 协议版本兼容性 | 否 |
| #EXT-X-TARGETDURATION | 最大分片时长（秒） | 是 |
| #EXTINF | 分片时长和URL | 是 |
遇到多码率自适应流时，会出现嵌套结构：
#EXTM3U #EXT-X-STREAM-INF:BANDWIDTH=1500000,RESOLUTION=720x480 video_medium.m3u8 #EXT-X-STREAM-INF:BANDWIDTH=800000,RESOLUTION=640x360 video_low.m3u8提示：实际开发中约30%的m3u8会采用这种多级结构，需要递归解析
2. 基础解析器搭建：从正则到状态机
先用最直接的字符串处理方式实现基础功能：
import re def naive_parser(m3u8_text): ts_urls = [] for line in m3u8_text.split('\n'): line = line.strip() if line and not line.startswith('#') and line.endswith('.ts'): ts_urls.append(line) return ts_urls这种方法简单但脆弱，无法处理：
- 带查询参数的URL（如
segment.ts?token=xxx） - 相对路径转绝对路径
- 加密流媒体（
#EXT-X-KEY）
更健壮的实现应采用状态机模式：
class M3U8Parser: def __init__(self, base_url): self.base_url = base_url self.ts_segments = [] self.current_key = None def parse(self, content): for line in content.split('\n'): line = line.strip() if not line: continue if line.startswith('#EXT-X-KEY'): self._parse_key(line) elif not line.startswith('#'): self._add_segment(line) def _parse_key(self, line): # 处理加密逻辑 pass def _add_segment(self, uri): if not uri.startswith('http'): uri = self.base_url + uri self.ts_segments.append(uri)3. 工程化增强：异常处理与性能优化
真实环境需要考虑的边界情况：
- 网络请求封装
def fetch_m3u8(url, max_retries=3): session = requests.Session() session.mount('https://', HTTPAdapter(max_retries=max_retries)) try: resp = session.get(url, timeout=10) resp.raise_for_status() return resp.text except requests.exceptions.SSLError: # 处理证书错误 return fetch_m3u8(url, verify=False) except Exception as e: print(f"Failed to fetch {url}: {str(e)}") return None- 相对路径处理算法
from urllib.parse import urljoin def resolve_url(base, relative): """ base: m3u8文件URL relative: 分片相对路径 """ if relative.startswith('http'): return relative base_dir = base[:base.rfind('/')+1] return urljoin(base_dir, relative)- **多码率自适应处理流程
- 解析主m3u8中的
#EXT-X-STREAM-INF - 选择最高码率版本（或根据网络条件动态选择）
- 递归获取最终分片列表
- 解析主m3u8中的
4. 实战：构建完整的下载管道
完整工作流示例：
def batch_download(m3u8_url, output_dir): # 1. 获取主列表 main_playlist = fetch_m3u8(m3u8_url) if not main_playlist: return False # 2. 判断是否多码率 if '#EXT-X-STREAM-INF' in main_playlist: playlist_url = select_variant(main_playlist, m3u8_url) variant_content = fetch_m3u8(playlist_url) else: variant_content = main_playlist # 3. 解析分片 parser = M3U8Parser(playlist_url) parser.parse(variant_content) # 4. 并行下载 with ThreadPoolExecutor(max_workers=8) as executor: futures = [] for idx, ts_url in enumerate(parser.ts_segments): save_path = f"{output_dir}/segment_{idx:04d}.ts" futures.append(executor.submit(download_ts, ts_url, save_path)) for future in as_completed(futures): future.result() # 触发异常传播注意：实际生产环境应添加速率限制、断点续传等功能
5. 高级技巧：解密与合并
遇到加密流时，需要先获取解密密钥：
def _parse_key(self, line): """解析类似：#EXT-X-KEY:METHOD=AES-128,URI="key.key" """ match = re.search(r'METHOD=([^,]+),URI="([^"]+)"', line) if match: self.current_key = { 'method': match.group(1), 'uri': resolve_url(self.base_url, match.group(2)) }下载完成后用PyCryptodome解密：
from Crypto.Cipher import AES def decrypt_ts(encrypted_data, key, iv): cipher = AES.new(key, AES.MODE_CBC, iv=iv) return cipher.decrypt(encrypted_data)最后用FFmpeg合并（需提前安装）：
ffmpeg -f concat -safe 0 -i file_list.txt -c copy output.mp4其中file_list.txt格式：
file 'segment_0001.ts' file 'segment_0002.ts'