### 使用 Python 下载 m3u8 视频的脚本 以下是基于 `requests` 和 `concurrent.futures` 的实现方法，用于从指定 URL 中解析并下载 m3u8 文件中的 ts 片段。此代码会处理多个片段的并发下载，并最终将它们合并成完整的视频文件。 #### 脚本说明 1. **发送 HTTP 请求**：通过 `requests` 库获取 m3u8 文件的内容。 2. **解析 TS 列表**：使用正则表达式或其他方式提取所有的 `.ts` 文件路径。 3. **多线程下载**：利用 `concurrent.futures.ThreadPoolExecutor` 并发下载所有 `.ts` 文件。 4. **合并文件**：将下载好的 `.ts` 文件按顺序拼接为一个完整的 MP4 或其他格式的视频文件。 ```python import os import re import requests from concurrent.futures import ThreadPoolExecutor def fetch_m3u8_content(url): """获取 m3u8 文件内容""" headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36' } response = requests.get(url, headers=headers) if response.status_code != 200: raise Exception(f"Failed to retrieve M3U8 content from {url}") return response.text def parse_ts_list(m3u8_text, base_url=""): """解析 m3u8 文件中的 .ts 链接列表""" lines = m3u8_text.splitlines() ts_files = [] for line in lines: if not line.startswith("#") and line.strip(): full_url = line if line.startswith("http") else f"{base_url}/{line}" ts_files.append(full_url) return ts_files def download_file(url, save_path): """下载单个 .ts 文件""" try: response = requests.get(url, stream=True) with open(save_path, "wb") as file: for chunk in response.iter_content(chunk_size=1024 * 1024): if chunk: file.write(chunk) except Exception as e: print(f"Error downloading {url}: {e}") def merge_ts_files(ts_paths, output_path): """合并多个 .ts 文件到一个视频文件""" with open(output_path, "wb") as merged_file: for ts_path in sorted(ts_paths): with open(ts_path, "rb") as part_file: merged_file.write(part_file.read()) def main(video_url, output_filename="output.mp4", max_workers=10): """主函数：从给定的 m3u8 URL 下载并保存视频""" # Step 1: Fetch the M3U8 playlist m3u8_content = fetch_m3u8_content(video_url) # Step 2: Parse all .ts URLs video_dir = os.path.dirname(video_url) ts_urls = parse_ts_list(m3u8_content, base_url=video_dir) # Step 3: Download each .ts segment concurrently temp_folder = "./temp" os.makedirs(temp_folder, exist_ok=True) downloaded_files = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = {} for idx, url in enumerate(ts_urls): filename = os.path.join(temp_folder, f"segment_{idx}.ts") future = executor.submit(download_file, url, filename) futures[future] = filename for future in futures: try: future.result() downloaded_files.append(futures[future]) except Exception as e: print(e) # Step 4: Merge all segments into a single video file merge_ts_files(downloaded_files, output_filename) print(f"Video saved to {output_filename}") if __name__ == "__main__": m3u8_video_url = input("Enter the M3U8 URL of the video: ") output_name = input("Enter the desired output filename (default is output.mp4): ") or "output.mp4" main(m3u8_video_url, output_name) ``` --- ### 关键点解释 1. **M3U8 解析逻辑**: 此处采用简单的字符串分割来提取 `.ts` 文件链接[^1]。如果遇到相对路径，则将其转换为绝对路径以便后续访问。 2. **并发下载机制**: 使用 `ThreadPoolExecutor` 实现高效的并发下载功能[^4]。相比串行下载，这种方法可以显著提升速度。 3. **错误重试机制**: 如果某些 `.ts` 文件未能成功下载，可以通过额外的循环重新尝试这些失败的任务。 4. **文件合并操作**: 将所有已下载的小片段按照原始顺序组合起来形成最终的媒体文件[^2]。 --- ### 注意事项 - 确保目标服务器允许抓取数据；部分站点可能设置防盗链保护措施。 - 对于非常大的视频流，请考虑磁盘空间以及网络带宽的影响。 - 可能需要调整最大工作线程数 (`max_workers`) 来适应不同的硬件配置和性能需求。 ---