Introducción al Protocolo HLS y Extracción de Contenido
El streaming moderno basado en HTTP Live Streaming (HLS) ha revolucionado la distribución de contenido multimedia. A diferencia de la transmisión tradicional de archivos MP4 completos, HLS fragmenta el contenido en múltiples segmantos de video cortos (normalmente de 2-10 segundos) codificados en formato TS. Estos segmentos se indexan mediante un archivo de texto M3U8, que actúa como guía para el reproudctor cliente.
El proceso de adquisición de contenido HLS implica cuatro fases principales: obtención del manifest M3U8, descarga concurrente de los segmentos TS, procesamiento de cifrado (si aplica) y fusión final en un archivo continuo.
Ventajas del Streaming Adaptativo
La adopción masiva de HLS se debe a sus características técnicas superiores:
- Infraestructura simplificada: No requiere servidores de streaming dedicados; cualquier servidor web estándar puede distribuir los segmentos.
- Inicio rápido: La reproducción comienza tras descargar solo unos pocos segmentos iniciales, reduciendo la latencia percibida.
- Calidad adaptativa: El cliente selecciona automáticamente la calidad de video según el ancho de banda disponible, cambiando entre diferentes representaciones sin interrupciones.
Anatomía del Archivo M3U8
El formato M3U8 es una extensión de M3U adaptada para UTF-8. Es fundamental comprender sus directivas para implementar un extractor funcional:
#EXTM3U // Identificador obligatorio del formato
#EXT-X-VERSION:3 // Versión del protocolo
#EXT-X-TARGETDURATION:10 // Duración máxima de cada segmento
#EXT-X-MEDIA-SEQUENCE:0 // Número de secuencia inicial
#EXT-X-KEY:METHOD=AES-128,URI="key.key",IV=0x... // Parámetros de cifrado
#EXTINF:9.009, // Duración del segmento siguiente
segment_001.ts // URL relativa del segmento
#EXT-X-ENDLIST // Fin del playlist
La directiva #EXT-X-KEY es crucial: indica el método de cifrado (generalmente AES-128), la URI donde obtener la clave de descifrado y el vector de inicialización (IV) opcional.
Desencriptado de Segmentos
Cuando el contenido está protegido, cada segmento TS se cifra mediante AES-128 en modo CBC. El sistema requiere:
- Descargar la clave de 16 bytes desde la URI especificada
- Obtener el vector de inicialización (16 bytes en hexadecimal)
- Aplicar el descifrado a cada segmento individual antes de la fusión
Si no se especifica IV, se utiliza un vector nulo (16 bytes a cero) por convención.
Implementación Práctica en Python
A continuación se presenta una solución completa que utiliza programación asíncrona para maximizar el rendimiento:
import asyncio
import aiohttp
import aiofiles
import requests
import re
import os
from urllib.parse import urljoin
from Crypto.Cipher import AES
from typing import List, Tuple, Optional
# Configuración del sistema
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
MAX_CONCURRENT = 80
SEGMENT_DIR = "./segments_raw"
DECRYPTED_DIR = "./segments_ready"
class VideoExtractor:
def __init__(self, m3u8_url: str):
self.m3u8_url = m3u8_url
self.base_path = "/".join(m3u8_url.split("/")[:-1]) + "/"
self.encryption_key: Optional[bytes] = None
self.iv_vector: Optional[bytes] = None
def retrieve_playlist(self) -> str:
"""Obtiene y almacena el archivo M3U8"""
response = requests.get(self.m3u8_url, headers=HEADERS)
playlist_data = response.text
with open("stream.m3u8", "w", encoding="utf-8") as f:
f.write(playlist_data)
return playlist_data
def extract_crypto_params(self, playlist_content: str):
"""Extrae parámetros de cifrado del playlist"""
pattern = r'#EXT-X-KEY:METHOD=(\w+),URI="([^"]+)"(?:,IV=(\w+))?'
match = re.search(pattern, playlist_content)
if match:
method, key_uri, iv_hex = match.groups()
if method == "AES-128":
# Descargar clave
key_url = urljoin(self.m3u8_url, key_uri)
self.encryption_key = requests.get(key_url, headers=HEADERS).content
# Procesar IV
if iv_hex:
self.iv_vector = bytes.fromhex(iv_hex.replace("0x", ""))
else:
self.iv_vector = b"\x00" * 16
print(f"🔐 Cifrado detectado: {method}")
async def fetch_segment(self, session: aiohttp.ClientSession,
segment_url: str, filename: str,
semaphore: asyncio.Semaphore) -> bool:
"""Descarga un segmento individual"""
async with semaphore:
output_path = os.path.join(SEGMENT_DIR, filename)
for retry in range(3):
try:
async with session.get(segment_url, headers=HEADERS) as response:
content = await response.read()
async with aiofiles.open(output_path, "wb") as f:
await f.write(content)
print(f"✅ {filename}")
return True
except Exception as error:
print(f"❌ {filename} (reintento {retry+1}): {error}")
await asyncio.sleep(0.5)
return False
async def download_segments(self, playlist_content: str):
"""Coordina descarga masiva de segmentos"""
# Preparar directorios
os.makedirs(SEGMENT_DIR, exist_ok=True)
os.makedirs(DECRYPTED_DIR, exist_ok=True)
# Extraer URLs
segments: List[Tuple[str, str]] = []
for line in playlist_content.splitlines():
line = line.strip()
if line and not line.startswith("#"):
full_url = urljoin(self.m3u8_url, line)
segments.append((full_url, os.path.basename(line)))
# Descargar con concurrencia limitada
async with aiohttp.ClientSession() as session:
semaphore = asyncio.Semaphore(MAX_CONCURRENT)
tasks = [
self.fetch_segment(session, url, name, semaphore)
for url, name in segments
]
results = await asyncio.gather(*tasks)
successful = sum(1 for r in results if r)
print(f"\nDescargados {successful}/{len(segments)} segmentos")
async def decrypt_single(self, filename: str):
"""Descifra un archivo TS"""
if not self.encryption_key:
# Copiar sin descifrar
src = os.path.join(SEGMENT_DIR, filename)
dst = os.path.join(DECRYPTED_DIR, filename)
async with aiofiles.open(src, "rb") as f_in, \
aiofiles.open(dst, "wb") as f_out:
await f_out.write(await f_in.read())
return
input_file = os.path.join(SEGMENT_DIR, filename)
output_file = os.path.join(DECRYPTED_DIR, filename)
async with aiofiles.open(input_file, "rb") as f:
encrypted = await f.read()
# Descifrar
cipher = AES.new(self.encryption_key, AES.MODE_CBC, self.iv_vector)
decrypted = cipher.decrypt(encrypted)
async with aiofiles.open(output_file, "wb") as f:
await f.write(decrypted)
print(f"🔓 {filename}")
async def decrypt_all(self, playlist_content: str):
"""Descifra todos los segmentos"""
if not self.encryption_key:
print("No se detectó cifrado, omitiendo descifrado")
return
# Obtener lista de archivos
filenames = [
os.path.basename(line.strip())
for line in playlist_content.splitlines()
if line.strip() and not line.startswith("#")
]
# Procesar en paralelo
tasks = [self.decrypt_single(name) for name in filenames]
await asyncio.gather(*tasks)
def assemble_video(self, final_name: str = "video_output.ts"):
"""Combina todos los segmentos"""
# Leer orden desde playlist
with open("stream.m3u8", "r", encoding="utf-8") as f:
content = f.read()
ordered_files = []
for line in content.splitlines():
line = line.strip()
if line and not line.startswith("#"):
filename = os.path.basename(line)
file_path = os.path.join(DECRYPTED_DIR, filename)
if os.path.exists(file_path):
ordered_files.append(file_path)
if not ordered_files:
print("No hay segmentos para unir")
return
# Fusionar binariamente
print(f"🎬 Creando {final_name}...")
with open(final_name, "wb") as output:
for ts_path in ordered_files:
with open(ts_path, "rb") as segment:
output.write(segment.read())
print("✨ Proceso completado")
async def run_extraction():
"""Pipeline completo de extracción"""
playlist_url = "https://cdn.example.com/videos/12345/master.m3u8"
extractor = VideoExtractor(playlist_url)
# 1. Obtener playlist
print("📡 Obteniendo información del stream...")
playlist = extractor.retrieve_playlist()
# 2. Extraer parámetros de cifrado
extractor.extract_crypto_params(playlist)
# 3. Descargar segmentos
print("\n⬇️ Descargando segmentos...")
await extractor.download_segments(playlist)
# 4. Descifrar
print("\n🔑 Procesando cifrado...")
await extractor.decrypt_all(playlist)
# 5. Ensamblar
print("\n🎞️ Generando video final...")
extractor.assemble_video("video_final.ts")
if __name__ == "__main__":
asyncio.run(run_extraction())
Este sistema implementa un pipeline completo con manejo robusto de errores, concurrencia controlada y soporte para contenido cifrado. La arquitectura basada en clases facilita la extensión y mantenimiento del código.