只处理有权访问的流

M3U8/HLS 很适合学习流媒体结构：一个 playlist 指向很多 .ts 或 .m4s 片段，可能还有多清晰度、嵌套 playlist 和 AES-128 加密信息。这里的边界也很明确：只处理自己有权访问和保存的内容，不绕过登录、付费、DRM 或版权限制。

Shark-Tools 里既有 Qt 播放器，也有早期下载脚本。把真实站点和资源地址去掉后，剩下的其实是一条很通用的工程链路。

解析嵌套 playlist

有些页面里拿到的 m3u8 不是最终片段列表，而是再指向一个子 playlist。第一步要解析相对路径：

from urllib.parse import urljoin

def resolve_nested_playlist(m3u8_url: str, content: str) -> str:
    lines = [
        line.strip()
        for line in content.splitlines()
        if line.strip() and not line.startswith("#")
    ]
    nested = next((line for line in lines if ".m3u8" in line), "")
    if not nested:
        return m3u8_url
    return nested if nested.startswith("http") else urljoin(m3u8_url, nested)

这一步看似简单，但很关键。很多下载失败不是网络问题，而是把 master playlist 当成 media playlist 处理了。

读取 EXT-X-KEY

普通 AES-128 HLS 会在 playlist 里出现 #EXT-X-KEY。工程上应该把 key URL、IV、是否加密这些信息放进一个独立结构里：

import re
from urllib.parse import urljoin

def parse_hls_key(m3u8_url: str, content: str) -> dict:
    info = {
        "encrypted": False,
        "key_url": None,
        "iv": bytes(16),
    }

    match = re.search(r'#EXT-X-KEY:METHOD=AES-128,URI="(.*?)"', content)
    if not match:
        return info

    key_uri = match.group(1)
    info["encrypted"] = True
    info["key_url"] = key_uri if key_uri.startswith("http") else urljoin(m3u8_url, key_uri)

    iv_match = re.search(r"IV=0x([0-9a-fA-F]+)", content)
    if iv_match:
        info["iv"] = bytes.fromhex(iv_match.group(1).zfill(32))

    return info

如果没有 key，代码应该明确标记“加密但缺 key”，而不是偷偷保存一堆无法播放的片段。

片段下载与解密

片段下载最容易写成无上限并发。我的经验是分批下载、保留序号、失败可重试，但不要无限打：

import asyncio
from Crypto.Cipher import AES

def decrypt_segment(data: bytes, key: bytes, iv: bytes) -> bytes:
    cipher = AES.new(key, AES.MODE_CBC, iv)
    return cipher.decrypt(data)

async def download_segment(session, url, index, *, key=None, iv=None):
    async with session.get(url, timeout=30) as response:
        if response.status in {401, 403}:
            raise PermissionError("访问被拒绝，停止下载")
        response.raise_for_status()
        data = await response.read()

    if key:
        data = decrypt_segment(data, key, iv or bytes(16))

    return index, data

async def download_in_batches(tasks, batch_size=20):
    results = []
    for offset in range(0, len(tasks), batch_size):
        batch = tasks[offset:offset + batch_size]
        results.extend(await asyncio.gather(*batch, return_exceptions=True))
    return results

我不喜欢“失败就疯狂重试一万次”的写法。遇到 401/403 应该停止，遇到网络抖动可以少量重试，遇到片段缺失则让用户决定是否继续合并。

合并清单

FFmpeg concat 模式需要一个顺序文件。只把成功下载且非空的片段写进去，能避免把坏文件硬塞进合并流程：

from pathlib import Path

def write_concat_list(segment_dir: Path, count: int) -> Path:
    list_file = segment_dir / "file.txt"
    with list_file.open("w", encoding="utf-8") as handle:
        for index in range(count):
            segment = segment_dir / f"{index:05d}.ts"
            if segment.exists() and segment.stat().st_size > 0:
                handle.write(f"file '{segment.name}'\n")
    return list_file

合并不是下载的结束，而是质量检查的开始。片段数量、文件大小、时长、音视频流信息都值得再校验一次。

播放器侧的小结

如果只是想预览，Qt 的 QMediaPlayer 可以直接吃一些 MP4 或 m3u8 链接，但对不同平台 HLS 的兼容性不稳定。播放器适合“快速验证链接能不能播”，下载器适合“分析 playlist 和片段结构”，两者不要混成一个大函数。

我现在更倾向于把 HLS 处理拆成四个模块：playlist 解析、key 解析、片段下载、合并/校验。每一层都有输入输出，后面接 GUI、CLI 或 Agent 工具都会更舒服。