1. 理解m3u8视频流的基本原理m3u8是一种基于HTTP Live StreamingHLS协议的视频播放列表格式它把整个视频分割成多个小的ts文件片段。这种设计最初是苹果公司提出的目的是为了适应不同网络条件下的视频流畅播放。在实际应用中你会发现很多视频网站都采用这种技术来传输视频内容。当你打开一个m3u8文件时通常会看到类似这样的内容#EXTM3U #EXT-X-VERSION:3 #EXT-X-TARGETDURATION:10 #EXT-X-MEDIA-SEQUENCE:0 #EXT-X-KEY:METHODAES-128,URIkey.key,IV0x00000000000000000000000000000000 #EXTINF:10.000000, segment0.ts #EXTINF:10.000000, segment1.ts这个文件结构其实很好理解#EXTM3U是文件头标识#EXT-X-KEY行表示视频是否加密以及加密方式#EXTINF行表示每个ts片段的时长最后是实际的ts文件路径我遇到过不少开发者第一次接触m3u8时都会困惑为什么要把视频切成这么多小文件其实这样做有几个明显优势适应不同网络环境可以根据带宽动态切换不同质量的视频流支持边下边播不需要等待整个视频下载完成便于CDN分发提高视频加载速度2. 搭建Python爬虫开发环境在开始编写爬虫之前我们需要准备好开发环境。这里我推荐使用Python 3.7版本因为后续我们要用到的异步库在这个版本上表现最好。首先安装必要的依赖库pip install requests pycryptodome fake-useragent aiohttp让我解释下这些库的作用requests最常用的HTTP请求库pycryptodomeAES解密需要的加密库比pycrypto更稳定fake-useragent生成随机User-Agent避免被反爬aiohttp异步HTTP客户端用于高性能下载我建议使用虚拟环境来管理这些依赖这样可以避免污染全局Python环境。创建虚拟环境的命令python -m venv m3u8_env source m3u8_env/bin/activate # Linux/Mac m3u8_env\Scripts\activate # Windows在实际项目中我经常遇到的一个坑是加密库的安装问题。如果你在Windows上遇到Microsoft Visual C 14.0 is required的错误可以直接安装编译好的wheel文件pip install pycryptodomex3. 解析m3u8文件结构解析m3u8文件是整个下载过程的第一步也是最重要的一步。我们需要从中提取出所有ts片段的URL以及加密信息。下面是一个完整的解析函数实现def parse_m3u8(m3u8_url): response requests.get(m3u8_url) if response.status_code ! 200: raise Exception(fFailed to fetch m3u8: {response.status_code}) content response.text.split(\n) ts_list [] key_info {} for line in content: line line.strip() if not line or line.startswith(#): if #EXT-X-KEY in line: # 处理加密信息 parts line.split(,) method parts[0].split()[1] key_info[method] method.strip() for part in parts[1:]: if URI in part: key_info[uri] part.split()[1].strip() elif IV in part: key_info[iv] part.split()[1] continue # 处理相对路径 if not line.startswith(http): line urljoin(m3u8_url, line) ts_list.append(line) return ts_list, key_info这个函数做了几件重要的事情下载并读取m3u8文件内容解析出所有ts文件的URL提取加密信息如果有处理相对路径问题在实际应用中我发现有几个常见问题需要注意有些网站的m3u8文件是嵌套的里面可能包含多级播放列表ts文件的路径可能是相对路径需要用urljoin处理加密信息可能出现在文件的不同位置4. 处理AES加密视频流当m3u8文件中包含#EXT-X-KEY标签时说明视频流是加密的通常采用AES-128 CBC模式加密。我们需要先获取密钥才能解密视频内容。解密过程主要分为三个步骤4.1 获取加密密钥密钥通常是一个16字节的二进制文件我们需要从指定的URI下载def fetch_key(key_url, refererNone): headers {Referer: referer} if referer else {} response requests.get(key_url, headersheaders) if response.status_code ! 200: raise Exception(fFailed to fetch key: {response.status_code}) return response.content4.2 初始化解密器使用pycryptodome库创建AES解密器from Crypto.Cipher import AES def create_decryptor(key, ivNone): if not iv: iv b0000000000000000 # 默认IV elif iv.startswith(0x): iv bytes.fromhex(iv[2:]) return AES.new(key, AES.MODE_CBC, iv)4.3 解密ts片段获取到解密器后我们就可以解密下载的ts文件了def decrypt_ts(encrypted_data, decryptor): return decryptor.decrypt(encrypted_data)在实际项目中我遇到过几种常见的加密情况简单的AES-128加密使用固定IV动态密钥密钥URI每次请求都会变化需要特定请求头才能获取密钥针对这些情况我们需要灵活调整代码。比如对于动态密钥的情况可能需要在每次下载ts片段前都重新获取密钥。5. 实现高效下载策略直接顺序下载所有ts文件效率很低我们可以采用多线程或异步IO来加速下载过程。5.1 多线程下载实现使用ThreadPoolExecutor实现多线程下载from concurrent.futures import ThreadPoolExecutor def download_ts_files(ts_urls, decryptorNone, max_workers10): def download_single(url): data requests.get(url).content if decryptor: data decryptor.decrypt(data) return data with ThreadPoolExecutor(max_workersmax_workers) as executor: futures [executor.submit(download_single, url) for url in ts_urls] return [f.result() for f in futures]5.2 异步IO下载实现对于更高性能的需求可以使用aiohttp实现异步下载import aiohttp import asyncio async def async_download(ts_urls, decryptorNone): async with aiohttp.ClientSession() as session: tasks [download_single(session, url, decryptor) for url in ts_urls] return await asyncio.gather(*tasks) async def download_single(session, url, decryptor): async with session.get(url) as response: data await response.read() if decryptor: data decryptor.decrypt(data) return data在实际测试中我发现异步IO的性能通常比多线程更好特别是在高并发场景下。但要注意服务器可能会限制并发请求数太高的并发反而会导致失败率上升。6. 合并ts文件与错误处理下载完所有ts片段后我们需要将它们合并成一个完整的视频文件。6.1 简单的文件合并最基本的合并方法就是按顺序将ts文件写入目标文件def merge_ts_files(ts_data_list, output_file): with open(output_file, wb) as f: for data in ts_data_list: f.write(data)6.2 处理下载失败的情况在实际应用中总会有一些ts片段下载失败。我们可以实现自动重试机制def download_with_retry(url, max_retries3, decryptorNone): for i in range(max_retries): try: data requests.get(url, timeout10).content if decryptor: data decryptor.decrypt(data) return data except Exception as e: print(fDownload failed (attempt {i1}): {e}) continue raise Exception(fFailed to download {url} after {max_retries} retries)6.3 校验文件完整性合并前最好检查下所有ts片段是否完整def validate_ts_files(ts_data_list): for i, data in enumerate(ts_data_list): if not data or len(data) 1024: # 假设每个ts至少1KB print(fWarning: ts file {i} seems corrupted) return False return True我在实际项目中总结了几点经验合并前最好按文件名排序确保顺序正确可以添加进度条显示合并进度对于大型视频可以考虑使用内存映射文件提高合并效率7. 完整代码实现与封装现在我们把所有功能整合成一个完整的类方便复用import os import requests from urllib.parse import urljoin from concurrent.futures import ThreadPoolExecutor from Crypto.Cipher import AES class M3U8Downloader: def __init__(self, max_workers10): self.max_workers max_workers self.session requests.Session() def parse_m3u8(self, m3u8_url): # 解析逻辑同上 pass def download(self, m3u8_url, output_file): ts_urls, key_info self.parse_m3u8(m3u8_url) decryptor None if key_info.get(method) AES-128: key self._fetch_key(key_info[uri]) iv key_info.get(iv) decryptor self._create_decryptor(key, iv) with ThreadPoolExecutor(self.max_workers) as executor: ts_data list(executor.map( lambda url: self._download_ts(url, decryptor), ts_urls )) self._merge_files(ts_data, output_file) def _fetch_key(self, key_url): # 密钥获取逻辑 pass def _create_decryptor(self, key, iv): # 创建解密器逻辑 pass def _download_ts(self, url, decryptor): # 下载单个ts文件逻辑 pass def _merge_files(self, ts_data, output_file): # 文件合并逻辑 pass使用这个类非常简单downloader M3U8Downloader(max_workers20) downloader.download(http://example.com/playlist.m3u8, output.mp4)8. 高级技巧与优化建议在实际项目中我还总结了一些高级技巧8.1 处理动态变化的m3u8有些直播流的m3u8会不断更新我们需要定期检查并下载新的ts片段def download_live_stream(m3u8_url, output_dir, duration60): start_time time.time() downloaded set() while time.time() - start_time duration: ts_urls, _ parse_m3u8(m3u8_url) new_urls [u for u in ts_urls if u not in downloaded] if new_urls: download_ts_files(new_urls) downloaded.update(new_urls) time.sleep(5) # 每5秒检查一次更新8.2 添加代理支持为了避免IP被封我们可以添加代理支持def download_with_proxy(url, proxyNone): proxies {http: proxy, https: proxy} if proxy else None return requests.get(url, proxiesproxies).content8.3 断点续传实现对于大视频文件可以实现断点续传功能def resume_download(m3u8_url, output_file, temp_dirtemp): if not os.path.exists(temp_dir): os.makedirs(temp_dir) # 检查已下载的文件 downloaded set(f.split(.)[0] for f in os.listdir(temp_dir)) ts_urls, key_info parse_m3u8(m3u8_url) remaining [u for i, u in enumerate(ts_urls) if str(i) not in downloaded] # 继续下载剩余文件 download_ts_files(remaining, temp_dir) # 合并所有文件 merge_ts_files(temp_dir, output_file)8.4 性能优化技巧使用连接池减少TCP连接开销适当调整并发数找到最佳性能点对ts文件进行预排序避免合并时的排序开销使用内存缓冲区减少磁盘IO9. 常见问题与解决方案在开发过程中我遇到过不少问题这里分享几个典型的9.1 解密后视频无法播放可能原因IV值不正确 - 确保使用正确的IV有些网站会使用非常规IV密钥错误 - 检查密钥URL是否正确有时需要添加Referer头加密模式不匹配 - 确认是AES-128 CBC模式9.2 下载速度慢解决方案增加并发数使用更快的DNS服务器尝试不同的CDN节点检查是否有带宽限制9.3 部分ts文件下载失败处理方法实现自动重试机制记录失败的URL最后统一重试检查是否被服务器限制9.4 合并后的视频有问题排查步骤检查ts文件顺序是否正确确认没有缺失的ts片段尝试用FFmpeg手动合并测试10. 安全与法律注意事项在开发和使用这类工具时有几个重要的法律和道德问题需要考虑仅下载你有权访问的内容尊重网站的robots.txt规定不要对目标服务器造成过大负担注意用户隐私保护技术本身是中性的但使用方式决定了它的性质。我建议大家在开发这类工具时添加适当的延迟避免对服务器造成冲击遵守网站的使用条款不要绕过明显的访问限制在实际项目中我通常会添加这样的限制# 限制下载速度 time.sleep(0.5) # 每个请求间隔0.5秒 # 尊重robots.txt from urllib.robotparser import RobotFileParser rp RobotFileParser() rp.set_url(https://example.com/robots.txt) rp.read() if not rp.can_fetch(*, m3u8_url): raise Exception(Not allowed by robots.txt)