Python 最简易版 m3u8 跨平台下载实现
带简易图形界面，之前有需求写的，几个月前就在github上开源，仓库名：“and-simple-downloader” 喜欢的希望点点小星星功能特性
图形用户界面: 操作直观，无需命令行知识。
多线程下载: 同时下载多个视频分片，显著提升下载速度。
支持主播放列表: 能自动检测包含多种清晰度的 M3U8 主列表，并选择最高码率的流进行下载。
AES-128 解密: 支持下载经过 AES-128 加密的视频流。
跨平台: 基于 Python 和 tkinter，可在 Windows, macOS 和 Linux 上运行。
import os
import requests
import m3u8
from Crypto.Cipher import AES
import tkinter as tk
from tkinter import ttk, filedialog, messagebox
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
# --- Core Downloader Logic ---
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‘
}
MAX_RETRIES = 3
MAX_WORKERS = 20# 并发下载线程数
class Downloader:
def __init__（self, m3u8_url, output_filename, progress_callback, status_callback）:
self.m3u8_url = m3u8_url
self.output_filename = output_filename
self.progress_callback = progress_callback
self.status_callback = status_callback
self.session = requests.Session（）
def run（self）:
try:
self.status_callback（“正在解析 M3U8...“）
playlist = self._get_playlist（self.m3u8_url）
if not playlist:
raise Exception（“无法获取或解析 M3U8 文件。“）
if playlist.is_variant:
self.status_callback（“检测到主播放列表，选择最高码率...“）
stream_info = sorted（playlist.playlists, key=lambda p: p.stream_info.bandwidth, reverse=True）
playlist_url = stream_info.absolute_uri
playlist = self._get_playlist（playlist_url）
if not playlist:
raise Exception（“获取子播放列表失败。“）
key = self._get_key（playlist）
self._download_and_merge（playlist, key）
self.status_callback（f“下载完成！文件保存在: {self.output_filename}“）
except Exception as e:
self.status_callback（f“错误: {e}“）
messagebox.showerror（“错误“, str（e））
def _get_playlist（self, url）:
try:
response = self.session.get（url, headers=HEADERS, timeout=10）
response.raise_for_status（）
return m3u8.loads（response.text, uri=url）
except requests.exceptions.RequestException as e:
print（f“获取 M3U8 文件失败: {e}“）
return None
def _get_key（self, playlist）:
if playlist.keys and playlist.keys:
key_uri = playlist.keys.absolute_uri
self.status_callback（f“正在获取密钥: {key_uri}“）
try:
response = self.session.get（key_uri, headers=HEADERS, timeout=10）
response.raise_for_status（）
self.status_callback（“密钥获取成功。“）
return response.content
except requests.exceptions.RequestException as e:
raise Exception（f“获取密钥失败: {e}“）
return None
def _download_segment（self, segment_info）:
index, segment = segment_info
ts_url = segment.absolute_uri
for _ in range（MAX_RETRIES）:
try:
response = self.session.get（ts_url, timeout=15）
response.raise_for_status（）
return index, response.content
except requests.exceptions.RequestException:
continue
print（f“分片 {index} 下载失败: {ts_url}“）
return index, None
def _download_and_merge（self, playlist, key）:
segments = list（enumerate（playlist.segments））
total_segments = len（segments）
self.status_callback（f“开始下载 {total_segments} 个分片...“）
# 使用字典来保证分片顺序
ts_contents = {}
with ThreadPoolExecutor（max_workers=MAX_WORKERS） as executor:
futures = {executor.submit（self._download_segment, s）: s for s in segments}
for i, future in enumerate（as_completed（futures））:
index, content = future.result（）
if content:
ts_contents = content
progress = （i + 1） / total_segments * 100
self.progress_callback（progress）
if len（ts_contents） != total_segments:
print（“警告：部分分片下载失败。“）
self.status_callback（“下载完成，开始合并文件...“）
with open（self.output_filename, ‘wb‘） as f:
for i in range（total_segments）:
content = ts_contents.get（i）
if not content:
continue
if key:
iv = （playlist.media_sequence + i）.to_bytes（16, ‘big‘）
cipher = AES.new（key, AES.MODE_CBC, iv）
try:
decrypted_content = cipher.decrypt（content）
# 简单的 unpad，可能不适用于所有情况
unpad = lambda s: s[:-ord（s）]
f.write（unpad（decrypted_content））
except Exception:
f.write（content） # 解密失败则直接写入
else:
f.write（content）
# --- GUI Application ---
class App（tk.Tk）:
def __init__（self）:
super（）.__init__（）
self.title（“M3U8 下载器“）
self.geometry（“500x200“）
self.url_label = ttk.Label（self, text=“M3U8 链接:“）
self.url_label.pack（pady=5）
self.url_entry = ttk.Entry（self, width=60）
self.url_entry.pack（pady=5, padx=10）
self.download_button = ttk.Button（self, text=“下载“, command=self.start_download）
self.download_button.pack（pady=10）
self.progress = ttk.Progressbar（self, orient=“horizontal“, length=400, mode=“determinate“）
self.progress.pack（pady=10）
self.status_label = ttk.Label（self, text=“等待下载...“）
self.status_label.pack（pady=5）
def start_download（self）:
m3u8_url = self.url_entry.get（）
if not m3u8_url:
messagebox.showwarning（“警告“, “请输入 M3U8 链接！“）
return
output_filename = filedialog.asksaveasfilename（
title=“保存视频文件“,
defaultextension=“.mp4“,
filetypes=[（“MPEG-4“, “*.mp4“）, （“All Files“, “*.*“）]
）
if not output_filename:
return
self.download_button.config（state=tk.DISABLED）
self.progress[“value“] = 0
downloader = Downloader（
m3u8_url,
output_filename,
self.update_progress,
self.update_status
）
# 在新线程中运行下载
threading.Thread（target=downloader.run, daemon=True）.start（）
def update_progress（self, value）:
self.progress[“value“] = value
self.update_idletasks（）
def update_status（self, text）:
self.status_label.config（text=text）
if “完成“ in text or “错误“ in text:
self.download_button.config（state=tk.NORMAL）
self.update_idletasks（）
if __name__ == “__main__“:
app = App（）
app.mainloop（）
复制代码
页:
[1]