import os import time import uuid # import librosa import numpy as np import soundfile as sf from threading import Thread from pydub import AudioSegment from pydub.silence import detect_nonsilent from concurrent.futures import ThreadPoolExecutor from apps.process.models import Status, ProcessedMedia from apps.process.utils import get_audio_parts import shutil import logging import mimetypes import subprocess from typing import Any, Optional # Create a logger for this file logger = logging.getLogger(__file__) def save_new_media( video_path: str, audio_path: str, output_video_path: str, video_codec: Optional[str] = 'libx264', audio_codec: Optional[str] = 'aac') -> None: """ Creates a new video by replacing the old audio with the generated audio input using FFmpeg. This function uses the `ffmpeg` command line utility to combine the video file at `video_path` and the audio file at `audio_path` into a new video file at `output_video_path`. Parameters: ==> Args: video_path : The path to the input video file. audio_path : The path to the input audio file that will be used to replace the audio in the input video. output_video_path : The path where the output video with the new audio will be saved. video_codec : The codec to use for the video stream. Default is 'libx264'. audio_codec : The codec to use for the audio stream. Default is 'aac'. Returns: """ # Check if the input video file exists and is a supported format if not os.path.exists(video_path) or not os.path.isfile(video_path): raise ValueError(f'Video file not found or is not a regular file: {video_path}') # if os.path.splitext(video_path)[1] not in ('.mp4', '.avi', '.mkv'): # raise ValueError(f'Unsupported video file extention: {os.path.splitext(video_path)[1]}') video_ext = os.path.splitext(video_path)[1] if video_ext in ('.mp3', '.wav', '.flac'): # If the video file is an audio file, save it directly to output_video_path shutil.copyfile(audio_path, output_video_path) return elif video_ext not in ('.mp4', '.avi', '.mkv'): raise ValueError(f'Unsupported video file extention: {video_ext}') # Check if the input audio file exists and is a supported format if not os.path.exists(audio_path) or not os.path.isfile(audio_path): raise ValueError(f'Audio file not found or is not a regular file: {audio_path}') if os.path.splitext(audio_path)[1] not in ('.mp3', '.wav', '.flac'): raise ValueError(f'Unsupported audio file extention: {os.path.splitext(audio_path)[1]}') # Build the FFmpeg command command = [ 'ffmpeg', "-hide_banner", '-i', video_path, '-i', audio_path, '-map', '0:v:0', '-map', '1:a:0', '-c:v', video_codec, '-c:a', audio_codec, '-strict', 'experimental', output_video_path ] try: # Use FFmpeg to extract the frames and audio and merge them into a new video subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except subprocess.CalledProcessError as e: logging.error(e.stdout) logging.error(e.stderr) raise def audiosegment_to_librosawav(sound): sound = sound.set_frame_rate(sound.frame_rate) channel_sounds = sound.split_to_mono() samples = [s.get_array_of_samples() for s in channel_sounds] fp_arr = np.array(samples).T.astype(np.float32) fp_arr /= np.iinfo(samples[0].typecode).max return fp_arr # def convert_video_to_audio_ffmpeg(video_file: str, output_ext="mp3"): # """Converts video to audio directly using `ffmpeg` command with the help of subprocess module""" # filename, ext = os.path.splitext(video_file) # # create paths to save origin audio and video # input_path = f"{video_file}" if ("original_audios" in ext) else f"media/original_videos/{video_file}" # output_path = f"{filename}.{output_ext}" if ("original_audios" in ext) else f"media/original_audios/{filename}.{output_ext}" # # Check if the output file exists # if os.path.exists(output_path): # return output_path # if not os.path.exists(f"media/original_audios"): # os.mkdir(f"media/original_audios") # command = ("ffmpeg -hide_banner -i "+ input_path + " " + output_path) # os.system(command) # return output_path def convert_video_to_audio_ffmpeg(video_file: str, output_ext="mp3"): """Converts video to audio directly using `ffmpeg` command with the help of the subprocess module""" filename, ext = os.path.splitext(video_file) # create paths to save the original audio and video input_path = os.path.join("media", "original_videos", video_file) if not "original_audios" in ext else video_file output_path = os.path.join("media", "original_audios", f"{filename}.{output_ext}") if not "original_audios" in ext else f"{filename}.{output_ext}" # Check if the output file exists if os.path.exists(output_path): return output_path os.makedirs(os.path.dirname(output_path), exist_ok=True) ffmpeg_command = f"ffmpeg -hide_banner -i {input_path} {output_path}" try: subprocess.run(ffmpeg_command, shell=True, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) except subprocess.CalledProcessError as e: raise Exception(f"Failed to convert video to audio: {e.stderr.decode().strip()}") return output_path def remove_sound(audio, part: list, MLmodel, save): """This function is used to remove the sounds from the audio part.""" # Testing if part is not None and if it is not empty and if it is a list or not. if part is not None and part != []: # extracted audio part from audio file extracted_audio = audio[part[0] : part[1]] # Transform from audiosegment to librosawav audio_librosa = audiosegment_to_librosawav(extracted_audio) audio_librosa = audio_librosa.T # Pass The Extracted Audio Part To The ML Model To Remove The background music from the audio part. voice_audio, background_audio = MLmodel.split(audio_librosa) # time.sleep(0.2) # Writing The Extracted Audio Part To A File. if not os.path.exists("media/music_output"): os.mkdir("media/music_output") if save: pass # sf.write(f"media/music_output/{filename}_{part[0]/1000}_{part[1]/1000}.wav",voice_audio,extracted_audio.frame_rate,) model_result = { "part": part, "voice": voice_audio, "background": background_audio, } return model_result def start_removing_sounds(MLmodel, parts_list: list, video_path: str, modia_id): """This function is used to remove the sounds from the video.""" print("Reciving File") print(video_path) print(modia_id) # convert part to audio audio_path = convert_video_to_audio_ffmpeg(video_path, output_ext="wav") # Testing if file_path existance and if it is a file.ArithmeticError() if (audio_path is not None and os.path.exists(audio_path) and os.path.isfile(audio_path)): # Reading Original Audio File audio = AudioSegment.from_file(audio_path) # Testing if parts_list is empty and if it is a list. if parts_list is not None and parts_list == []: # Create A New Parts List Of The Audio File With librosa. # parts_list = detect_nonsilent(audio, min_silence_len=1000, silence_thresh=-20, seek_step=1) parts_list= [[int(0),len(audio)]] # Create a Thread-Poll to remove the sounds from the audio part. with ThreadPoolExecutor() as executor: # Create a list of futures. futures = [executor.submit(remove_sound, audio, part, MLmodel, True) for part in parts_list] # Wait for the results to come back. results = [future.result() for future in futures] # Create a parts list for audio based on the part_list attribute all_parts = get_audio_parts(len(audio), parts_list) # Attrribute for Saving Final Audio audiofinal = AudioSegment.empty() # ---- for part in all_parts: # if part in parts_list and len(results) > 0: # for index in range(len(results)): # if results[index]["part"] == part: # audio_segment = results[index]["voice"] audio_segment = np.int16(audio_segment * 2**15) audio_segment = AudioSegment(audio_segment.tobytes(), frame_rate=audio.frame_rate, sample_width=(audio_segment).dtype.itemsize, channels=2,) # audiofinal += audio_segment # else: # audiofinal += audio[part[0] : part[1]] # For Saving Not Traited Parts # sf.write(f"media/music_output/non_selected_audio_{part[0]/1000}_{part[1]/1000}.wav", audiosegment_to_librosawav(audio[part[0]:part[1]]), audio.frame_rate) # ---- # media Object to get all info and use it to save processed media processedMedia = ProcessedMedia.objects.get(id = modia_id) # define path of the new audio :: new_audio_path = f"{audio_path.split('.')[0]}_{uuid.uuid4().hex}.wav" # define path of the old video :: save_video_path = f"{processedMedia.media.original_file.existingPath}" if (processedMedia.media.youtube_id != None ) else f"media/original_videos/{video_path}" # define path of the new video :: new_video_path = f"media/new_video/{processedMedia.media.youtube_id}_{uuid.uuid4().hex}.mp4" if (processedMedia.media.youtube_id != None ) else f"media/new_video/{video_path.split('.')[0]}_{uuid.uuid4().hex}.mp4" # check if folder of new videos exists if not os.path.exists("media/new_video"): os.mkdir("media/new_video") # Save Final Audio audiofinal.export(new_audio_path, format="wav") # for save || implementation sur vedio save_new_media(save_video_path, new_audio_path, new_video_path) # Saving Video In DataBase processedMedia.status = Status.Completed processedMedia.video_path = new_video_path processedMedia.audio_path = new_audio_path processedMedia.save() # save_final_video(media_id=modia_id, user=media.user, video_path=new_video_path, audio_path=new_audio_path, parts=all_parts)