Arquivos M3U8 funcionam como índices para transmissões baseadas em HLS (HTTP Live Streaming). Eles não contêm o vídeo em si, mas uma lista de pequenos segmentos no formato .ts. Muitas vezes, esses segmentos são protegidos por criptografia AES-128 para evitar downloads não autorizados. Este guia técnico detalha como automatizar o processo de captura, dsecriptografia e mesclagem desses arquivos em um único vídeo MP4 utilizando Python.

1. Estrutura do Arquivo M3U8

Um arquivo M3U8 é essencialmente um arquivo de texto que lista URLs de segmentos de vídeo. Para vídeos protegidos, o arquivo inclui a tag #EXT-X-KEY, que especifica o método de criptografia e a localização da chave de descriptografia.

import requests
import re
import os

def capturar_playlist(url_alvo, headers):
    requisicao = requests.get(url_alvo, headers=headers)
    requisicao.raise_for_status()
    return requisicao.text

def processar_m3u8(conteudo_raw):
    linhas = conteudo_raw.splitlines()
    info_criptografia = next((l for l in linhas if l.startswith("#EXT-X-KEY")), None)
    
    padrao = r"METHOD=(.*),URI=\"(.*)\""
    resultado = re.search(padrao, info_criptografia)
    
    if not resultado:
        raise ValueError("Não foi possível localizar os dados de criptografia.")
    
    metodo = resultado.group(1)
    url_chave = resultado.group(2)
    
    urls_segmentos = [l for l in linhas if l.endswith(".ts") and not l.startswith("#")]
    
    return metodo, url_chave, urls_segmentos

2. Descriptografia AES-128

A maioria dos fluxos HLS utiliza o modo CBC (Cipher Block Chaining). Para descriptografar os dados, precisamos da chave binária (obtida via URL) e, em alguns casos, de um Vetor de Inicialização (IV). Se o IV não estiver explícito na tag #EXT-X-KEY, ele geralmente corrseponde ao número sequencial do segmento.

from Crypto.Cipher import AES

def inicializar_descriptografador(chave_binaria):
    # Frequentemente, em fluxos M3U8, o IV pode ser omitido se for baseado no índice do segmento
    # Aqui utilizamos uma abordagem simplificada para o modo CBC
    return AES.new(chave_binaria, AES.MODE_CBC)

def salvar_segmento_decodificado(caminho_arquivo, dados_brutos, cifra):
    with open(caminho_arquivo, "wb") as f:
        # Nota: O conteúdo deve ser múltiplo de 16 bytes para AES
        f.write(cifra.decrypt(dados_brutos))

3. Dwonload Paralelo com Multiprocessing

Como vídeos podem ter centenas de fragmentos, o download sequencial é ineficiente. O uso de um pool de processos permite baixar e descriptografar múltiplos segmentos simultaneamente, reduzindo drasticamente o tempo total.

from multiprocessing import Pool
from tqdm import tqdm

def tarefa_download(dados):
    url_segmento, chave, pasta_destino, headers = dados
    nome_arquivo = os.path.join(pasta_destino, url_segmento.split('/')[-1])
    
    try:
        res = requests.get(url_segmento, headers=headers, timeout=10)
        cipher = AES.new(chave, AES.MODE_CBC)
        conteudo_decodificado = cipher.decrypt(res.content)
        
        with open(nome_arquivo, "wb") as f:
            f.write(conteudo_decodificado)
        return nome_arquivo
    except Exception as e:
        print(f"Erro no segmento {url_segmento}: {e}")
        return None

def gerenciador_downloads(lista_urls, chave, pasta, headers, threads=8):
    os.makedirs(pasta, exist_ok=True)
    parametros = [(u, chave, pasta, headers) for u in lista_urls]
    
    arquivos_locais = []
    with Pool(threads) as p:
        for resultado in tqdm(p.imap(tarefa_download, parametros), total=len(lista_urls)):
            if resultado:
                arquivos_locais.append(resultado)
    return arquivos_locais

4. Consolidação em MP4

Após o download de todos os fragmentos .ts, a etapa final consiste na concatenação binária. Como os arquivos .ts são projetados para serem contínuos, basta gravá-los em sequência dentro de um novo arquivo .mp4.

def consolidar_video(nome_final, lista_arquivos):
    with open(nome_final, "wb") as video_completo:
        for caminho in lista_arquivos:
            if os.path.exists(caminho):
                with open(caminho, "rb") as parte:
                    video_completo.write(parte.read())
    print(f"Vídeo exportado com sucesso: {nome_final}")

def limpar_temporarios(pasta):
    import shutil
    if os.path.exists(pasta):
        shutil.rmtree(pasta)

5. Fluxo de Execução Principal

O exemplo abaixo demonstra a integração de todos os módulos. É fundamental garantir que as URLs relativas no M3U8 sejam convertidas em URLs absolutas antes de iniciar o download.

if __name__ == "__main__":
    URL_BASE = "https://provedor-video.com/hls/"
    M3U8_URL = URL_BASE + "index.m3u8"
    HEADERS = {"User-Agent": "Mozilla/5.0"}
    PASTA_TMP = "temp_segments"

    # 1. Obter e processar a playlist
    conteudo_playlist = capturar_playlist(M3U8_URL, HEADERS)
    metodo, path_chave, segmentos = processar_m3u8(conteudo_playlist)

    # 2. Obter a chave de criptografia
    url_chave_completa = URL_BASE + path_chave
    chave_binaria = requests.get(url_chave_completa, headers=HEADERS).content

    # 3. Formatar URLs dos segmentos
    urls_completas = [URL_BASE + s for s in segmentos]

    # 4. Baixar em paralelo
    lista_ts = gerenciador_downloads(urls_completas, chave_binaria, PASTA_TMP, HEADERS)

    # 5. Mesclar e limpar
    consolidar_video("resultado_final.mp4", lista_ts)
    # limpar_temporarios(PASTA_TMP)