m3u8 是什么?
视频网站现在基本都用 HLS 分发：一个完整视频被切成几百上千个 .ts 小片段，再由一个 .m3u8 文本文件记录每个分片的位置和时长。播放器按进度去要分片，看完就扔。所以你想把整部片子存下来，就得把 playlist 里所有分片挨个拉全，再按顺序拼回去——手动做一遍，几百个文件能点到手废。
它干了哪几件事
- tkinter 做的界面，填链接、选保存路径、点下载，不用记命令行
- 线程池并发拉分片，代码里写死 20 个 worker，每个分片最多重试 3 次
- 遇到主播放列表（master playlist）会自动按
bandwidth排序，挑最高码率那一路，不用自己挑清晰度 - 支持 AES-128 加密的流，能自动去取密钥再解密
- 纯 Python + tkinter，Windows / macOS / Linux 都能跑
依赖就三个 requests、m3u8、pycryptodome。
核心逻辑
流程不复杂，读一遍就能懂：
playlist = m3u8.loads(resp.text, uri=url)
if playlist.is_variant: # 主列表，挑码率最高的
sub = sorted(playlist.playlists,
key=lambda p: p.stream_info.bandwidth,
reverse=True)[0]
playlist = self._get_playlist(sub.absolute_uri)
key = self._get_key(playlist) # 有 EXT-X-KEY 就去取密钥下载完用字典 ts_contents[index] = content 存着，最后按 index 顺序写文件。加密的片段用 AES-CBC 解：
iv = (playlist.media_sequence + i).to_bytes(16, 'big')
cipher = AES.new(key, AES.MODE_CBC, iv)
f.write(unpad(cipher.decrypt(content)))这个 IV 的算法是 HLS 规范里的默认值——playlist 里如果没显式给 IV=，就用「媒体序列号 + 分片序号」当 16 字节大端整数。很多人自己写的时候卡在这，下下来一片花屏。
适用边界
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)[0]
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[0]:
key_uri = playlist.keys[0].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[index] = 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[len(s)-1:])]
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()
