Correct Audio Duration Discrepancies with Multi-Tool Validation and Transcoding

1
import assemblyai as aai
2
import subprocess
3
4
aai.settings.api_key = "YOUR_API_KEY"
5
transcriber = aai.Transcriber()
6
7
def get_duration_ffprobe(file_path):
8
    command = [
9
        'ffprobe', '-v', 'error', '-show_entries',
10
        'format=duration', '-of',
11
        'default=noprint_wrappers=1:nokey=1', file_path
12
    ]
13
    try:
14
        duration = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
15
        return float(duration.stdout.strip())
16
    except ValueError:
17
        print("Error: Unable to parse duration from ffprobe output.")
18
        return None
19
20
def get_duration_sox(file_path):
21
    command = ['soxi', '-D', file_path]
22
    try:
23
        duration = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
24
        return float(duration.stdout.strip())
25
    except ValueError:
26
        print("Error: Unable to parse duration from SoX output.")
27
        return None
28
29
def get_duration_mediainfo(file_path):
30
    command = ['mediainfo', '--Output=General;%Duration%', file_path]
31
    try:
32
        duration_ms = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
33
        duration_str = duration_ms.stdout.strip()
34
        # Check if the output is empty or not a valid number
35
        if duration_str:
36
            return float(duration_str) / 1000
37
        else:
38
            print("Error: MediaInfo returned empty or invalid duration")
39
            return None
40
    except ValueError:
41
        print("Error: Unable to parse duration from MediaInfo output.")
42
        return None
43
44
def check_audio_durations(file_path):
45
    #Check if audio durations differ among the three tools.
46
    ffprobe_duration = get_duration_ffprobe(file_path)
47
    sox_duration = get_duration_sox(file_path)
48
    mediainfo_duration = get_duration_mediainfo(file_path)
49
50
    # Print all retrieved durations
51
    print(f"ffprobe duration: {ffprobe_duration:.6f} seconds" if ffprobe_duration is not None else "ffprobe duration: Error retrieving duration")
52
    print(f"SoX duration: {sox_duration:.6f} seconds" if sox_duration is not None else "SoX duration: Error retrieving duration")
53
    print(f"MediaInfo duration: {mediainfo_duration:.6f} seconds" if mediainfo_duration is not None else "MediaInfo duration: Error retrieving duration")
54
55
    # Return durations for further checks
56
    return (ffprobe_duration, sox_duration, mediainfo_duration)
57
58
def transcribe(file):
59
    print("Executing transcription as audio durations are consistent.")
60
    transcript = transcriber.transcribe(file)
61
    print(transcript.text)
62
63
def transcode(input_file, output_file):
64
    #Transcode audio file to a 16kHz WAV file.
65
    print(f"Transcoding file {input_file} to {output_file}...")
66
    command = [
67
        'ffmpeg', '-i', input_file, '-ar', '16000', '-ac', '1', output_file
68
    ]
69
    try:
70
        subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
71
        if os.path.exists(output_file):
72
            print(f"Transcoding complete. Output file: {output_file}")
73
        else:
74
            print("Error: Transcoding failed.")
75
    except subprocess.CalledProcessError as e:
76
        print("Warnings from ffmpeg")
77
        """Print errors or warnings from ffmpeg"""
78
        #print(e.stderr.decode())
79
80
def durations_are_consistent(durations, tolerance=0.01):
81
    #Check if durations are consistent within a given tolerance of 0.01 seconds.
82
    if None in durations:
83
        return False
84
    min_duration = min(durations)
85
    max_duration = max(durations)
86
    return (max_duration - min_duration) <= tolerance
87
88
def main(file_path):
89
    durations = check_audio_durations(file_path)
90
91
    if durations:
92
        if None in durations:
93
            print("Error: One or more duration values could not be retrieved.")
94
            transcoded_file = file_path.rsplit('.', 1)[0] + '_transcoded.wav'
95
            transcode(file_path, transcoded_file)
96
            new_durations = check_audio_durations(transcoded_file)
97
            if new_durations and durations_are_consistent(new_durations):
98
                transcribe(transcoded_file)
99
            else:
100
                print("Warning: The audio durations still differ or an error occurred with the transcoded file.")
101
        elif not durations_are_consistent(durations):
102
            print("Warning: The audio durations differ between tools.")
103
            transcoded_file = file_path.rsplit('.', 1)[0] + '_transcoded.wav'
104
            transcode(file_path, transcoded_file)
105
            new_durations = check_audio_durations(transcoded_file)
106
            if new_durations and durations_are_consistent(new_durations):
107
                transcribe(transcoded_file)
108
            else:
109
                print("Warning: The audio durations still differ or an error occurred with the transcoded file.")
110
        else:
111
            print("The audio durations are consistent.")
112
            transcribe(file_path)
113
114
audio_file="./audio.mp4"
115
116
if __name__ == "__main__":
117
    file_path = f"{audio_file}"
118
    main(file_path)

1
import assemblyai as aai
2
import subprocess
3
4
aai.settings.api_key = "YOUR_API_KEY"
5
transcriber = aai.Transcriber()

1
def get_duration_ffprobe(file_path):
2
    command = [
3
        'ffprobe', '-v', 'error', '-show_entries',
4
        'format=duration', '-of',
5
        'default=noprint_wrappers=1:nokey=1', file_path
6
    ]
7
    try:
8
        duration = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
9
        return float(duration.stdout.strip())
10
    except ValueError:
11
        print("Error: Unable to parse duration from ffprobe output.")
12
        return None

1
def get_duration_sox(file_path):
2
    command = ['soxi', '-D', file_path]
3
    try:
4
        duration = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
5
        return float(duration.stdout.strip())
6
    except ValueError:
7
        print("Error: Unable to parse duration from SoX output.")
8
        return None

1
def get_duration_mediainfo(file_path):
2
    command = ['mediainfo', '--Output=General;%Duration%', file_path]
3
    try:
4
        duration_ms = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
5
        duration_str = duration_ms.stdout.strip()
6
        # Check if the output is empty or not a valid number
7
        if duration_str:
8
            return float(duration_str) / 1000
9
        else:
10
            print("Error: MediaInfo returned empty or invalid duration")
11
            return None
12
    except ValueError:
13
        print("Error: Unable to parse duration from MediaInfo output.")
14
        return None

1
def check_audio_durations(file_path):
2
    #Check if audio durations differ among the three tools.
3
    ffprobe_duration = get_duration_ffprobe(file_path)
4
    sox_duration = get_duration_sox(file_path)
5
    mediainfo_duration = get_duration_mediainfo(file_path)
6
7
    # Print all retrieved durations
8
    print(f"ffprobe duration: {ffprobe_duration:.6f} seconds" if ffprobe_duration is not None else "ffprobe duration: Error retrieving duration")
9
    print(f"SoX duration: {sox_duration:.6f} seconds" if sox_duration is not None else "SoX duration: Error retrieving duration")
10
    print(f"MediaInfo duration: {mediainfo_duration:.6f} seconds" if mediainfo_duration is not None else "MediaInfo duration: Error retrieving duration")
11
12
    # Return durations for further checks
13
    return (ffprobe_duration, sox_duration, mediainfo_duration)

1
def transcribe(file):
2
    print("Executing transcription as audio durations are consistent.")
3
    transcript = transcriber.transcribe(file)
4
    print(transcript.text)

1
def transcode(input_file, output_file):
2
    #Transcode audio file to a 16kHz WAV file.
3
    print(f"Transcoding file {input_file} to {output_file}...")
4
    command = [
5
        'ffmpeg', '-i', input_file, '-ar', '16000', '-ac', '1', output_file
6
    ]
7
    try:
8
        subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
9
        if os.path.exists(output_file):
10
            print(f"Transcoding complete. Output file: {output_file}")
11
        else:
12
            print("Error: Transcoding failed.")
13
    except subprocess.CalledProcessError as e:
14
        print("Warnings from ffmpeg")
15
        """Print errors or warnings from ffmpeg"""
16
        #print(e.stderr.decode())

1
def durations_are_consistent(durations, tolerance=0.01):
2
    #Check if durations are consistent within a given tolerance of 0.01 seconds.
3
    if None in durations:
4
        return False
5
    min_duration = min(durations)
6
    max_duration = max(durations)
7
    return (max_duration - min_duration) <= tolerance

1
def main(file_path):
2
    durations = check_audio_durations(file_path)
3
4
    if durations:
5
        if None in durations:
6
            print("Error: One or more duration values could not be retrieved.")
7
            transcoded_file = file_path.rsplit('.', 1)[0] + '_transcoded.wav'
8
            transcode(file_path, transcoded_file)
9
            new_durations = check_audio_durations(transcoded_file)
10
            if new_durations and durations_are_consistent(new_durations):
11
                transcribe(transcoded_file)
12
            else:
13
                print("Warning: The audio durations still differ or an error occurred with the transcoded file.")
14
        elif not durations_are_consistent(durations):
15
            print("Warning: The audio durations differ between tools.")
16
            transcoded_file = file_path.rsplit('.', 1)[0] + '_transcoded.wav'
17
            transcode(file_path, transcoded_file)
18
            new_durations = check_audio_durations(transcoded_file)
19
            if new_durations and durations_are_consistent(new_durations):
20
                transcribe(transcoded_file)
21
            else:
22
                print("Warning: The audio durations still differ or an error occurred with the transcoded file.")
23
        else:
24
            print("The audio durations are consistent.")
25
            transcribe(file_path)
26
27
audio_file="./audio/8950.mp4"
28
29
if __name__ == "__main__":
30
    file_path = f"{audio_file}"
31
    main(file_path)