1. 项目概述:从M3U8到可播放视频的完整链路

最近在做一个视频素材收集的项目,经常遇到一些网站的视频流采用了M3U8格式,并且用AES-128进行了加密。浏览器开发者工具里能看到一个个 .ts 分片文件哗哗地加载,但直接下载下来的文件全是乱码,播放器根本打不开。这种“看得见却摸不着”的感觉,相信不少做爬虫或者多媒体处理的朋友都遇到过。于是,我花了一些时间,把整个解密和下载的流程彻底搞清楚了,并且用Python实现了一套完整的工具。今天就来手把手拆解这个过程,从原理到代码,让你不仅能下载,更能明白背后的每一步为什么这么做。

简单来说,我们的目标就是:拿到一个加密的M3U8链接,最终输出一个可以正常播放的MP4或TS文件。这中间涉及到几个关键环节:解析M3U8播放列表、提取解密密钥(Key)和初始化向量(IV)、下载加密的TS分片、用AES算法逐一分片解密,最后将所有分片合并成一个完整的视频文件。整个过程就像拼图,M3U8文件是图纸,TS分片是碎片,而AES密钥就是打开碎片包装的钥匙。下面,我们就按照这个逻辑,一步步深入。

2. 核心原理与前置知识拆解

在动手写代码之前,我们必须把几个核心概念和它们之间的关系理清楚。一知半解就去写代码,很容易在遇到异常情况时束手无策。

2.1 M3U8文件:视频流的“导航地图”

M3U8本质上是一个文本格式的播放列表,它是苹果公司推出的HTTP Live Streaming(HLS)协议的一部分。你可以用任何文本编辑器打开一个M3U8文件看看,它的结构非常直观。

一个典型的加密M3U8文件内容如下:

#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:10
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-PLAYLIST-TYPE:VOD
#EXT-X-KEY:METHOD=AES-128,URI="https://example.com/key.key",IV=0x1234567890abcdef1234567890abcdef
#EXTINF:10.0,
https://example.com/segment0.ts
#EXTINF:9.5,
https://example.com/segment1.ts
...
#EXT-X-ENDLIST

我们来解读几个关键标签:

  • #EXTM3U : 文件头,声明这是一个M3U播放列表。
  • #EXT-X-KEY 这是整个解密环节的灵魂所在 。它定义了视频流的加密方法。
    • METHOD=AES-128 : 指明加密算法是AES-128。这也是目前HLS加密最主流的方式。
    • URI="..." : 指向解密密钥文件的地址。这个文件通常是一个16字节(128位)的二进制文件。 我们的第一个任务就是下载它
    • IV=0x... : 初始化向量。AES加密除了密钥,通常还需要一个IV来增加安全性。如果这里没有指定,HLS规范默认使用媒体序列号( EXT-X-MEDIA-SEQUENCE )作为IV,这在我们的代码中需要特别注意处理。
  • #EXTINF: : 后面跟着一个TS分片的时长和URL。我们的主要下载目标就是这些 .ts 文件。

注意 : 一个M3U8文件中可能在不同位置出现多个 #EXT-X-KEY 标签,这意味着视频流的不同部分可能使用了不同的密钥。我们的示例代码为了简化,假设整个流使用同一个密钥。在实际处理中,你需要根据每个TS分片前面最近的KEY标签来决定使用哪个密钥解密,这会稍微复杂一些。

2.2 AES-128 CBC解密:解锁分片的“标准操作”

M3U8使用的AES-128,具体模式是CBC(Cipher Block Chaining,密码块链接)。这是一种分组加密模式,简单理解就是:第一个数据块的加密结果,会参与到下一个数据块的加密过程中,如此环环相扣。

解密时,我们需要三样东西:

  1. 密钥(Key) : 从 URI 指向的地址下载的16字节数据。
  2. 初始化向量(IV) : 来自 #EXT-X-KEY 标签中的 IV 属性。如果未指定,则使用分片的序列号(通常是一个整数)转换为16字节的十六进制格式(例如,序列号0对应的IV是 0x00000000000000000000000000000000 )。
  3. 密文 : 下载下来的 .ts 文件内容。

解密过程在Python中通常借助 cryptography 库或者 pycryptodome 库来完成。这两个库功能强大且稳定,比一些老旧或不再维护的库更值得推荐。我们将使用 pycryptodome ,因为它API清晰,且兼容性好。

2.3 网络请求与并发:效率的关键

一个高清视频可能由几百甚至上千个TS分片组成。如果用一个 for 循环顺序下载,速度会慢得令人发指。因此,引入并发下载是必须的。我们可以使用 asyncio + aiohttp 实现异步IO,或者使用 concurrent.futures ThreadPoolExecutor 实现线程池并发。前者在IO密集型任务中效率极高,后者代码相对简单直观。本文将展示线程池的方案,更容易理解且对新手友好。

3. 工具选型与环境搭建

工欲善其事,必先利其器。选择靠谱的库能避免很多底层坑。

  • HTTP请求库 requests 。简单易用,同步阻塞。对于并发下载,我们会用 ThreadPoolExecutor 来包装它。
  • AES解密库 pycryptodome 。这是 PyCrypto 的一个分支,持续维护,文档齐全。安装: pip install pycryptodome
  • 视频处理(可选) moviepy ffmpeg-python 。用于最后的合并与转码。最稳妥的方式是直接调用系统安装的FFmpeg命令行工具。我们将采用系统调用FFmpeg的方式,因为它最通用、最强大。
  • 进度显示 tqdm 。在命令行中显示漂亮的进度条,让你清楚知道下载和解密进行到哪一步。安装: pip install tqdm

确保你的系统已经安装了FFmpeg。可以在命令行输入 ffmpeg -version 检查。如果没有,请去FFmpeg官网下载并配置环境变量。

4. 分步实战:代码实现与深度解析

下面,我们按照逻辑顺序,一步步构建我们的解密下载器。我会对每一段关键代码进行详细解释,并说明其中容易踩坑的地方。

4.1 第一步:解析M3U8播放列表

这个函数负责从网络或本地文件读取M3U8内容,并从中提取出所有TS分片的URL、密钥信息以及IV。

import re
from urllib.parse import urljoin

def parse_m3u8(m3u8_url, m3u8_content=None):
    """
    解析M3U8文件内容。
    :param m3u8_url: M3U8文件的完整URL,用于拼接相对路径的TS和Key URL。
    :param m3u8_content: 可选的M3u8文件文本内容。如果为None,则从m3u8_url下载。
    :return: 返回一个字典,包含key_url, iv, 和ts_urls列表。
    """
    if m3u8_content is None:
        import requests
        resp = requests.get(m3u8_url, headers={'User-Agent': 'Mozilla/5.0'})
        resp.raise_for_status()
        content = resp.text
    else:
        content = m3u8_content

    lines = content.split('\n')
    
    ts_urls = []
    key_url = None
    iv = None
    base_url = m3u8_url.rsplit('/', 1)[0] + '/' if '/' in m3u8_url else ''

    i = 0
    while i < len(lines):
        line = lines[i].strip()
        if line.startswith('#EXT-X-KEY'):
            # 解析METHOD, URI, IV
            # 示例: #EXT-X-KEY:METHOD=AES-128,URI="https://xx.com/key.key",IV=0x123...
            match_method = re.search(r'METHOD=([^,]+)', line)
            if match_method and match_method.group(1) == 'AES-128':
                match_uri = re.search(r'URI="([^"]+)"', line)
                if match_uri:
                    key_url = match_uri.group(1)
                    # 处理相对路径
                    if not key_url.startswith(('http://', 'https://')):
                        key_url = urljoin(m3u8_url, key_url)
                match_iv = re.search(r'IV=([^,]+)', line)
                if match_iv:
                    iv_hex = match_iv.group(1)
                    # 去除0x前缀,并转换为字节
                    if iv_hex.startswith('0x') or iv_hex.startswith('0X'):
                        iv_hex = iv_hex[2:]
                    # 确保IV是32位十六进制字符串(16字节)
                    iv_hex = iv_hex.zfill(32)
                    iv = bytes.fromhex(iv_hex)
        elif line.startswith('#EXTINF'):
            # 下一行就是TS文件的URL
            i += 1
            if i < len(lines):
                ts_url = lines[i].strip()
                if ts_url and not ts_url.startswith('#'):
                    if not ts_url.startswith(('http://', 'https://')):
                        ts_url = urljoin(m3u8_url, ts_url)
                    ts_urls.append(ts_url)
        i += 1

    if not ts_urls:
        raise ValueError("未在M3U8文件中找到任何TS分片URL。")

    return {
        'ts_urls': ts_urls,
        'key_url': key_url,
        'iv': iv,
        'base_url': base_url
    }

关键点解析与避坑指南:

  1. 相对路径处理 : M3U8文件里的URI和TS URL经常是相对路径。 urljoin 函数能很好地处理这个问题,它需要基准URL(即M3u8文件自身的URL)。这就是为什么函数参数需要 m3u8_url
  2. IV的解析 : IV可能以 0x 开头,也可能直接是十六进制字符串。代码中需要统一处理,去掉 0x 前缀,并将字符串转换为字节对象( bytes.fromhex )。 特别注意 :AES-128 CBC模式要求的IV长度是16字节,所以转换后的字节长度必须是16。如果原始IV字符串不足32个十六进制字符(16字节*2),需要用 zfill(32) 在左侧补零。
  3. 未指定IV的情况 : 如果 #EXT-X-KEY 标签中没有 IV 属性,那么解密时应该使用媒体序列号( EXT-X-MEDIA-SEQUENCE )作为IV,通常第一个分片序列号为0。我们的代码目前没有处理这种情况,这是一个需要根据实际情况完善的细节。一个简单的处理方式是:如果 iv None ,则在解密每个分片时,用该分片的索引(从0开始)生成一个16字节的IV(例如,索引i的IV为 i.to_bytes(16, 'big') )。

4.2 第二步:下载密钥和TS分片

下载部分我们使用线程池来加速。这里定义一个通用的下载函数。

import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
import os

def download_file(url, save_path, headers=None):
    """下载单个文件到指定路径"""
    if headers is None:
        headers = {'User-Agent': 'Mozilla/5.0'}
    try:
        resp = requests.get(url, headers=headers, stream=True, timeout=30)
        resp.raise_for_status()
        with open(save_path, 'wb') as f:
            for chunk in resp.iter_content(chunk_size=8192):
                if chunk:
                    f.write(chunk)
        return True, save_path
    except Exception as e:
        return False, f"下载失败 {url}: {e}"

def download_ts_segments(ts_urls, key_url, iv, output_dir='./downloads', max_workers=10):
    """
    并发下载所有TS分片和密钥。
    :return: 返回成功下载的TS文件路径列表,以及密钥的二进制内容。
    """
    os.makedirs(output_dir, exist_ok=True)
    
    # 1. 下载密钥
    key_path = os.path.join(output_dir, 'encryption.key')
    success, msg = download_file(key_url, key_path)
    if not success:
        raise RuntimeError(f"密钥下载失败: {msg}")
    with open(key_path, 'rb') as f:
        key_data = f.read()
    if len(key_data) != 16:
        # 有些服务器的key文件可能包含多余字符(如换行符),我们只取前16字节
        print(f"警告:密钥文件长度是{len(key_data)}字节,非标准16字节。将取前16字节。")
        key_data = key_data[:16]
    
    # 2. 并发下载TS分片
    ts_paths = []
    tasks = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有下载任务
        future_to_url = {}
        for idx, ts_url in enumerate(ts_urls):
            ts_filename = f'segment_{idx:05d}.ts'
            ts_path = os.path.join(output_dir, ts_filename)
            future = executor.submit(download_file, ts_url, ts_path)
            future_to_url[future] = (ts_url, ts_path)
        
        # 使用tqdm显示进度
        from tqdm import tqdm
        for future in tqdm(as_completed(future_to_url), total=len(future_to_url), desc="下载TS分片"):
            ts_url, ts_path = future_to_url[future]
            success, msg = future.result()
            if success:
                ts_paths.append(ts_path)
            else:
                print(f"下载失败(可重试): {ts_url} -> {msg}")
                # 这里可以加入重试逻辑
    
    # 按索引排序,确保顺序正确
    ts_paths.sort()
    return ts_paths, key_data, iv

实操心得:

  1. 密钥长度校验 : 务必检查下载的密钥文件是否为16字节。我遇到过服务器返回的key文件末尾带了一个换行符( \n ),导致长度变成17字节,直接用于解密会报错。所以代码中做了截取前16字节的处理。
  2. 文件命名与排序 : TS分片下载是并发的,完成顺序不确定。我们按照索引 idx 来命名文件(如 segment_00000.ts ),下载完成后对路径列表进行排序,就能保证分片的原始顺序,这对后续合并至关重要。
  3. 错误处理与重试 : 网络请求总有可能失败。代码中只打印了错误信息,在生产环境中,你应该为 download_file 函数添加重试机制(例如,使用 tenacity 库),并对失败的分片进行记录和后续重试。
  4. User-Agent : 有些服务器会检查 User-Agent ,模仿浏览器的UA可以避免被简单的反爬策略拦截。

4.3 第三步:核心AES解密与分片处理

这是最核心的一步,我们将每个加密的TS分片读入内存,用AES-128 CBC模式解密,然后保存为新的文件。

from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad

def decrypt_ts_file(encrypted_ts_path, decrypted_ts_path, key, iv, segment_index=0):
    """
    解密单个TS文件。
    :param segment_index: 分片索引,用于在IV未指定时生成IV。
    """
    # 处理IV:如果传入的iv为None,则使用segment_index生成
    if iv is None:
        # HLS规范:使用媒体序列号作为IV,大端序,填充至16字节
        iv_to_use = segment_index.to_bytes(16, 'big')
    else:
        iv_to_use = iv
    
    cipher = AES.new(key, AES.MODE_CBC, iv_to_use)
    
    with open(encrypted_ts_path, 'rb') as f_in:
        encrypted_data = f_in.read()
    
    # 解密
    try:
        decrypted_data = cipher.decrypt(encrypted_data)
        # 注意:TS分片是流式加密,通常不需要padding,所以不调用unpad。
        # 如果解密后末尾有多余填充字节导致播放问题,可以尝试以下操作:
        # 但大多数情况不需要,直接写入即可。
        # decrypted_data = unpad(decrypted_data, AES.block_size)
    except Exception as e:
        raise RuntimeError(f"解密文件 {encrypted_ts_path} 失败: {e}")
    
    with open(decrypted_ts_path, 'wb') as f_out:
        f_out.write(decrypted_data)

def batch_decrypt_ts(ts_paths, key, iv, output_dir='./decrypted'):
    """批量解密所有TS分片"""
    os.makedirs(output_dir, exist_ok=True)
    decrypted_paths = []
    
    from tqdm import tqdm
    for idx, ts_path in enumerate(tqdm(ts_paths, desc="解密TS分片")):
        ts_filename = os.path.basename(ts_path)
        decrypted_filename = f'decrypted_{ts_filename}'
        decrypted_path = os.path.join(output_dir, decrypted_filename)
        
        decrypt_ts_file(ts_path, decrypted_path, key, iv, segment_index=idx)
        decrypted_paths.append(decrypted_path)
    
    return decrypted_paths

深度解析与常见陷阱:

  1. Padding问题 : 这是AES解密中最容易出错的地方。CBC模式要求数据长度是16字节(AES块大小)的整数倍,不足的部分需要填充(Padding)。 但是,在HLS的TS流加密中,通常不对整个TS文件进行填充,而是对每个16字节的块进行加密 。这意味着解密后的数据长度应该和加密前完全一致。如果你在解密后调用 unpad ,很可能会得到一个 ValueError: Padding is incorrect. 错误。 因此,对于TS文件,解密后直接写入即可,切勿进行 unpad 操作。 如果某些特殊源确实有填充,你会发现在文件末尾有多余的不可读字节,那时再考虑 unpad
  2. IV的生成 : 代码中演示了当IV未提供时,如何使用分片索引生成IV。 segment_index.to_bytes(16, 'big') 会将整数转换为16字节的大端序表示。例如,索引0就是16个 \x00 务必与M3U8文件的实际规范保持一致 ,有些实现可能使用小端序。
  3. 内存考虑 : 对于非常大的TS分片(几十MB以上),一次性读入内存解密可能占用过高内存。可以考虑分块读取和解密(例如每次读取 AES.block_size * 1024 字节),但实现会复杂一些。对于绝大多数网络视频流,TS分片通常在2-10MB,直接处理问题不大。

4.4 第四步:合并与转码为最终视频

解密后的TS分片是一堆小文件,我们需要将它们合并成一个文件。最可靠的工具是FFmpeg。

import subprocess

def merge_ts_to_mp4(decrypted_ts_paths, output_mp4_path='./output_final.mp4'):
    """
    使用FFmpeg将解密后的TS文件列表合并为MP4。
    """
    # 方法1:使用concat协议(推荐,速度快)
    # 创建一个文件列表
    list_file_path = './file_list.txt'
    with open(list_file_path, 'w', encoding='utf-8') as f:
        for ts_path in decrypted_ts_paths:
            # 对文件路径进行转义,防止特殊字符问题
            f.write(f"file '{ts_path}'\n")
    
    ffmpeg_cmd = [
        'ffmpeg',
        '-f', 'concat',
        '-safe', '0',  # 允许任意文件路径
        '-i', list_file_path,
        '-c', 'copy',  # 流复制,不重新编码,速度极快
        '-y',  # 覆盖输出文件
        output_mp4_path
    ]
    
    # 方法2:直接拼接二进制(不推荐,可能遇到编码问题)
    # with open(output_mp4_path, 'wb') as out_f:
    #     for ts_path in decrypted_ts_paths:
    #         with open(ts_path, 'rb') as in_f:
    #             out_f.write(in_f.read())
    
    try:
        print("正在使用FFmpeg合并视频...")
        result = subprocess.run(ffmpeg_cmd, check=True, capture_output=True, text=True)
        print("FFmpeg 输出:", result.stdout)
        if result.stderr:
            print("FFmpeg 警告/错误:", result.stderr)
        print(f"视频合并成功: {output_mp4_path}")
        # 清理临时列表文件
        os.remove(list_file_path)
        return True
    except subprocess.CalledProcessError as e:
        print(f"FFmpeg合并失败,返回码: {e.returncode}")
        print(f"标准错误输出: {e.stderr}")
        return False
    except FileNotFoundError:
        print("错误:未找到FFmpeg命令。请确保FFmpeg已安装并添加到系统环境变量PATH中。")
        return False

为什么用FFmpeg而不是直接二进制拼接? 直接按顺序将TS文件二进制拼接在一起,理论上对于简单的TS流是可行的。但是,TS格式包含复杂的包头、时间戳(PTS/DTS)等信息。直接拼接可能导致:

  • 时间戳不连续,播放器跳帧或卡顿。
  • 音视频不同步。
  • 某些播放器无法识别拼接后的文件。

FFmpeg的 concat 协议会解析每个TS文件的流信息,并重新生成一个合规的容器文件(如MP4),确保播放兼容性。 -c copy 参数表示“流复制”,即不进行耗时的重新编码,只是重新封装,所以速度非常快。

5. 完整脚本组装与使用示例

将上述所有函数整合起来,并添加一个主函数和参数解析,我们就得到了一个完整的命令行工具。

# save as: m3u8_downloader.py
import argparse
import os
import sys

def main():
    parser = argparse.ArgumentParser(description='下载并解密AES-128加密的M3U8视频流。')
    parser.add_argument('m3u8_url', help='M3U8文件的URL地址')
    parser.add_argument('-o', '--output', default='output.mp4', help='最终输出的MP4文件名(默认: output.mp4)')
    parser.add_argument('-d', '--temp-dir', default='./temp_video', help='临时文件存放目录(默认: ./temp_video)')
    parser.add_argument('-w', '--workers', type=int, default=10, help='并发下载线程数(默认: 10)')
    
    args = parser.parse_args()
    
    m3u8_url = args.m3u8_url
    final_output = args.output
    temp_dir = args.temp_dir
    max_workers = args.workers
    
    download_dir = os.path.join(temp_dir, 'encrypted')
    decrypt_dir = os.path.join(temp_dir, 'decrypted')
    
    try:
        print(f"[1/4] 正在解析M3U8文件: {m3u8_url}")
        m3u8_info = parse_m3u8(m3u8_url)
        ts_urls = m3u8_info['ts_urls']
        key_url = m3u8_info['key_url']
        iv = m3u8_info['iv']
        
        if not key_url:
            print("警告:M3U8文件未包含AES-128加密密钥信息。尝试直接下载...")
            # 这里可以改为直接下载非加密流,逻辑需调整
            return
        
        print(f"解析完成。共发现 {len(ts_urls)} 个TS分片。")
        print(f"密钥URL: {key_url}")
        print(f"IV: {iv}")
        
        print(f"[2/4] 正在下载密钥和TS分片到目录: {download_dir}")
        ts_paths, key_data, iv = download_ts_segments(ts_urls, key_url, iv, download_dir, max_workers)
        print(f"下载完成。密钥长度: {len(key_data)} 字节。")
        
        print(f"[3/4] 正在解密TS分片到目录: {decrypt_dir}")
        decrypted_paths = batch_decrypt_ts(ts_paths, key_data, iv, decrypt_dir)
        print(f"解密完成。")
        
        print(f"[4/4] 正在合并视频为: {final_output}")
        success = merge_ts_to_mp4(decrypted_paths, final_output)
        if success:
            print("\n🎉 全部流程完成!")
            # 可选:清理临时目录
            # import shutil
            # shutil.rmtree(temp_dir)
        else:
            print("\n❌ 视频合并失败。")
            sys.exit(1)
            
    except Exception as e:
        print(f"\n❌ 程序运行出错: {e}", file=sys.stderr)
        import traceback
        traceback.print_exc()
        sys.exit(1)

if __name__ == '__main__':
    main()

使用方式: 在命令行中执行:

python m3u8_downloader.py "https://your-video-site.com/path/to/playlist.m3u8" -o "我的视频.mp4"

6. 常见问题排查与进阶技巧

在实际操作中,你几乎一定会遇到各种问题。下面是我踩过坑后总结的排查清单。

6.1 问题排查速查表

问题现象 可能原因 排查步骤与解决方案
解密失败,报 Padding is incorrect 1. 密钥错误或长度不对。
2. IV不正确。
3. 源文件本身有填充,但代码未处理。
1. 打印并确认密钥内容(可输出前几位hex)。检查密钥URL是否能直接访问,内容是否为16字节。
2. 确认IV解析正确。尝试使用分片索引生成IV(将 iv 参数设为 None )。
3. 对于TS流,首先尝试注释掉 unpad 代码 。如果问题依旧,尝试对解密后数据的最后16字节进行检查。
解密后的TS文件无法播放,或合并后视频花屏、卡顿 1. 解密过程本身错误(密钥/IV错)。
2. TS分片下载顺序错乱或损坏。
3. FFmpeg合并时编码问题。
1. 用 hexdump xxd 查看加密和解密后文件头部,正常TS文件头是 0x47 (同步字节)。如果解密后开头不是 0x47 ,说明解密失败。
2. 确保 ts_paths 列表是按正确顺序排序的。可以手动用播放器(如VLC)打开单个解密后的TS文件测试。
3. 尝试不用 -c copy ,改用 -c:v libx264 -c:a aac 进行转码(速度慢但兼容性最强)。
下载TS分片时大量失败(403/404) 1. 网站有反爬机制(如Referer、Cookie验证)。
2. 分片URL过期(动态生成)。
3. 请求头不足。
1. 使用浏览器开发者工具,复制下载TS分片时的完整请求头(包括 Referer , Cookie , Origin 等),在 download_file 函数中模拟。
2. M3U8文件可能是动态更新的直播流。我们的代码主要针对VOD(点播)流,直播流需要循环请求M3U8并下载新分片,逻辑更复杂。
3. 增加 User-Agent ,并尝试添加 Accept , Accept-Encoding 等通用头。
合并后的MP4没有声音或音画不同步 1. 原始TS流包含多路音频或特殊编码。
2. FFmpeg流复制时时间戳处理问题。
1. 使用 ffprobe 分析原始TS文件流信息: ffprobe -i segment_00000.ts 。查看音频编码格式(是否是AAC/MP3)。
2. 尝试在FFmpeg命令中加入 -fflags +genpts 来生成时间戳。或者放弃 -c copy ,使用转码参数重新编码音视频流。
程序报错 Crypto.Cipher.AES 不存在 pycryptodome 安装或导入问题。 确认安装的是 pycryptodome ,不是 pycrypto 。导入语句应为 from Crypto.Cipher import AES 。如果还不行,尝试 pip uninstall pycrypto pycryptodome 然后重新 pip install pycryptodome

6.2 进阶技巧与优化建议

  1. 断点续传与重试 : 对于大型视频,在网络不稳定时,为 download_file 函数添加重试机制和断点续传功能非常有用。可以使用 requests stream=True headers={'Range': f'bytes={start}-'} 来实现断点续传。
  2. 代理支持 : 如果目标网站有地域限制,可以在 requests.get() 中设置 proxies 参数。
  3. 自适应码率(Adaptive Bitrate) : 很多M3U8文件是主列表(Master Playlist),里面包含了多个不同清晰度(如720p, 1080p)的子播放列表URL。你需要先解析主列表,选择其中一个子列表(通常是分辨率最高的那个)再进行上述流程。
  4. 性能优化 : 对于超多分片(>1000), ThreadPoolExecutor 可能不是最优选,可以考虑使用 asyncio + aiohttp ,能更高效地管理大量并发网络连接。
  5. 内存优化 : 解密和合并大文件时,可以采用流式处理。即下载一个分片,解密后立即写入最终的合并文件(或临时文件),然后删除该分片,这样可以显著降低磁盘和内存占用。

这个项目从原理到实现,基本覆盖了处理AES-128加密M3U8流的全链路。代码本身是一个可用的起点,但真实网络环境复杂,你需要根据遇到的具体问题,灵活运用排查技巧,并对代码进行相应的调整和增强。

Logo

这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容,并连接云服务、办公空间等稀缺资源,助你专注创造,无忧运营。

更多推荐