A Python-based HLS toolkit to download, decrypt (AES-128), extract audio/video/subtitle tracks, and mux streams from .m3u8 playlists using multithreading and PyAV.

streamtoolkit.py

import os
import re
import requests
import m3u8
import argparse
from pathlib import Path
from Crypto.Cipher import AES
import av
from concurrent.futures import ThreadPoolExecutor
from collections import defaultdict

class HLSToolkit:
    def __init__(self, url, threads=8, temp_dir="temp"):
        self.url = url
        self.threads = threads
        self.temp_dir = Path(temp_dir)
        self.temp_dir.mkdir(exist_ok=True)
        self.playlist = None
        self.key = None
        self.iv = None

    def get_best_stream_uri(self, media_type='audio'):
        """Get the highest quality stream URI for specified media type"""
        if media_type.lower() == 'audio' and "default_audio" in self.url and self.url.endswith(".m3u8"):
            return self.url

        master = m3u8.load(self.url)
        if not master.is_variant:
            return self.url if any(s.type == media_type for s in master.segments) else None

        streams = []
        for playlist in master.playlists:
            if media_type.lower() == 'video':
                streams.append((playlist.stream_info.bandwidth, playlist.uri))
            elif media_type.lower() == 'audio':
                for media in master.media:
                    if media.type == 'AUDIO' and media.uri:
                        bitrate = int(re.search(r"default_audio(\d+)", media.uri).group(1)) if re.search(r"default_audio(\d+)", media.uri) else 0
                        streams.append((bitrate, media.uri))

        if not streams:
            raise ValueError(f"No {media_type} streams found in master playlist.")
        return sorted(streams, reverse=True)[0][1]

    def get_decryption_key(self):
        """Get decryption key and IV from playlist"""
        if not self.playlist:
            self.playlist = m3u8.load(self.url)
        key_obj = self.playlist.keys[0] if self.playlist.keys else None
        if key_obj:
            self.key = requests.get(key_obj.uri).content
            self.iv = bytes.fromhex(key_obj.iv.replace("0x", "")) if key_obj.iv else None
        return self.key, self.iv

    def download_segments(self, playlist_url, output_name="output"):
        """Download and decrypt segments"""
        self.playlist = m3u8.load(playlist_url)
        self.get_decryption_key()
        segment_paths = []

        def process_segment(i, segment):
            seg_path = self.temp_dir / f"{output_name}_{i:04d}.ts"
            if seg_path.exists():
                return seg_path

            try:
                res = requests.get(segment.absolute_uri)
                res.raise_for_status()
                if self.key:
                    iv = self.iv or int(segment.sequence).to_bytes(16, 'big')
                    cipher = AES.new(self.key, AES.MODE_CBC, iv)
                    data = cipher.decrypt(res.content)
                else:
                    data = res.content
                
                with open(seg_path, "wb") as f:
                    f.write(data)
                print(f"[+] Processed segment {i+1}/{len(self.playlist.segments)}")
            except Exception as e:
                print(f"[!] Failed segment {i+1}: {e}")
            return seg_path

        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            futures = [executor.submit(process_segment, i, seg) 
                      for i, seg in enumerate(self.playlist.segments)]
            segment_paths = [f.result() for f in futures]

        return segment_paths

    def merge_files(self, paths, output):
        """Merge segment files into single file"""
        with open(output, "wb") as outfile:
            for path in paths:
                if path and path.exists():
                    with open(path, "rb") as infile:
                        outfile.write(infile.read())
        return output

    def extract_tracks(self, input_file, output_prefix="output"):
        """Extract all tracks from container"""
        container = av.open(input_file)
        outputs = defaultdict(list)
        
        for stream in container.streams:
            ext = {
                'audio': '.aac',
                'video': '.h264',
                'subtitle': '.subtitles'
            }.get(stream.type, f".{stream.codec.name}")
            
            output_path = f"{output_prefix}_{stream.type}_{stream.index}{ext}"
            with open(output_path, "wb") as f:
                for packet in container.demux(stream):
                    if packet.size > 0:
                        f.write(bytes(packet))
                outputs[stream.type].append(output_path)
                print(f"[+] Extracted {stream.type} track to {output_path}")
        
        return outputs

    def mux_tracks(self, output_file, **tracks):
        """Mux tracks into output container"""
        output = av.open(output_file, 'w')
        streams = {}
        
        # Create streams
        for track_type, files in tracks.items():
            for file in files:
                input = av.open(file)
                for stream in input.streams:
                    if stream.type == track_type:
                        streams[f"{track_type}_{len(streams)}"] = output.add_stream(
                            template=stream
                        )
                        break
        
        # Mux packets
        for track_type, files in tracks.items():
            for file in files:
                input = av.open(file)
                for stream in input.streams:
                    if stream.type == track_type:
                        out_stream = streams[f"{track_type}_{stream.index}"]
                        for packet in input.demux(stream):
                            if packet.size > 0:
                                packet.stream = out_stream
                                output.mux(packet)
        
        output.close()
        print(f"[+] Muxed output to {output_file}")

def main():
    parser = argparse.ArgumentParser(description="HLS Stream Toolkit - Extract/Add/Remove tracks without re-encoding")
    parser.add_argument('-u', '--url', required=True, help='Master/manifest URL')
    parser.add_argument('-t', '--threads', type=int, default=8, help='Download threads')
    parser.add_argument('-o', '--output', default="output.ts", help='Output file')
    
    # Track operations
    parser.add_argument('--extract', action='store_true', help='Extract all tracks to separate files')
    parser.add_argument('--audio', action='store_true', help='Include audio track')
    parser.add_argument('--video', action='store_true', help='Include video track')
    parser.add_argument('--subtitles', action='store_true', help='Include subtitle track')
    parser.add_argument('--add', nargs='+', help='Additional files to mux (video.mp4,audio.aac,subtitles.srt)')
    
    args = parser.parse_args()

    toolkit = HLSToolkit(args.url, args.threads)

    try:
        # Download and process streams
        tracks_to_mux = {}
        
        if args.audio:
            print("[*] Processing audio stream...")
            audio_url = toolkit.get_best_stream_uri('audio')
            audio_segments = toolkit.download_segments(audio_url, "audio")
            audio_file = toolkit.merge_files(audio_segments, "audio.ts")
            tracks_to_mux['audio'] = [audio_file]

        if args.video:
            print("[*] Processing video stream...")
            video_url = toolkit.get_best_stream_uri('video')
            video_segments = toolkit.download_segments(video_url, "video")
            video_file = toolkit.merge_files(video_segments, "video.ts")
            tracks_to_mux['video'] = [video_file]

        if args.extract:
            print("[*] Extracting tracks...")
            input_file = list(tracks_to_mux.values())[0][0] if tracks_to_mux else None
            if not input_file:
                raise ValueError("No tracks selected for extraction")
            toolkit.extract_tracks(input_file)

        if args.add:
            print("[*] Adding external files...")
            for file in args.add:
                ext = Path(file).suffix.lower()
                if ext in ('.aac', '.mp3', '.wav'):
                    tracks_to_mux.setdefault('audio', []).append(file)
                elif ext in ('.h264', '.mp4', '.ts'):
                    tracks_to_mux.setdefault('video', []).append(file)
                elif ext in ('.srt', '.vtt', '.ass'):
                    tracks_to_mux.setdefault('subtitle', []).append(file)

        if tracks_to_mux:
            print("[*] Muxing tracks...")
            toolkit.mux_tracks(args.output, **tracks_to_mux)
            print(f"[+] Successfully created {args.output}")

    except Exception as e:
        print(f"[!] Error: {e}")
    finally:
        # Cleanup temp files
        for file in toolkit.temp_dir.glob("*"):
            file.unlink()
        toolkit.temp_dir.rmdir()

if __name__ == "__main__":
    main()