在实际项目中有时需要从在线视频网站获取视频资源用于学习、分析或备份。很多视频网站采用 HLSHTTP Live Streaming协议通过 m3u8 索引文件分片传输视频内容。直接下载这类视频会遇到分片合并、解密、反爬虫限制等问题。本文面向有一定 Python 基础但尚未系统掌握 m3u8 视频爬取技术的开发者。我们将从 m3u8 协议原理讲起逐步完成环境准备、依赖配置、爬虫编写、分片下载、解密合并的全流程并重点解决过程中可能出现的网络超时、密钥解析失败、合并后音画不同步等典型问题。最终你将掌握一套可复用的技术方案能够针对多数采用 m3u8 的网站实现稳定下载。1. 理解 m3u8 协议与 HLS 流媒体工作机制1.1 HLS 协议的基本原理HLS 是苹果公司提出的基于 HTTP 的流媒体传输协议。它将整个视频流切分成多个小文件通常是 .ts 格式的分片通过一个索引文件.m3u8来组织这些分片的下载顺序。播放器按顺序下载并播放这些分片实现边下边播。m3u8 文件本质是一个文本格式的播放列表里面记录了分片地址、解密密钥、分片时长等信息。例如一个典型的 m3u8 文件内容如下#EXTM3U #EXT-X-VERSION:3 #EXT-X-TARGETDURATION:10 #EXT-X-MEDIA-SEQUENCE:0 #EXT-X-KEY:METHODAES-128,URIhttps://example.com/key.key,IV0x1234567890abcdef1234567890abcdef #EXTINF:10.0, https://example.com/segment0.ts #EXTINF:10.0, https://example.com/segment1.ts #EXTINF:10.0, https://example.com/segment2.ts #EXT-X-ENDLIST关键标签说明#EXTM3U文件头标识这是一个 m3u8 文件#EXT-X-TARGETDURATION每个分片的最大时长#EXT-X-KEY分片加密信息包含密钥地址和初始化向量#EXTINF分片时长和分片文件地址#EXT-X-ENDLIST播放列表结束标志1.2 m3u8 视频下载的技术挑战与传统直接下载 mp4 文件不同m3u8 视频下载面临几个技术难点分片数量多一个小时的视频可能包含数百个分片需要批量下载加密处理很多商业网站会对分片进行 AES-128 加密需要获取并应用解密密钥反爬虫机制网站可能通过 User-Agent 检查、IP 限制、Referer 验证等方式阻止爬取合并复杂度下载的分片需要按正确顺序合并并处理可能的音视频同步问题理解这些难点后我们就能有针对性地设计解决方案。2. 环境准备与依赖配置2.1 Python 环境要求本项目需要 Python 3.7 及以上版本。建议使用虚拟环境隔离项目依赖# 创建虚拟环境 python -m venv video_downloader # 激活虚拟环境Windows video_downloader\Scripts\activate # 激活虚拟环境Linux/Mac source video_downloader/bin/activate2.2 核心依赖库安装我们需要以下几个关键库pip install requests m3u8 pycryptodome ffmpeg-python各库的作用requests用于发送 HTTP 请求下载 m3u8 文件和分片m3u8专业解析 m3u8 文件的库能自动处理各种标签pycryptodome提供 AES 解密功能用于解密加密的分片ffmpeg-python合并分片和格式转换的工具接口注意如果遇到 pycryptodome 安装问题可以尝试安装 pycryptodomexpip install pycryptodomex使用时导入Crypto改为Cryptodome。2.3 FFmpeg 工具安装FFmpeg 是处理音视频合并的核心工具需要单独安装Windows 系统访问 https://ffmpeg.org/download.html 下载 Windows 版本解压到合适目录如C:\ffmpeg将bin目录如C:\ffmpeg\bin添加到系统 PATH 环境变量Linux 系统Ubuntu/Debiansudo apt update sudo apt install ffmpeg验证安装ffmpeg -version正确安装后会显示 FFmpeg 版本信息。3. 构建 m3u8 视频下载器核心组件3.1 项目结构设计创建以下文件结构video_downloader/ ├── m3u8_downloader.py # 主下载器类 ├── decryptor.py # 解密处理模块 ├── merger.py # 合并处理模块 ├── utils.py # 工具函数 └── main.py # 主程序入口3.2 m3u8 文件解析器实现首先实现 m3u8 文件解析功能在m3u8_downloader.py中import requests import m3u8 from urllib.parse import urljoin import os class M3U8Downloader: def __init__(self, headersNone): self.headers headers or { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Referer: https://www.example.com/ } self.session requests.Session() self.session.headers.update(self.headers) def parse_m3u8(self, m3u8_url): 解析 m3u8 文件提取分片信息和密钥 try: response self.session.get(m3u8_url, timeout10) response.raise_for_status() # 使用 m3u8 库解析内容 m3u8_obj m3u8.loads(response.text, urim3u8_url) # 构建完整的分片 URL 列表 base_url m3u8_url.rsplit(/, 1)[0] / if / in m3u8_url else segments [] for segment in m3u8_obj.segments: segment_url segment.uri if not segment_url.startswith((http://, https://)): segment_url urljoin(base_url, segment_url) segments.append({ url: segment_url, duration: segment.duration, key: segment.key }) return { segments: segments, target_duration: m3u8_obj.target_duration, is_encrypted: m3u8_obj.keys and m3u8_obj.keys[0] is not None, key_info: m3u8_obj.keys[0] if m3u8_obj.keys else None } except requests.RequestException as e: print(f下载 m3u8 文件失败: {e}) return None except Exception as e: print(f解析 m3u8 文件失败: {e}) return None3.3 分片下载器实现在同一个文件中继续添加下载功能class M3U8Downloader: # ... 前面的代码 ... def download_segment(self, segment_info, index, download_dir, retry_count3): 下载单个分片 segment_path os.path.join(download_dir, fsegment_{index:05d}.ts) for attempt in range(retry_count): try: response self.session.get(segment_info[url], timeout30) response.raise_for_status() with open(segment_path, wb) as f: f.write(response.content) print(f下载完成: segment_{index:05d}.ts) return segment_path except requests.RequestException as e: print(f下载分片 {index} 失败 (尝试 {attempt 1}/{retry_count}): {e}) if attempt retry_count - 1: return None def download_all_segments(self, m3u8_info, download_dir): 下载所有分片 if not os.path.exists(download_dir): os.makedirs(download_dir) segments m3u8_info[segments] downloaded_files [] for i, segment in enumerate(segments): file_path self.download_segment(segment, i, download_dir) if file_path: downloaded_files.append(file_path) else: print(f分片 {i} 下载失败跳过) return downloaded_files3.4 解密处理模块创建decryptor.py处理加密分片from Crypto.Cipher import AES import os class TSDecryptor: def __init__(self): self.key_cache {} def download_key(self, key_url, session): 下载解密密钥 if key_url in self.key_cache: return self.key_cache[key_url] try: response session.get(key_url, timeout10) response.raise_for_status() key response.content self.key_cache[key_url] key return key except Exception as e: print(f下载密钥失败: {e}) return None def decrypt_segment(self, segment_path, key, ivNone): 解密单个分片 try: with open(segment_path, rb) as f: encrypted_data f.read() # 处理 IV初始化向量 if iv is None: cipher AES.new(key, AES.MODE_CBC, ivbytes(16)) else: if isinstance(iv, str) and iv.startswith(0x): iv bytes.fromhex(iv[2:]) elif isinstance(iv, bytes): pass else: iv iv.to_bytes(16, big) if hasattr(iv, to_bytes) else bytes(16) cipher AES.new(key, AES.MODE_CBC, iviv) # AES 解密要求数据长度是 16 的倍数需要处理填充 decrypted_data cipher.decrypt(encrypted_data) # 写入解密后的文件 decrypted_path segment_path.replace(.ts, _decrypted.ts) with open(decrypted_path, wb) as f: f.write(decrypted_data) return decrypted_path except Exception as e: print(f解密分片失败 {segment_path}: {e}) return None def decrypt_all_segments(self, segments_info, downloaded_files, session): 解密所有分片 if not segments_info[is_encrypted]: return downloaded_files # 未加密直接返回 key_info segments_info[key_info] if not key_info: print(视频已加密但未找到密钥信息) return downloaded_files # 下载密钥 key self.download_key(key_info.uri, session) if not key: print(无法获取解密密钥) return downloaded_files decrypted_files [] for i, file_path in enumerate(downloaded_files): # 获取该分片对应的 IV iv None if key_info.iv: iv key_info.iv elif i 0 and hasattr(key_info, iv) and key_info.iv: iv key_info.iv decrypted_path self.decrypt_segment(file_path, key, iv) if decrypted_path: decrypted_files.append(decrypted_path) # 删除原始加密文件 os.remove(file_path) else: decrypted_files.append(file_path) return decrypted_files4. 分片合并与格式转换4.1 使用 FFmpeg 合并分片创建merger.py处理合并逻辑import subprocess import os class TSFileMerger: def __init__(self, ffmpeg_pathffmpeg): self.ffmpeg_path ffmpeg_path def merge_with_ffmpeg(self, file_list, output_path): 使用 FFmpeg 合并分片文件 try: # 创建临时文件列表 list_file file_list.txt with open(list_file, w, encodingutf-8) as f: for file_path in file_list: f.write(ffile {os.path.abspath(file_path)}\n) # 使用 FFmpeg concat 协议合并 cmd [ self.ffmpeg_path, -f, concat, -safe, 0, -i, list_file, -c, copy, # 直接流复制不重新编码 -y, # 覆盖输出文件 output_path ] result subprocess.run(cmd, capture_outputTrue, textTrue, timeout300) # 清理临时文件 if os.path.exists(list_file): os.remove(list_file) if result.returncode 0: print(f合并完成: {output_path}) return True else: print(fFFmpeg 合并失败: {result.stderr}) return False except subprocess.TimeoutExpired: print(合并操作超时) return False except Exception as e: print(f合并过程出错: {e}) return False def simple_concat(self, file_list, output_path): 简单二进制合并适用于未加密的 TS 文件 try: with open(output_path, wb) as outfile: for file_path in file_list: with open(file_path, rb) as infile: outfile.write(infile.read()) print(f简单合并完成: {output_path}) return True except Exception as e: print(f简单合并失败: {e}) return False4.2 工具函数模块创建utils.py包含辅助功能import re import os def validate_url(url): 验证 URL 格式 pattern re.compile( r^(?:http|ftp)s?:// # http:// or https:// r(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)| # domain... rlocalhost| # localhost... r\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) # ...or ip r(?::\d)? # optional port r(?:/?|[/?]\S)$, re.IGNORECASE) return re.match(pattern, url) is not None def clean_filename(filename): 清理文件名中的非法字符 invalid_chars :/\\|?* for char in invalid_chars: filename filename.replace(char, _) return filename def get_video_duration(ffmpeg_path, video_path): 获取视频时长 try: cmd [ffmpeg_path, -i, video_path] result subprocess.run(cmd, capture_outputTrue, textTrue) # 从输出中解析时长信息 duration_match re.search(rDuration: (\d):(\d):(\d\.\d), result.stderr) if duration_match: hours int(duration_match.group(1)) minutes int(duration_match.group(2)) seconds float(duration_match.group(3)) total_seconds hours * 3600 minutes * 60 seconds return total_seconds return None except: return None def cleanup_temp_files(directory, keep_finalTrue): 清理临时文件 for file in os.listdir(directory): if file.endswith(.ts) and (decrypted in file or segment in file): os.remove(os.path.join(directory, file)) print(临时文件清理完成)5. 完整流程集成与测试5.1 主程序实现创建main.py整合所有功能import os import sys from m3u8_downloader import M3U8Downloader from decryptor import TSDecryptor from merger import TSFileMerger from utils import validate_url, clean_filename, cleanup_temp_files def main(): if len(sys.argv) 2: print(用法: python main.py m3u8_url [output_directory]) sys.exit(1) m3u8_url sys.argv[1] output_dir sys.argv[2] if len(sys.argv) 2 else downloads if not validate_url(m3u8_url): print(错误的 URL 格式) sys.exit(1) # 创建下载器实例 downloader M3U8Downloader() decryptor TSDecryptor() merger TSFileMerger() # 步骤1: 解析 m3u8 文件 print(步骤1: 解析 m3u8 文件...) m3u8_info downloader.parse_m3u8(m3u8_url) if not m3u8_info: print(解析 m3u8 文件失败) return print(f发现 {len(m3u8_info[segments])} 个分片) if m3u8_info[is_encrypted]: print(视频已加密需要解密处理) # 步骤2: 下载所有分片 print(步骤2: 下载分片...) downloaded_files downloader.download_all_segments(m3u8_info, output_dir) if not downloaded_files: print(分片下载失败) return print(f成功下载 {len(downloaded_files)} 个分片) # 步骤3: 解密处理如果需要 if m3u8_info[is_encrypted]: print(步骤3: 解密分片...) downloaded_files decryptor.decrypt_all_segments(m3u8_info, downloaded_files, downloader.session) # 步骤4: 合并分片 print(步骤4: 合并分片...) output_filename clean_filename(m3u8_url.split(/)[-1].replace(.m3u8, )) or output output_path os.path.join(output_dir, f{output_filename}.mp4) # 按文件名排序确保正确顺序 downloaded_files.sort() if merger.merge_with_ffmpeg(downloaded_files, output_path): print(视频下载合并完成!) # 清理临时文件 cleanup_temp_files(output_dir) else: print(合并失败尝试简单合并...) output_path_ts os.path.join(output_dir, f{output_filename}.ts) if merger.simple_concat(downloaded_files, output_path_ts): print(fTS 格式合并完成: {output_path_ts}) if __name__ __main__: main()5.2 测试运行准备一个测试用的 m3u8 URL请使用合法的测试资源# test.py from main import main import sys # 模拟命令行参数 sys.argv [main.py, https://example.com/test.m3u8, test_download] main()运行测试python test.py6. 常见问题排查与解决方案6.1 网络请求相关问题问题1: 403 Forbidden 错误现象下载 m3u8 文件或分片时返回 403 状态码原因网站检测到爬虫行为通过 User-Agent 或 Referer 验证解决方案# 更新请求头模拟真实浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://原视频网站域名/, Accept: */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Origin: https://原视频网站域名 } downloader M3U8Downloader(headersheaders)问题2: 连接超时现象请求长时间无响应或超时原因网络不稳定或服务器限制解决方案增加超时时间和重试机制# 在下载方法中增加超时控制 response self.session.get(url, timeout(10, 30)) # 连接超时10秒读取超时30秒6.2 解密相关问题问题3: 密钥下载失败现象无法下载解密密钥提示 404 或 403原因密钥 URL 需要特定 Referer 或 Cookie解决方案确保密钥请求使用与分片相同的会话# 在 decryptor.py 中确保使用相同的 session key self.download_key(key_info.uri, session) # 传入相同的 session 对象问题4: 解密后视频无法播放现象解密合并后的视频无法正常播放原因IV初始化向量处理错误或填充问题解决方案检查 IV 格式和处理方式# 确保 IV 正确处理 if isinstance(iv, str) and len(iv) 2 and iv.startswith(0x): iv_bytes bytes.fromhex(iv[2:].zfill(32)) # 确保16字节 else: iv_bytes iv.to_bytes(16, big) if hasattr(iv, to_bytes) else bytes(16)6.3 合并相关问题问题5: 合并后音画不同步现象播放时音频和视频不同步原因分片下载顺序错误或某些分片损坏解决方案确保分片按正确顺序合并检查每个分片完整性# 在合并前验证分片顺序 downloaded_files.sort(keylambda x: int(.join(filter(str.isdigit, x))))问题6: FFmpeg 合并失败现象FFmpeg 报错或合并过程中断原因分片格式不一致或损坏解决方案尝试简单合并或逐个分片检查# 备用合并方案 if not merger.merge_with_ffmpeg(files, output_path): # 尝试 TS 格式简单合并 merger.simple_concat(files, output_path.replace(.mp4, .ts))7. 生产环境最佳实践7.1 性能优化建议并发下载使用多线程下载分片提高速度from concurrent.futures import ThreadPoolExecutor, as_completed def concurrent_download_segments(self, m3u8_info, download_dir, max_workers5): 并发下载分片 with ThreadPoolExecutor(max_workersmax_workers) as executor: futures { executor.submit(self.download_segment, segment, i, download_dir): i for i, segment in enumerate(m3u8_info[segments]) } downloaded_files [] for future in as_completed(futures): result future.result() if result: downloaded_files.append(result) return sorted(downloaded_files)断点续传记录下载进度支持中断后继续def load_progress(self, progress_file): 加载下载进度 if os.path.exists(progress_file): with open(progress_file, r) as f: return set(f.read().splitlines()) return set() def save_progress(self, progress_file, downloaded_segments): 保存下载进度 with open(progress_file, w) as f: for segment in downloaded_segments: f.write(segment \n)7.2 错误处理与日志记录完善日志系统import logging def setup_logging(): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(video_downloader.log), logging.StreamHandler() ] )异常处理策略try: # 主要下载逻辑 result self.download_segment(segment_info, index, download_dir) if not result: logging.error(f分片 {index} 下载失败) # 根据业务需求决定是重试、跳过还是终止 except Exception as e: logging.exception(f下载过程出现未预期错误: {e}) # 记录详细错误信息便于排查7.3 法律与合规注意事项重要在使用爬虫技术时必须遵守相关法律法规和网站的使用条款。尊重 robots.txt检查目标网站的 robots.txt 文件控制访问频率避免对服务器造成过大压力仅用于合法用途确保下载内容用于个人学习或已获授权版权意识不传播、不商用有版权保护的内容实际项目中建议添加速率限制import time from functools import wraps def rate_limit(seconds): 速率限制装饰器 def decorator(func): wraps(func) def wrapper(*args, **kwargs): time.sleep(seconds) return func(*args, **kwargs) return wrapper return decorator # 使用示例 rate_limit(1) # 每秒最多1次请求 def download_segment(self, segment_info, index, download_dir): # 下载逻辑这套方案提供了从 m3u8 视频解析到下载合并的完整流程涵盖了加密处理、错误恢复、性能优化等生产级需求。在实际应用中还需要根据具体网站的反爬机制进行相应的调整和优化。