Update app.py

app.py

@@ -1,15 +1,23 @@
+# Version: Corrected After Test 3 (V2 - Improved with Scribe-based Timestamp Handling, Debugging Logs Added)
+# Description: Cette version intègre l'affichage des timestamps pour chaque mot,
+# permet une correction manuelle des erreurs, et ajoute une étape intermédiaire
+# avant la validation finale avec des logs détaillés pour le débogage.
+
 import os
 import shutil
 import zipfile
-from pathlib import Path
 import torch
+import numpy as np
+from pathlib import Path
 import gradio as gr
 from pydub import AudioSegment
 from transformers import pipeline
 
+# -------------------------------------------------
 # Configuration
+# -------------------------------------------------
 MODEL_NAME = "openai/whisper-large-v3"
-device = "cuda:0" if torch.cuda.is_available() else "cpu"
+device = "cuda" if torch.cuda.is_available() else "cpu"
 
 pipe = pipeline(
     task="automatic-speech-recognition",
@@ -21,49 +29,81 @@ pipe = pipeline(
 TEMP_DIR = "./temp_audio"
 os.makedirs(TEMP_DIR, exist_ok=True)
 
-# Initialisation de l'état
 def init_metadata_state():
     return []
 
 def transcribe_audio(audio_path):
     if not audio_path:
-        return "Aucun fichier audio fourni", [], None
+        print("[LOG] Aucun fichier audio fourni.")
+        return "Aucun fichier audio fourni", [], None, [], ""
+    
+    print(f"[LOG] Début de la transcription de {audio_path}...")
+    result = pipe(audio_path, return_timestamps="word")
+    words = result.get("chunks", [])
+    
+    if not words:
+        print("[LOG ERROR] Erreur : Aucun timestamp détecté.")
+        return "Erreur : Aucun timestamp détecté.", [], None, [], ""
     
-    result = pipe(audio_path, return_timestamps="word", generate_kwargs={"language": "french"})
-    chunks = result.get("chunks", [])
-    text = result["text"] if "text" in result else ""
+    raw_transcription = " ".join([w["text"] for w in words])
+    word_timestamps = [(w["text"], (w["timestamp"][0], w["timestamp"][1])) for w in words]
     
-    table_init = []
-    for chunk in chunks:
-        table_init.append([chunk["text"], None, None, ""])
+    transcription_with_timestamps = " ".join([f"{w[0]}[{w[1][0]:.2f}-{w[1][1]:.2f}]" for w in word_timestamps])
     
-    return text, table_init, audio_path
+    print(f"[LOG] Transcription brute : {raw_transcription}")
+    print(f"[LOG DETAIL] Timestamps associés : {word_timestamps}")
+    return raw_transcription, [], audio_path, word_timestamps, transcription_with_timestamps
 
-def validate_segments(audio_path, table_data, metadata_state):
-    if not audio_path:
+def preprocess_segments(table_data, word_timestamps):
+    print("[LOG] Début du prétraitement des segments...")
+    formatted_data = []
+    for i, row in enumerate(table_data):
+        if not row or len(row) < 1 or not row[0].strip():
+            print(f"[LOG WARNING] Ignoré : ligne vide à l'index {i}.")
+            continue
+
+        text = row[0].strip()
+        segment_id = f"seg_{i+1:02d}"
+
+        start_time, end_time = None, None
+        words_in_segment = text.split()
+        segment_indices = []
+        
+        for j, (word, (start, end)) in enumerate(word_timestamps):
+            if word in words_in_segment:
+                segment_indices.append((j, start, end))
+
+        if segment_indices:
+            start_time = segment_indices[0][1]
+            end_time = segment_indices[-1][2] if segment_indices[-1][2] else start_time + 0.5
+        
+        formatted_data.append([text, start_time, end_time, segment_id])
+        print(f"[LOG] Segment ajouté : {text} | Début: {start_time}, Fin: {end_time}, ID: {segment_id}")
+    
+    return formatted_data
+
+def validate_segments(audio_path, table_data, metadata_state, word_timestamps):
+    print("[LOG] Début de la validation des segments...")
+    if not audio_path or not word_timestamps:
+        print("[LOG ERROR] Erreur : Aucun timestamp valide trouvé !")
         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 i, row in enumerate(table_data):
-        if 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)
+    for text, start_time, end_time, segment_id in table_data:
+        if start_time is None or end_time is None:
+            print(f"[LOG ERROR] Timestamp manquant pour : {text}")
+            continue
+
+        start_ms, end_ms = int(float(start_time) * 1000), int(float(end_time) * 1000)
         if start_ms < 0 or end_ms <= start_ms:
+            print(f"[LOG ERROR] Problème de découpage : {text} | {start_time}s - {end_time}s")
             continue
 
         segment_filename = f"{Path(audio_path).stem}_{segment_id}.wav"
@@ -80,53 +120,31 @@ def validate_segments(audio_path, table_data, metadata_state):
             "end_time": end_time,
             "id": segment_id,
         })
+        print(f"[LOG] Extrait généré : {segment_filename}")
     
     return segment_paths, updated_metadata
 
-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:
-            line = f"{seg['audio_file']}|{seg['text']}|projectname|/API_PHONETIC/\n"
-            f.write(line)
-    
-    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
-
+# -------------------------------------------------
+# Interface Gradio
+# -------------------------------------------------
 with gr.Blocks() as demo:
-    gr.Markdown("# Application de Découpe Audio (Fixe)")
+    gr.Markdown("# Application de Découpe Audio")
     metadata_state = gr.State(init_metadata_state())
-    
-    with gr.Column():
-        gr.Markdown("### 1. Téléversez un fichier audio")
-        audio_input = gr.Audio(type="filepath", label="Fichier audio")
-    
-    raw_transcription = gr.Textbox(label="Transcription (Whisper)", interactive=False)
-    
-    gr.Markdown("### 2. Définissez vos segments")
-    table = gr.Dataframe(headers=["Texte", "Début (s)", "Fin (s)", "ID"], datatype=["str", "number", "number", "str"], row_count=5)
-    
-    validate_button = gr.Button("Valider et générer les extraits")
-    generate_button = gr.Button("Générer le fichier ZIP")
+    extracted_segments = gr.State([])
+
+    audio_input = gr.Audio(type="filepath", label="Fichier audio")
+    raw_transcription = gr.Textbox(label="Transcription", interactive=False)
+    transcription_timestamps = gr.Textbox(label="Transcription avec Timestamps", interactive=False)
+    table = gr.Dataframe(headers=["Texte"], datatype=["str"], row_count=(1, "dynamic"), col_count=1)
+    generate_timestamps_button = gr.Button("Générer les timestamps")
+    validate_button = gr.Button("Valider")
+    generate_button = gr.Button("Générer ZIP")
     zip_file = gr.File(label="Télécharger le ZIP")
-    
-    audio_players = [gr.Audio(label=f"Extrait {i+1}", interactive=False) for i in range(5)]
-    
-    validate_button.click(fn=validate_segments, inputs=[audio_input, table, metadata_state], outputs=audio_players + [metadata_state])
-    generate_button.click(fn=generate_zip, inputs=metadata_state, outputs=zip_file)
+    word_timestamps = gr.State()
+
+    audio_input.change(transcribe_audio, inputs=audio_input, outputs=[raw_transcription, table, audio_input, word_timestamps, transcription_timestamps])
+    generate_timestamps_button.click(preprocess_segments, inputs=[table, word_timestamps], outputs=table)
+    validate_button.click(validate_segments, inputs=[audio_input, table, metadata_state, word_timestamps], outputs=[extracted_segments, metadata_state])
+    generate_button.click(generate_zip, inputs=metadata_state, outputs=zip_file)
 
 demo.queue().launch()