Update app.py

app.py

@@ -8,13 +8,13 @@ from pydub import AudioSegment
 from transformers import pipeline
 
 # -------------------------------------------------
-# Configuration du modèle
+# Configuration
 # -------------------------------------------------
 MODEL_NAME = "openai/whisper-large-v3"
 device = "cuda" if torch.cuda.is_available() else "cpu"
 
 pipe = pipeline(
-    "automatic-speech-recognition",
+    task="automatic-speech-recognition",
     model=MODEL_NAME,
     device=device,
     model_kwargs={"low_cpu_mem_usage": True},
@@ -23,73 +23,70 @@ pipe = pipeline(
 TEMP_DIR = "./temp_audio"
 os.makedirs(TEMP_DIR, exist_ok=True)
 
-# -------------------------------------------------
-# Initialisation de l'état
-# -------------------------------------------------
 def init_metadata_state():
     return []
 
-# -------------------------------------------------
-# Étape 2 : Transcription avec Whisper
-# -------------------------------------------------
 def transcribe_audio(audio_path):
     if not audio_path:
-        return "Aucun fichier audio fourni", [], None
-
-    print("📌 Début de la transcription avec Whisper...")
+        return "Aucun fichier audio fourni", [], None, []
     
-    generate_kwargs = {
-        "language": "french",
-        "max_new_tokens": 448,
-        "num_beams": 1,
-        "condition_on_prev_tokens": False,
-        "compression_ratio_threshold": 1.35,
-        "temperature": (0.0, 0.2, 0.4, 0.6, 0.8, 1.0),
-        "logprob_threshold": -1.0,
-        "no_speech_threshold": 0.6,
-        "return_timestamps": True,
-    }
-
-    result = pipe(audio_path, return_timestamps="word", generate_kwargs=generate_kwargs)
+    result = pipe(audio_path, return_timestamps="word", generate_kwargs={"language": "french"})
+    words = result.get("chunks", [])
     
-    print("📌 DEBUG - Timestamps reçus de Whisper:", result["chunks"])
+    if not words:
+        return "Erreur lors de la récupération des timestamps", [], None, []
     
-    raw_transcription = " ".join([chunk["text"] for chunk in result["chunks"]])
-    word_timestamps = [(chunk["text"], (chunk["timestamp"][0], chunk["timestamp"][1])) for chunk in result["chunks"]]
-
-    return raw_transcription, word_timestamps, audio_path
-
-# -------------------------------------------------
-# Étape 5 : Validation des segments
-# -------------------------------------------------
-def validate_segments(audio_path, table_data, metadata_state, word_timestamps):
-    if not audio_path or not word_timestamps:
-        return [], metadata_state
-
-    original_audio = AudioSegment.from_file(audio_path)
-    segment_paths = []
-    updated_metadata = []
+    raw_transcription = " ".join([w["text"] for w in words])
+    word_timestamps = [(w["text"], w["timestamp"]) for w in words]
+    
+    return raw_transcription, [], audio_path, word_timestamps
 
+def preprocess_segments(table_data, word_timestamps):
+    formatted_data = []
+    
     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]
+
+        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:
-            print(f"⚠️ Aucun timestamp trouvé pour '{text}', estimation en cours...")
-            start_time = 0
-            end_time = start_time + 1.0
+            start_time, end_time = None, None
+
+        formatted_data.append([text, start_time, end_time, segment_id])
 
-        start_ms = int(float(start_time) * 1000)
-        end_ms = int(float(end_time) * 1000)
+    return formatted_data
+
+def validate_segments(audio_path, table_data, metadata_state, word_timestamps):
+    if not audio_path or not word_timestamps:
+        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")
 
@@ -101,34 +98,30 @@ def validate_segments(audio_path, table_data, metadata_state, word_timestamps):
             "end_time": end_time,
             "id": segment_id,
         })
-
+    
     return segment_paths, updated_metadata
 
-# -------------------------------------------------
-# Étape 8 : Génération du ZIP
-# -------------------------------------------------
 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)
-
+            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
 
 # -------------------------------------------------
@@ -136,26 +129,36 @@ def generate_zip(metadata_state):
 # -------------------------------------------------
 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")
+    zip_file = gr.File(label="Télécharger le ZIP")
 
-    with gr.Column():
-        gr.Markdown("### 1. Téléversez un fichier audio")
-        audio_input = gr.Audio(type="filepath", label="Fichier audio")
+    word_timestamps = gr.State()
 
-    raw_transcription = gr.Textbox(label="Transcription (Whisper)", interactive=False)
-    table = gr.Dataframe(headers=["Texte"], datatype=["str"], row_count="dynamic")
-    validate_button = gr.Button("Valider et générer les extraits")
+    audio_input.change(transcribe_audio, inputs=audio_input, outputs=[raw_transcription, table, audio_input, word_timestamps])
 
-    @gr.render(inputs=table)
-    def render_audio_players(data):
-        return [gr.Audio(label=f"Extrait {i+1}", interactive=False) for i in range(len(data))]
+    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],
+    )
 
-    generate_button = gr.Button("Générer le fichier ZIP")
-    zip_file = gr.File(label="Télécharger le ZIP")
+    @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)
 
-    audio_input.change(transcribe_audio, inputs=audio_input, outputs=[raw_transcription, table, audio_input])
-    validate_button.click(validate_segments, inputs=[audio_input, table, metadata_state], outputs=[metadata_state])
     generate_button.click(generate_zip, inputs=metadata_state, outputs=zip_file)
 
-demo.queue().launch()
+demo.queue().launch()