资讯详情

深入了解 · 专业解读

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

发布时间：2026/8/2 2:09:41
Python爬虫实战：m3u8视频下载与FFmpeg合并完整指南 在线视频网站爬虫实战m3u8视频下载与合并完整指南在日常开发和学习中我们经常需要从在线视频网站获取视频资源用于分析研究。但很多网站采用m3u8流媒体协议直接将视频分割成多个小片段给下载带来挑战。本文将完整讲解如何通过Python爬虫技术识别m3u8索引文件下载所有ts片段并使用ffmpeg合并成完整视频。1. m3u8流媒体技术原理与背景1.1 什么是m3u8格式m3u8是HTTP Live StreamingHLS协议使用的索引文件格式基于UTF-8编码的m3u播放列表。与传统的mp4等完整视频文件不同m3u8文件本身不包含视频数据而是包含一系列指向视频片段的URL地址。典型的m3u8文件结构如下#EXTM3U #EXT-X-VERSION:3 #EXT-X-TARGETDURATION:10 #EXT-X-MEDIA-SEQUENCE:0 #EXTINF:10.0, http://example.com/segment0.ts #EXTINF:10.0, http://example.com/segment1.ts #EXTINF:10.0, http://example.com/segment2.ts #EXT-X-ENDLIST1.2 为什么视频网站使用m3u8采用m3u8格式有多个技术优势支持自适应码率切换可以根据用户网络状况动态调整视频质量支持边下载边播放提升用户体验便于CDN分发和缓存优化。但这也增加了普通用户下载完整视频的难度。1.3 技术实现思路要实现m3u8视频的完整下载需要经过三个关键步骤首先通过爬虫技术获取m3u8索引文件然后批量下载所有的ts视频片段最后使用ffmpeg工具将这些片段合并成完整视频文件。2. 环境准备与工具安装2.1 Python环境配置本项目需要Python 3.6及以上版本。建议使用conda或virtualenv创建独立的Python环境。# 创建虚拟环境 python -m venv video_spider source video_spider/bin/activate # Linux/Mac # video_spider\Scripts\activate # Windows # 安装必要依赖 pip install requests beautifulsoup4 pycryptodome2.2 FFmpeg安装配置FFmpeg是处理视频合并的核心工具需要单独安装。Windows系统安装访问FFmpeg官网下载Windows版本解压到合适目录如C:\ffmpeg将bin目录添加到系统PATH环境变量在命令行验证ffmpeg -versionLinux系统安装# Ubuntu/Debian sudo apt update sudo apt install ffmpeg # CentOS/RHEL sudo yum install epel-release sudo yum install ffmpegmacOS系统安装brew install ffmpeg安装完成后在终端运行ffmpeg -version确认安装成功。2.3 开发工具准备推荐使用VS Code、PyCharm等现代IDE安装Python扩展支持。同时准备浏览器开发者工具F12用于分析网络请求。3. 爬虫基础与网页分析技术3.1 网页结构分析在开始编写爬虫前需要先分析目标视频网站的页面结构。打开浏览器开发者工具切换到Network标签播放视频并观察网络请求。关键查找目标查找包含.m3u8的请求分析请求头信息Headers注意是否有加密参数或Token验证3.2 请求头模拟为了避免被网站反爬机制识别需要模拟真实浏览器的请求头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/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Upgrade-Insecure-Requests: 1, }3.3 动态内容处理很多现代网站使用JavaScript动态加载内容简单的requests库可能无法获取到完整的页面内容。这时可以考虑使用Selenium模拟浏览器行为from selenium import webdriver from selenium.webdriver.chrome.options import Options def setup_selenium(): chrome_options Options() chrome_options.add_argument(--headless) # 无界面模式 chrome_options.add_argument(--no-sandbox) chrome_options.add_argument(--disable-dev-shm-usage) driver webdriver.Chrome(optionschrome_options) return driver4. m3u8索引文件识别与解析4.1 自动识别m3u8链接通过正则表达式在网页源码中搜索m3u8链接import re import requests def find_m3u8_urls(html_content): 从HTML内容中提取m3u8链接 # 匹配常见的m3u8 URL模式 patterns [ rhttps?://[^\s\]\.m3u8[^\s\]*, r[^\s\]\.m3u8(?:\?[^\s\]*)? ] m3u8_urls [] for pattern in patterns: matches re.findall(pattern, html_content, re.IGNORECASE) m3u8_urls.extend(matches) return list(set(m3u8_urls)) # 去重4.2 解析m3u8文件内容获取到m3u8文件后需要解析其中的ts片段信息def parse_m3u8_content(m3u8_url, headersNone): 解析m3u8文件提取ts片段列表 if headers is None: headers {} response requests.get(m3u8_url, headersheaders) if response.status_code ! 200: raise Exception(f无法获取m3u8文件: {response.status_code}) content response.text lines content.split(\n) ts_segments [] base_url /.join(m3u8_url.split(/)[:-1]) / for line in lines: line line.strip() if line and not line.startswith(#): if line.startswith(http): ts_segments.append(line) else: ts_segments.append(base_url line) return ts_segments4.3 处理加密的m3u8文件有些网站会对m3u8文件或ts片段进行加密需要额外的解密处理from Crypto.Cipher import AES import base64 def decrypt_ts_content(encrypted_data, key_url, ivNone): 解密AES加密的ts片段 # 获取解密密钥 key_response requests.get(key_url) key key_response.content # 设置解密模式 if iv: cipher AES.new(key, AES.MODE_CBC, iviv) else: cipher AES.new(key, AES.MODE_CBC, ivkey) decrypted_data cipher.decrypt(encrypted_data) return decrypted_data5. 完整爬虫实现视频下载器开发5.1 项目结构设计创建完整的项目目录结构video_downloader/ ├── main.py # 主程序入口 ├── m3u8_parser.py # m3u8解析模块 ├── downloader.py # 下载器模块 ├── merger.py # 视频合并模块 ├── utils.py # 工具函数 ├── config/ # 配置文件目录 │ └── headers.json # 请求头配置 └── downloads/ # 下载文件存储目录5.2 核心下载器类实现import os import requests from concurrent.futures import ThreadPoolExecutor, as_completed import threading class VideoDownloader: def __init__(self, max_workers5, timeout30): self.max_workers max_workers self.timeout timeout self.downloaded_count 0 self.lock threading.Lock() def download_ts_segment(self, ts_url, filename, headersNone): 下载单个ts片段 try: if headers is None: headers {} response requests.get(ts_url, headersheaders, timeoutself.timeout) if response.status_code 200: with open(filename, wb) as f: f.write(response.content) with self.lock: self.downloaded_count 1 return True else: print(f下载失败 {ts_url}: HTTP {response.status_code}) return False except Exception as e: print(f下载失败 {ts_url}: {str(e)}) return False def download_all_segments(self, ts_urls, output_dir, headersNone): 并发下载所有ts片段 if not os.path.exists(output_dir): os.makedirs(output_dir) total_segments len(ts_urls) self.downloaded_count 0 with ThreadPoolExecutor(max_workersself.max_workers) as executor: future_to_url {} for i, ts_url in enumerate(ts_urls): filename os.path.join(output_dir, fsegment_{i:05d}.ts) future executor.submit(self.download_ts_segment, ts_url, filename, headers) future_to_url[future] (ts_url, filename) for future in as_completed(future_to_url): ts_url, filename future_to_url[future] try: success future.result() if success: print(f进度: {self.downloaded_count}/{total_segments}) except Exception as e: print(f异常 {ts_url}: {str(e)}) return self.downloaded_count total_segments5.3 主程序流程控制import json from urllib.parse import urljoin class M3U8VideoDownloader: def __init__(self, config_fileconfig/headers.json): self.downloader VideoDownloader() self.load_config(config_file) def load_config(self, config_file): 加载配置文件 try: with open(config_file, r, encodingutf-8) as f: self.config json.load(f) except FileNotFoundError: self.config { headers: { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } } def process_video(self, video_url, output_filename): 处理视频下载完整流程 print(f开始处理视频: {video_url}) # 1. 获取视频页面查找m3u8链接 m3u8_urls self.find_m3u8_links(video_url) if not m3u8_urls: print(未找到m3u8链接) return False main_m3u8_url self.select_quality_m3u8(m3u8_urls) print(f使用m3u8文件: {main_m3u8_url}) # 2. 解析m3u8文件 ts_segments parse_m3u8_content(main_m3u8_url, self.config[headers]) if not ts_segments: print(未找到ts片段) return False print(f找到 {len(ts_segments)} 个ts片段) # 3. 下载所有ts片段 temp_dir ftemp_{hash(video_url)} success self.downloader.download_all_segments( ts_segments, temp_dir, self.config[headers] ) if not success: print(部分片段下载失败) # 可以选择继续合并已下载的部分 # 4. 合并视频 merger VideoMerger() result merger.merge_ts_files(temp_dir, output_filename) # 5. 清理临时文件 self.cleanup_temp_files(temp_dir) if result: print(f视频下载完成: {output_filename}) else: print(视频合并失败) return result def find_m3u8_links(self, video_url): 从视频页面查找m3u8链接 response requests.get(video_url, headersself.config[headers]) if response.status_code ! 200: return [] return find_m3u8_urls(response.text) def select_quality_m3u8(self, m3u8_urls): 选择合适质量的m3u8文件 # 简单实现返回第一个可用的m3u8 for url in m3u8_urls: try: response requests.head(url, headersself.config[headers], timeout5) if response.status_code 200: return url except: continue return m3u8_urls[0] if m3u8_urls else None def cleanup_temp_files(self, temp_dir): 清理临时文件 import shutil try: shutil.rmtree(temp_dir) print(f已清理临时目录: {temp_dir}) except Exception as e: print(f清理临时文件失败: {str(e)})6. 使用FFmpeg合并视频文件6.1 FFmpeg合并方法FFmpeg提供了多种合并视频文件的方法针对ts片段最有效的是concat协议import subprocess import os class VideoMerger: def merge_ts_files(self, ts_dir, output_file): 使用FFmpeg合并ts文件 # 方法1: 使用concat协议推荐 file_list self.create_file_list(ts_dir) concat_file os.path.join(ts_dir, file_list.txt) try: # 写入文件列表 with open(concat_file, w, encodingutf-8) as f: for ts_file in file_list: f.write(ffile {os.path.abspath(ts_file)}\n) # 使用FFmpeg合并 cmd [ ffmpeg, -f, concat, -safe, 0, -i, concat_file, -c, copy, output_file ] result subprocess.run(cmd, capture_outputTrue, textTrue, timeout300) if result.returncode 0: print(视频合并成功) return True else: print(fFFmpeg错误: {result.stderr}) return False except subprocess.TimeoutExpired: print(合并操作超时) return False except Exception as e: print(f合并异常: {str(e)}) return False def create_file_list(self, ts_dir): 创建按顺序排列的ts文件列表 ts_files [] for filename in os.listdir(ts_dir): if filename.endswith(.ts): ts_files.append(os.path.join(ts_dir, filename)) # 按文件名排序假设文件名包含序号 ts_files.sort() return ts_files def alternative_merge_method(self, ts_dir, output_file): 备选合并方法直接拼接二进制文件 ts_files self.create_file_list(ts_dir) try: with open(output_file, wb) as outfile: for ts_file in ts_files: with open(ts_file, rb) as infile: outfile.write(infile.read()) return True except Exception as e: print(f二进制合并失败: {str(e)}) return False6.2 视频格式转换合并后的视频可能需要转换为更通用的格式def convert_video_format(self, input_file, output_file, formatmp4): 转换视频格式 cmd [ ffmpeg, -i, input_file, -c, copy, # 直接流复制不重新编码 output_file ] try: result subprocess.run(cmd, capture_outputTrue, textTrue, timeout300) return result.returncode 0 except Exception as e: print(f格式转换失败: {str(e)}) return False7. 高级功能与优化7.1 断点续传功能实现下载中断后能够继续下载的功能class ResumeDownloader(VideoDownloader): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.progress_file download_progress.json def load_progress(self, download_id): 加载下载进度 try: with open(self.progress_file, r) as f: progress json.load(f) return progress.get(download_id, {}) except FileNotFoundError: return {} def save_progress(self, download_id, progress): 保存下载进度 try: with open(self.progress_file, r) as f: all_progress json.load(f) except FileNotFoundError: all_progress {} all_progress[download_id] progress with open(self.progress_file, w) as f: json.dump(all_progress, f) def download_with_resume(self, ts_urls, output_dir, download_id, headersNone): 支持断点续传的下载 progress self.load_progress(download_id) downloaded_files progress.get(downloaded, []) # 过滤已下载的文件 remaining_urls [] for i, ts_url in enumerate(ts_urls): filename os.path.join(output_dir, fsegment_{i:05d}.ts) if filename not in downloaded_files or not os.path.exists(filename): remaining_urls.append(ts_url) if remaining_urls: success self.download_all_segments(remaining_urls, output_dir, headers) if success: progress[downloaded] [ os.path.join(output_dir, fsegment_{i:05d}.ts) for i in range(len(ts_urls)) ] self.save_progress(download_id, progress) return success else: print(所有文件已下载完成) return True7.2 多线程优化与速率限制import time from threading import Semaphore class RateLimitedDownloader(VideoDownloader): def __init__(self, max_workers5, requests_per_second2, *args, **kwargs): super().__init__(max_workers, *args, **kwargs) self.semaphore Semaphore(requests_per_second) self.last_request_time 0 self.min_interval 1.0 / requests_per_second def rate_limited_get(self, url, headersNone, timeout30): 带速率限制的请求 with self.semaphore: current_time time.time() elapsed current_time - self.last_request_time if elapsed self.min_interval: time.sleep(self.min_interval - elapsed) response requests.get(url, headersheaders, timeouttimeout) self.last_request_time time.time() return response def download_ts_segment(self, ts_url, filename, headersNone): 重写下载方法加入速率限制 try: response self.rate_limited_get(ts_url, headers, self.timeout) if response.status_code 200: with open(filename, wb) as f: f.write(response.content) with self.lock: self.downloaded_count 1 return True else: print(f下载失败 {ts_url}: HTTP {response.status_code}) return False except Exception as e: print(f下载失败 {ts_url}: {str(e)}) return False8. 常见问题与解决方案8.1 下载失败问题排查问题1无法找到m3u8链接原因网页内容动态加载、需要特定请求头、链接被JavaScript生成解决方案使用Selenium模拟浏览器、检查网络请求、尝试不同的User-Agent问题2ts片段下载失败原因网络超时、服务器限制、IP被封解决方案增加超时时间、添加重试机制、使用代理IP问题3合并后的视频无法播放原因ts片段顺序错误、文件损坏、编码问题解决方案检查文件顺序、验证每个ts文件完整性、尝试重新编码8.2 性能优化建议并发控制根据目标网站承受能力调整并发数避免被封IP缓存策略对已解析的m3u8文件进行缓存减少重复请求错误重试实现指数退避的重试机制提高下载成功率内存管理处理大文件时使用流式下载避免内存溢出8.3 安全与合规注意事项遵守robots.txt尊重网站的爬虫协议访问频率控制避免对目标网站造成过大压力版权意识仅将下载内容用于个人学习研究用户代理标识使用明确的User-Agent标识爬虫身份9. 完整使用示例9.1 基础使用示例def main(): # 创建下载器实例 downloader M3U8VideoDownloader() # 下载视频 video_url https://example.com/video-page output_file downloaded_video.mp4 success downloader.process_video(video_url, output_file) if success: print(视频下载成功) else: print(视频下载失败) if __name__ __main__: main()9.2 高级配置示例def advanced_example(): # 自定义配置 custom_config { headers: { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://example.com/, Cookie: your_cookie_here }, download: { max_workers: 3, timeout: 60, retry_times: 3 } } # 保存配置 with open(config/custom_config.json, w) as f: json.dump(custom_config, f) # 使用自定义配置 downloader M3U8VideoDownloader(config/custom_config.json) downloader.process_video(video_url, output.mp4)10. 最佳实践与工程建议10.1 代码组织规范模块化设计将功能拆分为独立的模块提高代码复用性配置文件管理将所有可配置参数集中管理便于维护日志记录实现完整的日志系统便于问题排查异常处理对可能出现的异常情况进行妥善处理10.2 生产环境注意事项代理池管理在需要大量下载时使用代理IP池分布式部署将下载任务分布到多个节点执行监控告警实现系统监控和异常告警机制数据备份定期备份重要配置和进度数据10.3 法律与道德边界在使用爬虫技术时必须严格遵守相关法律法规仅用于技术学习和研究目的尊重网站的服务条款和使用协议不进行商业用途的批量下载不干扰网站的正常运营本文提供的技术方案主要面向开发者学习网络编程和视频处理技术在实际应用中请确保遵守相关法律法规和网站的使用条款。通过本教程您应该能够掌握m3u8视频下载的核心技术栈并具备根据实际需求定制化开发的能力。

相关新闻