File size: 2,868 Bytes
c336648 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 |
import os
import subprocess
import concurrent.futures
from datetime import timedelta
def convert_mp4_to_wav(mp4_file, wav_file):
cmd = 'ffmpeg -i {} -acodec pcm_s16le -ar 48000 {}'
if wav_file is None or len(wav_file) == 0:
wav_file = mp4_file.replace('.mp4', '.wav')
os.system(cmd.format(mp4_file, wav_file))
segment_duration = 30 * 60
output_pattern = f"{wav_file.replace('.wav', '')}_%03d.wav"
cmd = f'ffmpeg -i "{wav_file}" -f segment -segment_time {segment_duration} -c copy "{output_pattern}"'
os.system(cmd)
def duration_get(file_path):
ffprobe_command = f'ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 "{file_path}"'
result = subprocess.check_output(ffprobe_command, shell=True, universal_newlines=True)
duration = float(result)
print(f'{file_path} : {duration}')
return duration
def duration_sum(directory):
total_duration = 0
with concurrent.futures.ThreadPoolExecutor() as executor:
# Collect the futures
futures = []
for filename in os.listdir(directory):
filepath = os.path.join(directory, filename)
if os.path.isfile(filepath):
try:
future = executor.submit(duration_get, filepath)
futures.append(future)
except (subprocess.CalledProcessError, ValueError):
continue
# Collect the results
for future in concurrent.futures.as_completed(futures):
try:
duration = future.result()
total_duration += duration
except Exception as e:
print(f"Error occurred: {e}")
return total_duration
def duration_split(input_dir, output_dir, num_parts=2, file_ext="wav"):
if not input_dir:
print("Input directory not provided.")
return
if not output_dir:
output_dir = os.path.join(input_dir, "output")
os.makedirs(output_dir, exist_ok=True)
for file_name in os.listdir(input_dir):
if file_name.endswith("." + file_ext):
file_path = os.path.join(input_dir, file_name)
duration = duration_get(file_path)
part_duration = duration / num_parts
print("[ * ] Duration:", duration)
print("[ * ] Number of Parts:", num_parts)
print("[ * ] Part Duration:", part_duration)
base_name = os.path.splitext(file_name)[0]
for i in range(num_parts):
part_output_file = os.path.join(output_dir, f"{base_name}_part{i + 1}.{file_ext}")
start_time = i * part_duration
ffmpeg_split_command = f'ffmpeg -i "{file_path}" -ss {start_time} -t {part_duration} -c copy "{part_output_file}"'
subprocess.call(ffmpeg_split_command, shell=True)
|