在线视频网站爬虫实战m3u8视频下载与合并完整指南在日常开发和学习中我们经常需要从在线视频网站获取视频资源用于分析研究。面对网站采用m3u8流媒体协议的情况传统下载方式往往无法直接获取完整视频。本文将完整介绍基于Python的在线视频爬虫开发重点解决m3u8视频的下载、解密和合并问题。1. m3u8流媒体技术解析1.1 什么是m3u8格式m3u8是HTTP Live StreamingHLS协议使用的播放列表格式基于文本文件存储媒体片段信息。与传统的MP4等单一文件格式不同m3u8将视频分割成多个小的tsTransport 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/HLSMP4文件结构多个ts片段索引文件单个文件下载难度需要处理多个文件直接下载自适应码率支持不支持网络适应性更好一般处理复杂度较高较低理解这些差异有助于我们设计合适的下载策略。m3u8下载的核心在于解析索引文件、批量下载ts片段最后合并成完整视频。2. 开发环境准备2.1 Python环境配置本项目需要Python 3.7及以上版本推荐使用Anaconda或Miniconda管理环境。创建专用环境conda create -n video_spider python3.8 conda activate video_spider2.2 必要依赖库安装核心依赖包pip install requests beautifulsoup4 pycryptodome m3u8各库作用说明requests网络请求库用于获取网页内容和视频片段beautifulsoup4HTML解析库用于提取视频链接pycryptodome加解密库用于处理加密的ts片段m3u8专业的m3u8解析库简化索引文件处理2.3 FFmpeg工具安装FFmpeg是视频处理的核心工具用于ts片段的合并和格式转换。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 ffmpeg验证FFmpeg安装ffmpeg -version正常输出版本信息即表示安装成功。3. 网页视频链接抓取技术3.1 分析视频网站结构在开始编码前需要先了解目标网站的视频加载机制。以典型视频网站为例视频链接通常通过以下方式加载直接链接页面中直接包含视频URL异步加载通过Ajax请求获取视频信息动态渲染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/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, } def get_page_content(self, url): 获取网页内容 try: response self.session.get(url, headersself.headers, timeout10) 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 None3.3 提取m3u8链接的正则表达式不同网站的m3u8链接格式各异需要针对性地编写提取规则def extract_m3u8_urls(self, html_content): 从HTML内容中提取m3u8链接 m3u8_patterns [ rhttps?://[^\s]?\.m3u8(?:\?[^\s]*)?, rhttps?://[^\s]?/index\.m3u8, rsrc:\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[propertyog: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_video4. m3u8文件解析与下载4.1 使用m3u8库解析索引文件import m3u8 import os from urllib.parse import urljoin, urlparse class M3U8Downloader: def __init__(self, output_dirdownloads): self.output_dir output_dir self.session requests.Session() os.makedirs(output_dir, exist_okTrue) def parse_m3u8(self, m3u8_url): 解析m3u8文件内容 try: response self.session.get(m3u8_url, timeout10) if response.status_code 200: # 解析m3u8内容 m3u8_content m3u8.loads(response.text, urim3u8_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 segments4.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, timeout10) 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, ivNone): 解密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_data4.3 多线程下载ts片段为了提高下载效率使用多线程并行下载ts片段import threading from concurrent.futures import ThreadPoolExecutor, as_completed class ParallelDownloader: def __init__(self, max_workers5): 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, timeout30) 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_workersself.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, fsegment_{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_pathffmpeg): self.ffmpeg_path ffmpeg_path def merge_with_concat(self, segment_files, output_path): 使用concat协议合并ts片段 # 创建临时文件列表 with tempfile.NamedTemporaryFile(modew, suffix.txt, deleteFalse) as f: for segment_file in segment_files: f.write(ffile {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_outputTrue, textTrue, timeout300) 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 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合并进度: {i1}/{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 False6. 完整实战案例视频下载器实现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_dirdownloads, max_workers5): 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_okTrue) def download_video(self, url, video_titleNone): 主下载流程 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_okTrue) # 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_outputTrue, textTrue) return result.returncode 0 except: return False7. 常见问题与解决方案7.1 网络请求相关问题问题1403 Forbidden错误原因网站反爬虫机制阻止访问解决方案完善请求头信息添加Referer、Cookie等字段def enhance_headers(self, refererNone): 增强请求头信息 self.headers.update({ Accept-Encoding: gzip, deflate, br, Cache-Control: no-cache, Referer: referer or https://www.example.com/, Connection: keep-alive, })问题2TS片段下载超时原因网络不稳定或服务器限制解决方案增加重试机制和超时时间def download_with_retry(self, url, max_retries3, timeout30): 带重试的下载函数 for attempt in range(max_retries): try: response self.session.get(url, timeouttimeout) if response.status_code 200: return response.content except Exception as e: if attempt max_retries - 1: raise e print(f下载失败第{attempt1}次重试...) return None7.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, retries3, backoff_factor0.3): 创建带重试机制的Session session requests.Session() retry_strategy Retry( totalretries, backoff_factorbackoff_factor, status_forcelist[429, 500, 502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) return session8. 高级功能扩展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正在下载第 {i1}/{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 results8.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.369. 法律与道德注意事项9.1 合法使用原则在开发和使用视频爬虫时必须遵守相关法律法规仅用于个人学习不得用于商业用途或大规模分发尊重版权不下载明确声明禁止下载的内容控制访问频率避免对目标网站造成过大压力遵守robots.txt尊重网站的爬虫协议9.2 技术学习价值本项目的主要技术学习点HTTP协议和网络请求处理流媒体协议解析多线程编程和并发控制文件处理和格式转换错误处理和重试机制10. 项目部署与使用10.1 快速开始指南环境准备git clone repository-url cd video_downloader pip install -r requirements.txt下载单个视频python main.py https://example.com/video-page批量下载python batch_download.py url_list.txt10.2 自定义配置创建配置文件config.yaml来自定义行为download: output_dir: /path/to/downloads max_workers: 8 timeout: 60 logging: level: INFO file: download.log通过本项目的学习你不仅掌握了m3u8视频下载的技术细节还了解了完整的爬虫项目开发流程。在实际应用中请务必遵守相关法律法规将技术用于正当的学习和研究目的。
在线视频网站爬虫实战m3u8视频下载与合并完整指南在日常开发和学习中我们经常需要从在线视频网站获取视频资源用于分析研究。面对网站采用m3u8流媒体协议的情况传统下载方式往往无法直接获取完整视频。本文将完整介绍基于Python的在线视频爬虫开发重点解决m3u8视频的下载、解密和合并问题。1. m3u8流媒体技术解析1.1 什么是m3u8格式m3u8是HTTP Live StreamingHLS协议使用的播放列表格式基于文本文件存储媒体片段信息。与传统的MP4等单一文件格式不同m3u8将视频分割成多个小的tsTransport 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/HLSMP4文件结构多个ts片段索引文件单个文件下载难度需要处理多个文件直接下载自适应码率支持不支持网络适应性更好一般处理复杂度较高较低理解这些差异有助于我们设计合适的下载策略。m3u8下载的核心在于解析索引文件、批量下载ts片段最后合并成完整视频。2. 开发环境准备2.1 Python环境配置本项目需要Python 3.7及以上版本推荐使用Anaconda或Miniconda管理环境。创建专用环境conda create -n video_spider python3.8 conda activate video_spider2.2 必要依赖库安装核心依赖包pip install requests beautifulsoup4 pycryptodome m3u8各库作用说明requests网络请求库用于获取网页内容和视频片段beautifulsoup4HTML解析库用于提取视频链接pycryptodome加解密库用于处理加密的ts片段m3u8专业的m3u8解析库简化索引文件处理2.3 FFmpeg工具安装FFmpeg是视频处理的核心工具用于ts片段的合并和格式转换。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 ffmpeg验证FFmpeg安装ffmpeg -version正常输出版本信息即表示安装成功。3. 网页视频链接抓取技术3.1 分析视频网站结构在开始编码前需要先了解目标网站的视频加载机制。以典型视频网站为例视频链接通常通过以下方式加载直接链接页面中直接包含视频URL异步加载通过Ajax请求获取视频信息动态渲染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/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, } def get_page_content(self, url): 获取网页内容 try: response self.session.get(url, headersself.headers, timeout10) 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 None3.3 提取m3u8链接的正则表达式不同网站的m3u8链接格式各异需要针对性地编写提取规则def extract_m3u8_urls(self, html_content): 从HTML内容中提取m3u8链接 m3u8_patterns [ rhttps?://[^\s]?\.m3u8(?:\?[^\s]*)?, rhttps?://[^\s]?/index\.m3u8, rsrc:\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[propertyog: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_video4. m3u8文件解析与下载4.1 使用m3u8库解析索引文件import m3u8 import os from urllib.parse import urljoin, urlparse class M3U8Downloader: def __init__(self, output_dirdownloads): self.output_dir output_dir self.session requests.Session() os.makedirs(output_dir, exist_okTrue) def parse_m3u8(self, m3u8_url): 解析m3u8文件内容 try: response self.session.get(m3u8_url, timeout10) if response.status_code 200: # 解析m3u8内容 m3u8_content m3u8.loads(response.text, urim3u8_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 segments4.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, timeout10) 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, ivNone): 解密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_data4.3 多线程下载ts片段为了提高下载效率使用多线程并行下载ts片段import threading from concurrent.futures import ThreadPoolExecutor, as_completed class ParallelDownloader: def __init__(self, max_workers5): 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, timeout30) 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_workersself.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, fsegment_{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_pathffmpeg): self.ffmpeg_path ffmpeg_path def merge_with_concat(self, segment_files, output_path): 使用concat协议合并ts片段 # 创建临时文件列表 with tempfile.NamedTemporaryFile(modew, suffix.txt, deleteFalse) as f: for segment_file in segment_files: f.write(ffile {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_outputTrue, textTrue, timeout300) 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 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合并进度: {i1}/{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 False6. 完整实战案例视频下载器实现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_dirdownloads, max_workers5): 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_okTrue) def download_video(self, url, video_titleNone): 主下载流程 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_okTrue) # 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_outputTrue, textTrue) return result.returncode 0 except: return False7. 常见问题与解决方案7.1 网络请求相关问题问题1403 Forbidden错误原因网站反爬虫机制阻止访问解决方案完善请求头信息添加Referer、Cookie等字段def enhance_headers(self, refererNone): 增强请求头信息 self.headers.update({ Accept-Encoding: gzip, deflate, br, Cache-Control: no-cache, Referer: referer or https://www.example.com/, Connection: keep-alive, })问题2TS片段下载超时原因网络不稳定或服务器限制解决方案增加重试机制和超时时间def download_with_retry(self, url, max_retries3, timeout30): 带重试的下载函数 for attempt in range(max_retries): try: response self.session.get(url, timeouttimeout) if response.status_code 200: return response.content except Exception as e: if attempt max_retries - 1: raise e print(f下载失败第{attempt1}次重试...) return None7.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, retries3, backoff_factor0.3): 创建带重试机制的Session session requests.Session() retry_strategy Retry( totalretries, backoff_factorbackoff_factor, status_forcelist[429, 500, 502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) return session8. 高级功能扩展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正在下载第 {i1}/{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 results8.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.369. 法律与道德注意事项9.1 合法使用原则在开发和使用视频爬虫时必须遵守相关法律法规仅用于个人学习不得用于商业用途或大规模分发尊重版权不下载明确声明禁止下载的内容控制访问频率避免对目标网站造成过大压力遵守robots.txt尊重网站的爬虫协议9.2 技术学习价值本项目的主要技术学习点HTTP协议和网络请求处理流媒体协议解析多线程编程和并发控制文件处理和格式转换错误处理和重试机制10. 项目部署与使用10.1 快速开始指南环境准备git clone repository-url cd video_downloader pip install -r requirements.txt下载单个视频python main.py https://example.com/video-page批量下载python batch_download.py url_list.txt10.2 自定义配置创建配置文件config.yaml来自定义行为download: output_dir: /path/to/downloads max_workers: 8 timeout: 60 logging: level: INFO file: download.log通过本项目的学习你不仅掌握了m3u8视频下载的技术细节还了解了完整的爬虫项目开发流程。在实际应用中请务必遵守相关法律法规将技术用于正当的学习和研究目的。
)
