The following code demonstrates how to implement multi-threaded downloading of TS segments and merge them into an MP4 file using Python. The approach involves downloading TS segments via multiple threads based on the links, decrypting them using a key if necessary, and then merging them with FFmpeg into an MP4 file.1.Parse the M3U8 file to obtain video segment information, and decrypt if necessary.
def parse_m3u8(self): """Parse the M3U8 file to obtain video segment information""" logger.info(f"Parsing M3U8 file: {self.m3u8_url}") try: response = requests.get(self.m3u8_url, headers=self.headers, timeout=30) response.raise_for_status() content = response.text except Exception as e: logger.error(f"Failed to download M3U8 file: {e}") return False # Check if the M3U8 is encrypted is_encrypted = False key_url = None lines = content.strip().split('\n') for i, line in enumerate(lines): line = line.strip() # Check for encryption information if line.startswith('#EXT-X-KEY'): is_encrypted = True # Parse key information match = re.search(r'METHOD=([^,]+)', line) if match and match.group(1) != 'NONE': method = match.group(1) logger.info(f"Detected encryption method: {method}") # Get key URL key_match = re.search(r'URI="([^"]+)"', line) if key_match: key_url = key_match.group(1) if not key_url.startswith('http'): key_url = urljoin(self.m3u8_url, key_url) logger.info(f"Key URL: {key_url}") # Get IV iv_match = re.search(r'IV=([^,]+)', line) if iv_match: self.iv = iv_match.group(1).replace('0x', '') # Get video segment URL if line and not line.startswith('#'): if not line.startswith('http'): line = urljoin(self.m3u8_url, line) self.segments.append({ 'url': line, 'index': len(self.segments), 'encrypted': is_encrypted }) # Download key if is_encrypted and key_url and not self.key: if not self.download_key(key_url): return False self.total_segments = len(self.segments) logger.info(f"Parsing complete, total {self.total_segments} video segments") return True2. Extract the key from the link.
def download_key(self, key_url): """Download AES decryption key""" logger.info(f"Downloading decryption key: {key_url}") try: response = requests.get(key_url, headers=self.headers, timeout=30) response.raise_for_status() self.key = response.content logger.info(f"Key downloaded successfully, length: {len(self.key)} bytes") return True except Exception as e: logger.error(f"Failed to download key: {e}") return False3. The main method for decryption.
def decrypt_segment(self, data, segment_info): """Decrypt video segment""" if not self.key: logger.error("No decryption key") return None try: # Prepare IV if self.iv: iv = bytes.fromhex(self.iv) else: # If no IV is specified, use the segment index as IV iv = segment_info['index'].to_bytes(16, byteorder='big') # Create AES decryptor cipher = AES.new(self.key, AES.MODE_CBC, iv) # Decrypt data decrypted_data = unpad(cipher.decrypt(data), AES.block_size) return decrypted_data except Exception as e: logger.error(f"Decryption failed: {e}") return None4. Download all segments.
def download_segment(self, segment_info): """Download a single video segment""" index = segment_info['index'] url = segment_info['url'] encrypted = segment_info['encrypted'] try: # Download segment response = requests.get(url, headers=self.headers, timeout=30) response.raise_for_status() data = response.content # If decryption is needed if encrypted and self.key: data = self.decrypt_segment(data, segment_info) if data is None: logger.error(f"Segment {index} decryption failed") return False # Save segment segment_path = os.path.join(self.output_dir, 'segments', f'segment_{index:06d}.ts') with open(segment_path, 'wb') as f: f.write(data) self.downloaded_segments += 1 progress = (self.downloaded_segments / self.total_segments) * 100 logger.info(f"Download progress: {progress:.1f}% ({self.downloaded_segments}/{self.total_segments})") return True except Exception as e: logger.error(f"Failed to download segment {index}: {e}") return False def download_all_segments(self): """Download all video segments""" logger.info(f"Starting to download video segments, concurrency: {self.max_workers}") with ThreadPoolExecutor(max_workers=self.max_workers) as executor: # Submit all download tasks future_to_segment = { executor.submit(self.download_segment, segment): segment for segment in self.segments } # Wait for all tasks to complete failed_segments = [] for future in as_completed(future_to_segment): segment = future_to_segment[future] try: success = future.result() if not success: failed_segments.append(segment) except Exception as e: logger.error(f"Download task exception: {e}") failed_segments.append(segment) # Retry failed segments if failed_segments: logger.info(f"Retrying {len(failed_segments)} failed segments") for segment in failed_segments: self.download_segment(segment) logger.info("All segments downloaded")5. Useffmpeg to merge into MP4 and clean up segments.
def merge_segments(self): """Merge all segments into an MP4 file""" logger.info("Starting to merge video segments") output_path = os.path.join(self.output_dir, f"{self.output_name}.mp4") segment_list_path = os.path.join(self.output_dir, 'segments', 'segment_list.txt') # Create segment list file with open(segment_list_path, 'w', encoding='utf-8') as f: for i in range(self.total_segments): segment_path = os.path.join(self.output_dir, 'segments', f'segment_{i:06d}.ts') if os.path.exists(segment_path): f.write(f"file '{segment_path}'\n") # Use ffmpeg to merge segments try: cmd = [ 'ffmpeg', '-f', 'concat', '-safe', '0', '-i', segment_list_path, '-c', 'copy', '-bsf:a', 'aac_adtstoasc', '-y', output_path ] result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode == 0: logger.info(f"Video merged successfully: {output_path}") return True else: logger.error(f"FFmpeg merge failed: {result.stderr}") return False except FileNotFoundError: logger.error("FFmpeg not found, please install FFmpeg first") logger.info("You can manually merge segments:") logger.info(f"copy /b {os.path.join(self.output_dir, 'segments', '*.ts')} {output_path}") return False def clean_temp_files(self): """Clean temporary files""" logger.info("Cleaning temporary files") segments_dir = os.path.join(self.output_dir, 'segments') if os.path.exists(segments_dir): for file in os.listdir(segments_dir): file_path = os.path.join(segments_dir, file) if os.path.isfile(file_path): os.remove(file_path) os.rmdir(segments_dir) def run(self, clean_temp=True): """Run the complete download process""" logger.info("="*50) logger.info("M3U8 Video Downloader is starting") logger.info("="*50) start_time = time.time() # Parse M3U8 file if not self.parse_m3u8(): return False # Download all segments self.download_all_segments() # Merge segments if self.merge_segments(): # Clean temporary files if clean_temp: self.clean_temp_files() end_time = time.time() duration = end_time - start_time logger.info("="*50) logger.info(f"Download complete! Time taken: {duration:.1f} seconds") logger.info(f"Output file: {os.path.join(self.output_dir, self.output_name)}.mp4") logger.info("="*50) return True else: logger.error("Video merge failed") return False