Python 最简易版 m3u8 跨平台下载实现

小K

带简易图形界面，之前有需求写的，几个月前就在github上开源，仓库名：“and-simple-downloader” 喜欢的希望点点小星星

功能特性
图形用户界面: 操作直观，无需命令行知识。
多线程下载: 同时下载多个视频分片，显著提升下载速度。
支持主播放列表: 能自动检测包含多种清晰度的 M3U8 主列表，并选择最高码率的流进行下载。
AES-128 解密: 支持下载经过 AES-128 加密的视频流。
跨平台: 基于 Python 和 tkinter，可在 Windows, macOS 和 Linux 上运行。

  

1. [code]
2. import os
3. import requests
4. import m3u8
5. from Crypto.Cipher import AES
6. import tkinter as tk
7. from tkinter import ttk, filedialog, messagebox
8. import threading
9. from concurrent.futures import ThreadPoolExecutor, as_completed
10. # --- Core Downloader Logic ---
11. HEADERS = {
12.     ‘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‘
13. }
14. MAX_RETRIES = 3
15. MAX_WORKERS = 20  # 并发下载线程数
16. class Downloader:
17.     def __init__（self, m3u8_url, output_filename, progress_callback, status_callback）:
18.         self.m3u8_url = m3u8_url
19.         self.output_filename = output_filename
20.         self.progress_callback = progress_callback
21.         self.status_callback = status_callback
22.         self.session = requests.Session（）
23.     def run（self）:
24.         try:
25.             self.status_callback（“正在解析 M3U8...“）
26.             playlist = self._get_playlist（self.m3u8_url）
27.             if not playlist:
28.                 raise Exception（“无法获取或解析 M3U8 文件。“）
29.             if playlist.is_variant:
30.                 self.status_callback（“检测到主播放列表，选择最高码率...“）
31.                 stream_info = sorted（playlist.playlists, key=lambda p: p.stream_info.bandwidth, reverse=True）[0]
32.                 playlist_url = stream_info.absolute_uri
33.                 playlist = self._get_playlist（playlist_url）
34.                 if not playlist:
35.                     raise Exception（“获取子播放列表失败。“）
36.             key = self._get_key（playlist）
37.             
38.             self._download_and_merge（playlist, key）
39.             
40.             self.status_callback（f“下载完成！文件保存在: {self.output_filename}“）
41.         except Exception as e:
42.             self.status_callback（f“错误: {e}“）
43.             messagebox.showerror（“错误“, str（e））
44.     def _get_playlist（self, url）:
45.         try:
46.             response = self.session.get（url, headers=HEADERS, timeout=10）
47.             response.raise_for_status（）
48.             return m3u8.loads（response.text, uri=url）
49.         except requests.exceptions.RequestException as e:
50.             print（f“获取 M3U8 文件失败: {e}“）
51.             return None
52.     def _get_key（self, playlist）:
53.         if playlist.keys and playlist.keys[0]:
54.             key_uri = playlist.keys[0].absolute_uri
55.             self.status_callback（f“正在获取密钥: {key_uri}“）
56.             try:
57.                 response = self.session.get（key_uri, headers=HEADERS, timeout=10）
58.                 response.raise_for_status（）
59.                 self.status_callback（“密钥获取成功。“）
60.                 return response.content
61.             except requests.exceptions.RequestException as e:
62.                 raise Exception（f“获取密钥失败: {e}“）
63.         return None
64.     def _download_segment（self, segment_info）:
65.         index, segment = segment_info
66.         ts_url = segment.absolute_uri
67.         for _ in range（MAX_RETRIES）:
68.             try:
69.                 response = self.session.get（ts_url, timeout=15）
70.                 response.raise_for_status（）
71.                 return index, response.content
72.             except requests.exceptions.RequestException:
73.                 continue
74.         print（f“分片 {index} 下载失败: {ts_url}“）
75.         return index, None
76.     def _download_and_merge（self, playlist, key）:
77.         segments = list（enumerate（playlist.segments））
78.         total_segments = len（segments）
79.         self.status_callback（f“开始下载 {total_segments} 个分片...“）
80.         # 使用字典来保证分片顺序
81.         ts_contents = {}
82.          
83.         with ThreadPoolExecutor（max_workers=MAX_WORKERS） as executor:
84.             futures = {executor.submit（self._download_segment, s）: s for s in segments}
85.             
86.             for i, future in enumerate（as_completed（futures））:
87.                 index, content = future.result（）
88.                 if content:
89.                     ts_contents[index] = content
90.                  
91.                 progress = （i + 1） / total_segments * 100
92.                 self.progress_callback（progress）
93.         if len（ts_contents） != total_segments:
94.             print（“警告：部分分片下载失败。“）
95.         self.status_callback（“下载完成，开始合并文件...“）
96.         with open（self.output_filename, ‘wb‘） as f:
97.             for i in range（total_segments）:
98.                 content = ts_contents.get（i）
99.                 if not content:
100.                     continue
101.                 if key:
102.                     iv = （playlist.media_sequence + i）.to_bytes（16, ‘big‘）
103.                     cipher = AES.new（key, AES.MODE_CBC, iv）
104.                     try:
105.                         decrypted_content = cipher.decrypt（content）
106.                         # 简单的 unpad，可能不适用于所有情况
107.                         unpad = lambda s: s[:-ord（s[len（s）-1:]）]
108.                         f.write（unpad（decrypted_content））
109.                     except Exception:
110.                          f.write（content） # 解密失败则直接写入
111.                 else:
112.                     f.write（content）
113. # --- GUI Application ---
114. class App（tk.Tk）:
115.     def __init__（self）:
116.         super（）.__init__（）
117.         self.title（“M3U8 下载器“）
118.         self.geometry（“500x200“）
119.         self.url_label = ttk.Label（self, text=“M3U8 链接:“）
120.         self.url_label.pack（pady=5）
121.         self.url_entry = ttk.Entry（self, width=60）
122.         self.url_entry.pack（pady=5, padx=10）
123.         self.download_button = ttk.Button（self, text=“下载“, command=self.start_download）
124.         self.download_button.pack（pady=10）
125.         self.progress = ttk.Progressbar（self, orient=“horizontal“, length=400, mode=“determinate“）
126.         self.progress.pack（pady=10）
127.         self.status_label = ttk.Label（self, text=“等待下载...“）
128.         self.status_label.pack（pady=5）
129.     def start_download（self）:
130.         m3u8_url = self.url_entry.get（）
131.         if not m3u8_url:
132.             messagebox.showwarning（“警告“, “请输入 M3U8 链接！“）
133.             return
134.         output_filename = filedialog.asksaveasfilename（
135.             title=“保存视频文件“,
136.             defaultextension=“.mp4“,
137.             filetypes=[（“MPEG-4“, “*.mp4“）, （“All Files“, “*.*“）]
138.         ）
139.         if not output_filename:
140.             return
141.         self.download_button.config（state=tk.DISABLED）
142.         self.progress[“value“] = 0
143.          
144.         downloader = Downloader（
145.             m3u8_url,
146.             output_filename,
147.             self.update_progress,
148.             self.update_status
149.         ）
150.          
151.         # 在新线程中运行下载
152.         threading.Thread（target=downloader.run, daemon=True）.start（）
153.     def update_progress（self, value）:
154.         self.progress[“value“] = value
155.         self.update_idletasks（）
156.     def update_status（self, text）:
157.         self.status_label.config（text=text）
158.         if “完成“ in text or “错误“ in text:
159.             self.download_button.config（state=tk.NORMAL）
160.         self.update_idletasks（）
161. if __name__ == “__main__“:
162.     app = App（）
163.     app.mainloop（）
164. [i]复制代码[/i]

复制代码

[/code]