带简易图形界面，之前有需求写的，几个月前就在 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)[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()© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
暂无评论内容