Python爬虫实战：m3u8视频下载与合并完整技术指南

在线视频网站爬虫实战：m3u8视频下载与合并完整指南

在日常开发和学习中，我们经常需要从在线视频网站获取视频资源用于分析研究。面对网站采用m3u8流媒体协议的情况，传统下载方式往往无法直接获取完整视频。本文将完整介绍基于Python的在线视频爬虫开发，重点解决m3u8视频的下载、解密和合并问题。

1. m3u8流媒体技术解析

1.1 什么是m3u8格式

m3u8是HTTP Live Streaming（HLS）协议使用的播放列表格式，基于文本文件存储媒体片段信息。与传统的MP4等单一文件格式不同，m3u8将视频分割成多个小的ts（Transport Stream）文件，通过索引文件组织播放顺序。

m3u8文件结构示例：

#EXTM3U #EXT-X-VERSION:3 #EXT-X-TARGETDURATION:10 #EXT-X-MEDIA-SEQUENCE:0 #EXTINF:10.0, segment0.ts #EXTINF:10.0, segment1.ts #EXTINF:10.0, segment2.ts #EXT-X-ENDLIST

这种分段设计的优势在于支持自适应码率、快速seek和容错恢复，但给完整视频下载带来了挑战。

1.2 m3u8与MP4格式对比

理解这些差异有助于我们设计合适的下载策略。m3u8下载的核心在于解析索引文件、批量下载ts片段，最后合并成完整视频。

2. 开发环境准备

2.1 Python环境配置

本项目需要Python 3.7及以上版本，推荐使用Anaconda或Miniconda管理环境。

创建专用环境：

conda create -n video_spider python=3.8 conda activate video_spider

2.2 必要依赖库安装

核心依赖包：

pip install requests beautifulsoup4 pycryptodome m3u8

各库作用说明：

• requests：网络请求库，用于获取网页内容和视频片段
• beautifulsoup4：HTML解析库，用于提取视频链接
• pycryptodome：加解密库，用于处理加密的ts片段
• m3u8：专业的m3u8解析库，简化索引文件处理

2.3 FFmpeg工具安装

FFmpeg是视频处理的核心工具，用于ts片段的合并和格式转换。

Windows系统安装：

1. 访问FFmpeg官网下载Windows版本
2. 解压到指定目录，如C:\ffmpeg
3. 将bin目录添加到系统PATH环境变量
4. 验证安装：ffmpeg -version

Linux系统安装：

# Ubuntu/Debian sudo apt update sudo apt install ffmpeg # CentOS/RHEL sudo yum install epel-release sudo yum install ffmpeg

验证FFmpeg安装：

ffmpeg -version

正常输出版本信息即表示安装成功。

3. 网页视频链接抓取技术

3.1 分析视频网站结构

在开始编码前，需要先了解目标网站的视频加载机制。以典型视频网站为例，视频链接通常通过以下方式加载：

1. 直接链接：页面中直接包含视频URL
2. 异步加载：通过Ajax请求获取视频信息
3. 动态渲染：JavaScript动态生成视频播放器

使用浏览器开发者工具（F12）的Network面板可以监控视频请求过程，找到真实的m3u8链接。

3.2 使用Requests库获取页面内容

import requests from bs4 import BeautifulSoup import re class VideoSpider: def __init__(self): self.session = requests.Session() # 设置请求头，模拟浏览器访问 self.headers = { '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', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', } def get_page_content(self, url): """获取网页内容""" try: response = self.session.get(url, headers=self.headers, timeout=10) response.encoding = 'utf-8' if response.status_code == 200: return response.text else: print(f"请求失败，状态码：{response.status_code}") return None except Exception as e: print(f"网络请求异常：{e}") return None

3.3 提取m3u8链接的正则表达式

不同网站的m3u8链接格式各异，需要针对性地编写提取规则：

def extract_m3u8_urls(self, html_content): """从HTML内容中提取m3u8链接""" m3u8_patterns = [ r'https?://[^\s<>"]+?\.m3u8(?:\?[^\s<>"]*)?', r'https?://[^\s<>"]+?/index\.m3u8', r'src:\s*[\'"](https?://[^\s<>"]+?\.m3u8)[\'"]', ] m3u8_urls = [] for pattern in m3u8_patterns: matches = re.findall(pattern, html_content, re.IGNORECASE) m3u8_urls.extend(matches) # 去重并返回 return list(set(m3u8_urls)) def extract_video_title(self, html_content): """提取视频标题""" soup = BeautifulSoup(html_content, 'html.parser') # 尝试多种标题选择器 title_selectors = [ 'h1', '.video-title', '.title', '[class*="title"]', 'meta[property="og:title"]', 'title' ] for selector in title_selectors: elements = soup.select(selector) if elements: title = elements[0].get_text().strip() if title and len(title) > 3: return self.sanitize_filename(title) return "unknown_video"

4. m3u8文件解析与下载

4.1 使用m3u8库解析索引文件

import m3u8 import os from urllib.parse import urljoin, urlparse class M3U8Downloader: def __init__(self, output_dir="downloads"): self.output_dir = output_dir self.session = requests.Session() os.makedirs(output_dir, exist_ok=True) def parse_m3u8(self, m3u8_url): """解析m3u8文件内容""" try: response = self.session.get(m3u8_url, timeout=10) if response.status_code == 200: # 解析m3u8内容 m3u8_content = m3u8.loads(response.text, uri=m3u8_url) return m3u8_content else: print(f"获取m3u8文件失败: {response.status_code}") return None except Exception as e: print(f"解析m3u8文件异常: {e}") return None def get_ts_segments(self, m3u8_content): """获取所有ts片段信息""" segments = [] base_url = m3u8_content.base_uri for segment in m3u8_content.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

4.2 处理加密的ts片段

很多网站会对ts片段进行AES加密，需要解密后才能正常播放：

from Crypto.Cipher import AES import base64 class TSDecryptor: def __init__(self): self.cache_keys = {} def download_key(self, key_info): """下载解密密钥""" if key_info is None: return None key_url = key_info.uri if not key_url.startswith(('http://', 'https://')): key_url = urljoin(key_info.base_uri, key_url) if key_url in self.cache_keys: return self.cache_keys[key_url] try: response = requests.get(key_url, timeout=10) if response.status_code == 200: key = response.content self.cache_keys[key_url] = key return key except Exception as e: print(f"下载密钥失败: {e}") return None def decrypt_ts_data(self, ts_data, key, iv=None): """解密ts数据""" if key is None: return ts_data try: if iv is None: # 使用默认IV iv = bytes(16) # 16字节的0 elif isinstance(iv, str) and iv.startswith('0x'): # 处理16进制格式的IV iv = bytes.fromhex(iv[2:]) cipher = AES.new(key, AES.MODE_CBC, iv) decrypted_data = cipher.decrypt(ts_data) return decrypted_data except Exception as e: print(f"解密失败: {e}") return ts_data

4.3 多线程下载ts片段

为了提高下载效率，使用多线程并行下载ts片段：

import threading from concurrent.futures import ThreadPoolExecutor, as_completed class ParallelDownloader: def __init__(self, max_workers=5): self.max_workers = max_workers self.downloaded_count = 0 self.lock = threading.Lock() def download_segment(self, segment_info, index, total_segments, decryptor): """下载单个ts片段""" segment_url = segment_info['url'] key_info = segment_info.get('key') try: response = requests.get(segment_url, timeout=30) if response.status_code == 200: ts_data = response.content # 处理加密片段 if key_info: key = decryptor.download_key(key_info) iv = getattr(key_info, 'iv', None) ts_data = decryptor.decrypt_ts_data(ts_data, key, iv) # 更新进度 with self.lock: self.downloaded_count += 1 progress = (self.downloaded_count / total_segments) * 100 print(f"\r下载进度: {self.downloaded_count}/{total_segments} ({progress:.1f}%)", end='') return index, ts_data else: print(f"\n下载片段 {index} 失败: {response.status_code}") return index, None except Exception as e: print(f"\n下载片段 {index} 异常: {e}") return index, None def download_all_segments(self, segments, decryptor, output_dir): """并行下载所有ts片段""" total_segments = len(segments) self.downloaded_count = 0 segment_files = [] with ThreadPoolExecutor(max_workers=self.max_workers) as executor: # 提交所有下载任务 future_to_index = { executor.submit(self.download_segment, segment, i, total_segments, decryptor): i for i, segment in enumerate(segments) } # 收集下载结果 for future in as_completed(future_to_index): index, ts_data = future.result() if ts_data: # 保存ts片段 filename = os.path.join(output_dir, f"segment_{index:05d}.ts") with open(filename, 'wb') as f: f.write(ts_data) segment_files.append(filename) print("\n所有片段下载完成!") return sorted(segment_files) # 按索引排序

5. 使用FFmpeg合并视频片段

5.1 创建文件列表合并

最可靠的合并方式是使用FFmpeg的concat协议：

import subprocess import tempfile class VideoMerger: def __init__(self, ffmpeg_path='ffmpeg'): self.ffmpeg_path = ffmpeg_path def merge_with_concat(self, segment_files, output_path): """使用concat协议合并ts片段""" # 创建临时文件列表 with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f: for segment_file in segment_files: f.write(f"file '{os.path.abspath(segment_file)}'\n") filelist_path = f.name try: # 使用FFmpeg合并 cmd = [ self.ffmpeg_path, '-f', 'concat', '-safe', '0', '-i', filelist_path, '-c', 'copy', # 直接流复制，不重新编码 '-y', # 覆盖输出文件 output_path ] print("开始合并视频片段...") result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) if result.returncode == 0: print(f"视频合并成功: {output_path}") return True else: print(f"FFmpeg合并失败: {result.stderr}") return False except subprocess.TimeoutExpired: print("合并操作超时") return False except Exception as e: print(f"合并过程异常: {e}") return False finally: # 清理临时文件 if os.path.exists(filelist_path): os.unlink(filelist_path)

5.2 二进制直接合并（备选方案）

如果FFmpeg不可用，可以使用二进制方式合并：

def binary_merge(self, segment_files, output_path): """二进制方式直接合并ts文件""" try: with open(output_path, 'wb') as outfile: for i, segment_file in enumerate(segment_files): print(f"\r合并进度: {i+1}/{len(segment_files)}", end='') with open(segment_file, 'rb') as infile: outfile.write(infile.read()) print(f"\n二进制合并完成: {output_path}") return True except Exception as e: print(f"\n二进制合并失败: {e}") return False

6. 完整实战案例：视频下载器实现

6.1 项目结构设计

video_downloader/ ├── main.py # 主程序入口 ├── spider.py # 网页爬虫模块 ├── m3u8_parser.py # m3u8解析模块 ├── downloader.py # 下载器模块 ├── merger.py # 视频合并模块 └── utils.py # 工具函数

6.2 核心主程序实现

# main.py import os import sys from spider import VideoSpider from m3u8_parser import M3U8Downloader from downloader import ParallelDownloader, TSDecryptor from merger import VideoMerger from utils import sanitize_filename class VideoDownloader: def __init__(self, output_dir="downloads", max_workers=5): self.output_dir = output_dir self.max_workers = max_workers self.spider = VideoSpider() self.m3u8_parser = M3U8Downloader(output_dir) self.downloader = ParallelDownloader(max_workers) self.decryptor = TSDecryptor() self.merger = VideoMerger() os.makedirs(output_dir, exist_ok=True) def download_video(self, url, video_title=None): """主下载流程""" print(f"开始处理视频: {url}") # 1. 获取页面内容 html_content = self.spider.get_page_content(url) if not html_content: print("获取页面内容失败") return False # 2. 提取视频标题 if not video_title: video_title = self.spider.extract_video_title(html_content) print(f"视频标题: {video_title}") # 3. 提取m3u8链接 m3u8_urls = self.spider.extract_m3u8_urls(html_content) if not m3u8_urls: print("未找到m3u8链接") return False m3u8_url = m3u8_urls[0] # 取第一个找到的链接 print(f"找到m3u8链接: {m3u8_url}") # 4. 解析m3u8文件 m3u8_content = self.m3u8_parser.parse_m3u8(m3u8_url) if not m3u8_content: print("解析m3u8文件失败") return False # 5. 获取ts片段信息 segments = self.m3u8_parser.get_ts_segments(m3u8_content) if not segments: print("未找到视频片段") return False print(f"共发现 {len(segments)} 个视频片段") # 6. 创建视频专用目录 video_dir = os.path.join(self.output_dir, sanitize_filename(video_title)) os.makedirs(video_dir, exist_ok=True) # 7. 下载所有ts片段 segment_files = self.downloader.download_all_segments( segments, self.decryptor, video_dir ) if not segment_files: print("下载视频片段失败") return False # 8. 合并视频 output_path = os.path.join(self.output_dir, f"{sanitize_filename(video_title)}.mp4") success = self.merger.merge_with_concat(segment_files, output_path) # 9. 清理临时文件 if success: self.cleanup_temp_files(segment_files) return success def cleanup_temp_files(self, segment_files): """清理临时ts文件""" for file_path in segment_files: try: os.unlink(file_path) except: pass print("临时文件清理完成") if __name__ == "__main__": if len(sys.argv) < 2: print("用法: python main.py <视频页面URL> [视频标题]") sys.exit(1) url = sys.argv[1] title = sys.argv[2] if len(sys.argv) > 2 else None downloader = VideoDownloader() success = downloader.download_video(url, title) if success: print("视频下载完成!") else: print("视频下载失败!")

6.3 工具函数模块

# utils.py import re import os def sanitize_filename(filename): """清理文件名中的非法字符""" # 移除非法文件名字符 invalid_chars = r'[<>:"/\\|?*]' filename = re.sub(invalid_chars, '_', filename) # 限制文件名长度 if len(filename) > 100: filename = filename[:100] return filename.strip() def get_file_size(file_path): """获取文件大小（人类可读格式）""" size = os.path.getsize(file_path) for unit in ['B', 'KB', 'MB', 'GB']: if size < 1024.0: return f"{size:.2f} {unit}" size /= 1024.0 return f"{size:.2f} TB" def check_ffmpeg_available(): """检查FFmpeg是否可用""" try: import subprocess result = subprocess.run(['ffmpeg', '-version'], capture_output=True, text=True) return result.returncode == 0 except: return False

7. 常见问题与解决方案

7.1 网络请求相关问题

问题1：403 Forbidden错误

• 原因：网站反爬虫机制阻止访问
• 解决方案：完善请求头信息，添加Referer、Cookie等字段

def enhance_headers(self, referer=None): """增强请求头信息""" self.headers.update({ 'Accept-Encoding': 'gzip, deflate, br', 'Cache-Control': 'no-cache', 'Referer': referer or 'https://www.example.com/', 'Connection': 'keep-alive', })

问题2：TS片段下载超时

• 原因：网络不稳定或服务器限制
• 解决方案：增加重试机制和超时时间

def download_with_retry(self, url, max_retries=3, timeout=30): """带重试的下载函数""" for attempt in range(max_retries): try: response = self.session.get(url, timeout=timeout) if response.status_code == 200: return response.content except Exception as e: if attempt == max_retries - 1: raise e print(f"下载失败，第{attempt+1}次重试...") return None

7.2 视频处理相关问题

问题3：合并后的视频无法播放

• 原因：TS片段顺序错误或文件损坏
• 解决方案：验证每个片段的完整性，确保按正确顺序合并

def validate_ts_files(self, segment_files): """验证TS文件完整性""" valid_files = [] for file_path in segment_files: try: with open(file_path, 'rb') as f: data = f.read() # 检查文件头（TS文件通常以0x47开头） if len(data) > 0 and data[0] == 0x47: valid_files.append(file_path) else: print(f"文件损坏: {file_path}") except Exception as e: print(f"读取文件失败: {file_path}, {e}") return valid_files

问题4：加密视频解密失败

• 原因：密钥获取失败或IV值不正确
• 解决方案：手动分析m3u8文件，确认加密方式

7.3 性能优化问题

问题5：下载速度过慢

• 原因：单线程下载或网络限制
• 解决方案：调整线程数，使用连接池

from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_retry_session(self, retries=3, backoff_factor=0.3): """创建带重试机制的Session""" session = requests.Session() retry_strategy = Retry( total=retries, backoff_factor=backoff_factor, status_forcelist=[429, 500, 502, 503, 504], ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) return session

8. 高级功能扩展

8.1 支持断点续传

对于大文件下载，实现断点续传功能：

class ResumeDownloader: def __init__(self, download_dir): self.download_dir = download_dir self.progress_file = os.path.join(download_dir, 'download_progress.json') def save_progress(self, segment_index, total_segments): """保存下载进度""" progress = { 'downloaded': segment_index + 1, 'total': total_segments, 'timestamp': time.time() } with open(self.progress_file, 'w') as f: json.dump(progress, f) def load_progress(self): """加载下载进度""" if os.path.exists(self.progress_file): with open(self.progress_file, 'r') as f: return json.load(f) return None def cleanup_progress(self): """清理进度文件""" if os.path.exists(self.progress_file): os.unlink(self.progress_file)

8.2 批量下载支持

扩展支持批量URL下载：

def batch_download(self, url_list, output_dir): """批量下载多个视频""" results = [] for i, url in enumerate(url_list): print(f"\n正在下载第 {i+1}/{len(url_list)} 个视频") try: success = self.download_video(url) results.append({'url': url, 'success': success}) except Exception as e: print(f"下载失败: {e}") results.append({'url': url, 'success': False, 'error': str(e)}) # 生成下载报告 self.generate_report(results, output_dir) return results

8.3 配置文件管理

使用配置文件管理常用参数：

# config.yaml download: output_dir: "./downloads" max_workers: 5 timeout: 30 retry_times: 3 ffmpeg: path: "ffmpeg" output_format: "mp4" headers: user_agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"

9. 法律与道德注意事项

9.1 合法使用原则

在开发和使用视频爬虫时，必须遵守相关法律法规：

1. 仅用于个人学习：不得用于商业用途或大规模分发
2. 尊重版权：不下载明确声明禁止下载的内容
3. 控制访问频率：避免对目标网站造成过大压力
4. 遵守robots.txt：尊重网站的爬虫协议

9.2 技术学习价值

本项目的主要技术学习点：

• HTTP协议和网络请求处理
• 流媒体协议解析
• 多线程编程和并发控制
• 文件处理和格式转换
• 错误处理和重试机制

10. 项目部署与使用

10.1 快速开始指南

1. 环境准备：

git clone <repository-url> cd video_downloader pip install -r requirements.txt

2. 下载单个视频：

python main.py "https://example.com/video-page"

3. 批量下载：

python batch_download.py url_list.txt

10.2 自定义配置

创建配置文件config.yaml来自定义行为：

download: output_dir: "/path/to/downloads" max_workers: 8 timeout: 60 logging: level: "INFO" file: "download.log"

通过本项目的学习，你不仅掌握了m3u8视频下载的技术细节，还了解了完整的爬虫项目开发流程。在实际应用中，请务必遵守相关法律法规，将技术用于正当的学习和研究目的。