Update app.py

app.py

@@ -7,9 +7,7 @@ import gradio as gr
 from pydub import AudioSegment
 from transformers import pipeline
 
-# -------------------------------------------------
 # Configuration
-# -------------------------------------------------
 MODEL_NAME = "openai/whisper-large-v3"
 device = "cuda" if torch.cuda.is_available() else "cpu"
 
@@ -28,137 +26,181 @@ def init_metadata_state():
 
 def transcribe_audio(audio_path):
     if not audio_path:
-        return "Aucun fichier audio fourni", [], None, []
+        return "Aucun fichier audio fourni", [], None
     
-    result = pipe(audio_path, return_timestamps="word", generate_kwargs={"language": "french"})
-    words = result.get("chunks", [])
+    # Utilisation explicite des timestamps au niveau des mots
+    result = pipe(audio_path, return_timestamps="word")
     
-    if not words:
-        return "Erreur lors de la récupération des timestamps", [], None, []
+    # Extraction du texte complet
+    text = result["text"]
     
-    raw_transcription = " ".join([w["text"] for w in words])
-    word_timestamps = [(w["text"], w["timestamp"]) for w in words]
+    # Création des segments initiaux à partir des timestamps
+    chunks = result["chunks"]
+    table_data = []
     
-    return raw_transcription, [], audio_path, word_timestamps
-
-def preprocess_segments(table_data, word_timestamps):
-    formatted_data = []
+    # Regrouper les mots en phrases (séparées par la ponctuation)
+    current_segment = []
+    current_start = None
     
-    for i, row in enumerate(table_data):
-        if not row or len(row) < 1 or not row[0].strip():
-            continue
-
-        text = row[0].strip()
-        segment_id = f"seg_{i+1:02d}"
-
-        matching_timestamps = [
-            (start, end) for word, (start, end) in word_timestamps if word in text
-        ]
-
-        if matching_timestamps:
-            start_time, end_time = matching_timestamps[0]
-        else:
-            start_time, end_time = None, None
-
-        formatted_data.append([text, start_time, end_time, segment_id])
-
-    return formatted_data
+    for chunk in chunks:
+        if not current_start:
+            current_start = chunk["timestamp"][0]
+        
+        current_segment.append(chunk["text"])
+        
+        # Si on trouve une ponctuation de fin ou si le segment est assez long
+        if any(punct in chunk["text"] for punct in ".!?") or len(" ".join(current_segment)) > 100:
+            segment_text = " ".join(current_segment).strip()
+            if segment_text:
+                table_data.append([
+                    segment_text,
+                    round(current_start, 2),
+                    round(chunk["timestamp"][1], 2),
+                    f"seg_{len(table_data)+1:02d}"
+                ])
+                current_segment = []
+                current_start = None
+    
+    # Ajouter le dernier segment s'il existe
+    if current_segment:
+        last_chunk = chunks[-1]
+        segment_text = " ".join(current_segment).strip()
+        if segment_text:
+            table_data.append([
+                segment_text,
+                round(current_start, 2),
+                round(last_chunk["timestamp"][1], 2),
+                f"seg_{len(table_data)+1:02d}"
+            ])
+    
+    # S'assurer qu'on a au moins 5 lignes pour l'interface
+    while len(table_data) < 5:
+        table_data.append(["", None, None, ""])
+    
+    return text, table_data, audio_path
 
-def validate_segments(audio_path, table_data, metadata_state, word_timestamps):
-    if not audio_path or not word_timestamps:
+def validate_segments(audio_path, table_data, metadata_state):
+    if not audio_path:
         return [], metadata_state
     
     if os.path.exists(TEMP_DIR):
         shutil.rmtree(TEMP_DIR)
     os.makedirs(TEMP_DIR, exist_ok=True)
 
-    original_audio = AudioSegment.from_file(audio_path)
-    segment_paths = []
-    updated_metadata = []
-
-    for text, start_time, end_time, segment_id in table_data:
-        if start_time is None or end_time is None:
-            continue
-
-        start_ms, end_ms = int(float(start_time) * 1000), int(float(end_time) * 1000)
-        if start_ms < 0 or end_ms <= start_ms:
-            continue
-
-        segment_filename = f"{Path(audio_path).stem}_{segment_id}.wav"
-        segment_path = os.path.join(TEMP_DIR, segment_filename)
-
-        extract = original_audio[start_ms:end_ms]
-        extract.export(segment_path, format="wav")
-
-        segment_paths.append(segment_path)
-        updated_metadata.append({
-            "audio_file": segment_filename,
-            "text": text,
-            "start_time": start_time,
-            "end_time": end_time,
-            "id": segment_id,
-        })
-    
-    return segment_paths, updated_metadata
+    try:
+        original_audio = AudioSegment.from_file(audio_path)
+        segment_paths = []
+        updated_metadata = []
+
+        for i, row in enumerate(table_data):
+            if not row or len(row) < 4:
+                continue
+                
+            text, start_time, end_time, segment_id = row
+
+            if not text or start_time is None or end_time is None:
+                continue
+                
+            if not segment_id:
+                segment_id = f"seg_{i+1:02d}"
+
+            start_ms = int(float(start_time) * 1000)
+            end_ms = int(float(end_time) * 1000)
+            
+            if start_ms < 0 or end_ms <= start_ms or end_ms > len(original_audio):
+                print(f"Invalid timestamps for segment {segment_id}: {start_time}-{end_time}")
+                continue
+
+            segment_filename = f"{Path(audio_path).stem}_{segment_id}.wav"
+            segment_path = os.path.join(TEMP_DIR, segment_filename)
+            
+            try:
+                segment_audio = original_audio[start_ms:end_ms]
+                segment_audio.export(segment_path, format="wav")
+                
+                segment_paths.append(segment_path)
+                updated_metadata.append({
+                    "audio_file": segment_filename,
+                    "text": text,
+                    "start_time": start_time,
+                    "end_time": end_time,
+                    "id": segment_id,
+                })
+            except Exception as e:
+                print(f"Error processing segment {segment_id}: {str(e)}")
+                continue
+
+        return segment_paths, updated_metadata
+
+    except Exception as e:
+        print(f"Error in validate_segments: {str(e)}")
+        return [], metadata_state
 
 def generate_zip(metadata_state):
     if not metadata_state:
         return None
     
-    zip_path = os.path.join(TEMP_DIR, "dataset.zip")
-    if os.path.exists(zip_path):
-        os.remove(zip_path)
-    
-    metadata_csv_path = os.path.join(TEMP_DIR, "metadata.csv")
-    with open(metadata_csv_path, "w", encoding="utf-8") as f:
-        f.write("audio_file|text|speaker_name|API\n")
-        for seg in metadata_state:
-            f.write(f"{seg['audio_file']}|{seg['text']}|projectname|/API_PHONETIC/\n")
-    
-    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
-        zf.write(metadata_csv_path, "metadata.csv")
-        for seg in metadata_state:
-            file_path = os.path.join(TEMP_DIR, seg["audio_file"])
-            if os.path.exists(file_path):
-                zf.write(file_path, seg["audio_file"])
-    
-    return zip_path
+    try:
+        zip_path = os.path.join(TEMP_DIR, "dataset.zip")
+        if os.path.exists(zip_path):
+            os.remove(zip_path)
+        
+        metadata_csv_path = os.path.join(TEMP_DIR, "metadata.csv")
+        with open(metadata_csv_path, "w", encoding="utf-8") as f:
+            f.write("audio_file|text|speaker_name|API\n")
+            for seg in metadata_state:
+                f.write(f"{seg['audio_file']}|{seg['text']}|projectname|/API_PHONETIC/\n")
+        
+        with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
+            zf.write(metadata_csv_path, "metadata.csv")
+            for seg in metadata_state:
+                file_path = os.path.join(TEMP_DIR, seg["audio_file"])
+                if os.path.exists(file_path):
+                    zf.write(file_path, seg["audio_file"])
+        
+        return zip_path
+    except Exception as e:
+        print(f"Error generating ZIP: {str(e)}")
+        return None
 
-# -------------------------------------------------
 # Interface Gradio
-# -------------------------------------------------
 with gr.Blocks() as demo:
     gr.Markdown("# Application de Découpe Audio")
     metadata_state = gr.State(init_metadata_state())
-    extracted_segments = gr.State([])
 
-    audio_input = gr.Audio(type="filepath", label="Fichier audio")
-    raw_transcription = gr.Textbox(label="Transcription", interactive=False)
-    table = gr.Dataframe(headers=["Texte"], datatype=["str"], row_count=(1, "dynamic"), col_count=1)
-    validate_button = gr.Button("Valider")
-    generate_button = gr.Button("Générer ZIP")
+    with gr.Row():
+        audio_input = gr.Audio(type="filepath", label="Fichier audio")
+        raw_transcription = gr.Textbox(label="Transcription", interactive=False)
+    
+    table = gr.Dataframe(
+        headers=["Texte", "Début (s)", "Fin (s)", "ID"],
+        datatype=["str", "number", "number", "str"],
+        row_count=5,
+        col_count=4,
+        interactive=True
+    )
+    
+    with gr.Row():
+        validate_button = gr.Button("Valider")
+        generate_button = gr.Button("Générer ZIP")
+    
+    status_message = gr.Textbox(label="Status", interactive=False)
     zip_file = gr.File(label="Télécharger le ZIP")
 
-    word_timestamps = gr.State()
-
-    audio_input.change(transcribe_audio, inputs=audio_input, outputs=[raw_transcription, table, audio_input, word_timestamps])
+    def process_validation(audio_path, table_data, metadata_state):
+        if not audio_path:
+            return [], [], "Aucun fichier audio fourni"
+        
+        segments, metadata = validate_segments(audio_path, table_data, metadata_state)
+        status = f"{len(segments)} segments générés" if segments else "Aucun segment généré"
+        return table_data, metadata, status
 
+    audio_input.change(transcribe_audio, inputs=audio_input, outputs=[raw_transcription, table, audio_input])
     validate_button.click(
-        fn=lambda table_data, word_timestamps, audio_path, metadata_state: validate_segments(
-            audio_path, preprocess_segments(table_data, word_timestamps), metadata_state, word_timestamps
-        ),
-        inputs=[table, word_timestamps, audio_input, metadata_state],
-        outputs=[extracted_segments, metadata_state],
+        process_validation,
+        inputs=[audio_input, table, metadata_state],
+        outputs=[table, metadata_state, status_message]
     )
-
-    @gr.render(inputs=extracted_segments)
-    def show_audio_excerpts(segments):
-        if not segments:
-            gr.Markdown("Aucun extrait généré.")
-        else:
-            for i, seg in enumerate(segments):
-                gr.Audio(label=f"Extrait {i+1}", value=seg)
-
     generate_button.click(generate_zip, inputs=metadata_state, outputs=zip_file)
 
-demo.queue().launch()
+demo.queue().launch()