### Python实现TS文件解密与多线程下载 以下是使用Python实现TS文件解密并结合多线程技术进行下载的完整解决方案。该方案包括解析M3U8文件、获取加密密钥、解密TS文件以及多线程下载。 #### 1. 解析M3U8文件 M3U8文件是流媒体播放列表，包含多个TS文件的URL地址以及加密信息。通过读取M3U8文件中的`#EXT-X-KEY`字段，可以获取加密方式和密钥地址[^1]。 ```python import requests from concurrent.futures import ThreadPoolExecutor import os import base64 import m3u8 import urllib.parse def parse_m3u8(url): response = requests.get(url) if response.status_code == 200: playlist = m3u8.loads(response.text) return playlist else: raise Exception("Failed to fetch M3U8 playlist") ``` #### 2. 获取加密密钥 如果M3U8文件中存在`#EXT-X-KEY`字段，则表示TS文件是加密的。需要从指定的URI地址下载密钥[^1]。 ```python def get_key(playlist, base_url): key_info = playlist.keys[-1] if key_info and key_info.uri: key_url = urllib.parse.urljoin(base_url, key_info.uri) response = requests.get(key_url) if response.status_code == 200: return response.content else: raise Exception("Failed to fetch encryption key") return None ``` #### 3. 多线程下载TS文件 使用`ThreadPoolExecutor`实现多线程下载TS文件，提高下载效率[^2]。 ```python def download_ts_file(ts_url, output_path, headers): try: response = requests.get(ts_url, headers=headers) if response.status_code == 200: with open(output_path, 'wb') as f: f.write(response.content) return True else: print(f"Failed to download {ts_url}") return False except Exception as e: print(f"Error downloading {ts_url}: {e}") return False def download_ts_files(ts_urls, base_output_dir, headers): os.makedirs(base_output_dir, exist_ok=True) with ThreadPoolExecutor(max_workers=10) as executor: futures = [] for index, ts_url in enumerate(ts_urls): output_path = os.path.join(base_output_dir, f"{index:04d}.ts") futures.append(executor.submit(download_ts_file, ts_url, output_path, headers)) results = [future.result() for future in futures] return all(results) ``` #### 4. 解密TS文件 如果TS文件是加密的，则需要使用AES算法对每个TS文件进行解密。 ```python from Crypto.Cipher import AES def decrypt_ts_file(input_path, output_path, key): with open(input_path, 'rb') as encrypted_file: cipher_text = encrypted_file.read() cipher = AES.new(key, AES.MODE_CBC, iv=cipher_text[:16]) decrypted_data = cipher.decrypt(cipher_text[16:]) with open(output_path, 'wb') as decrypted_file: decrypted_file.write(decrypted_data) ``` #### 5. 合并TS文件 下载并解密所有TS文件后，将它们合并为一个完整的视频文件[^3]。 ```python def merge_ts_files(input_dir, output_file): ts_files = sorted([f for f in os.listdir(input_dir) if f.endswith('.ts')]) with open(output_file, 'wb') as merged_file: for ts_file in ts_files: with open(os.path.join(input_dir, ts_file), 'rb') as fragment: merged_file.write(fragment.read()) ``` #### 6. 完整流程示例 以下是一个完整的流程示例，展示如何解析M3U8文件、下载TS文件、解密并合并视频。 ```python def main(m3u8_url, output_dir, headers): playlist = parse_m3u8(m3u8_url) base_url = os.path.dirname(m3u8_url) # 获取密钥 key = get_key(playlist, base_url) # 下载TS文件 ts_urls = [urllib.parse.urljoin(base_url, seg.uri) for seg in playlist.segments] ts_output_dir = os.path.join(output_dir, "ts_files") download_success = download_ts_files(ts_urls, ts_output_dir, headers) if not download_success: print("Some TS files failed to download") return # 解密TS文件 if key: decrypted_output_dir = os.path.join(output_dir, "decrypted_ts_files") os.makedirs(decrypted_output_dir, exist_ok=True) for ts_file in os.listdir(ts_output_dir): input_path = os.path.join(ts_output_dir, ts_file) output_path = os.path.join(decrypted_output_dir, ts_file) decrypt_ts_file(input_path, output_path, key) ts_output_dir = decrypted_output_dir # 合并TS文件 final_output_file = os.path.join(output_dir, "final_video.mp4") merge_ts_files(ts_output_dir, final_output_file) print(f"Video saved to {final_output_file}") if __name__ == "__main__": m3u8_url = "https://example.com/video.m3u8" output_dir = "./output" headers = {"User-Agent": "Mozilla/5.0"} main(m3u8_url, output_dir, headers) ``` ### 注意事项 - 确保安装了必要的依赖库：`requests`, `m3u8`, `pycryptodome`。 - 如果M3U8文件或TS文件受到DRM保护，则无法通过上述方法解密。 - 在实际应用中，需遵守相关法律法规及网站的使用条款。