Update app.py

app.py

@@ -2,6 +2,7 @@ import os
 import shutil
 import zipfile
 import torch
+import numpy as np
 import pandas as pd
 from pathlib import Path
 import gradio as gr
@@ -27,173 +28,115 @@ TEMP_DIR = "./temp_audio"
 os.makedirs(TEMP_DIR, exist_ok=True)
 
 def init_metadata_state():
-    """Initialise un DataFrame vide pour stocker les segments."""
-    return pd.DataFrame(columns=["Texte", "Début (s)", "Fin (s)", "ID"])
+    return []
 
 # -------------------------------------------------
 # 2. Transcription de l'audio avec Whisper
 # -------------------------------------------------
 def transcribe_audio(audio_path):
-    """Effectue la transcription de l'audio et génère les timestamps."""
     if not audio_path:
         print("[LOG] Aucun fichier audio fourni.")
-        return "Aucun fichier audio fourni", None, ""
-
+        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] Aucun timestamp détecté.")
-        return "Erreur : Aucun timestamp détecté.", None, ""
-
+        print("[LOG ERROR] Erreur : Aucun timestamp détecté.")
+        return "Erreur : Aucun timestamp détecté.", None, [], ""
+    
     raw_transcription = " ".join([w["text"] for w in words])
     word_timestamps = [(w["text"], w["timestamp"][0]) for w in words]
-
-    transcription_with_timestamps = " ".join(
-        [f"{w[0]}[{w[1]:.2f}]" for w in word_timestamps]
-    )
-
+    transcription_with_timestamps = " ".join([f"{w[0]}[{w[1]:.2f}]" for w in word_timestamps])
+    
     print(f"[LOG] Transcription brute : {raw_transcription}")
-    print(f"[LOG DETAIL] Timestamps associés : {word_timestamps}")
-
-    return raw_transcription, word_timestamps, transcription_with_timestamps
+    return raw_transcription, word_timestamps, transcription_with_timestamps, audio_path
 
 # -------------------------------------------------
-# 3. Gestion du tableau éditable (Ajout dynamique)
+# 3. Enregistrement des segments définis par l'utilisateur
 # -------------------------------------------------
-def add_row(metadata_state, new_rows):
-    """Ajoute dynamiquement des lignes au tableau en suivant le format structuré."""
-    if new_rows is None:
-        new_rows = []
-
-    formatted_rows = []
-    
-    # Gestion flexible des entrées (dictionnaires, listes, tuples, DataFrame)
-    if isinstance(new_rows, list):
-        for row in new_rows:
-            if isinstance(row, dict):
-                texte = row.get("Texte", "")
-                debut = row.get("Début (s)", None)
-                fin = row.get("Fin (s)", None)
-            elif isinstance(row, (list, tuple)) and len(row) >= 3:
-                texte, debut, fin = row[:3]
-            else:
-                continue
-            formatted_rows.append([texte, debut, fin, ""])
-
-    elif isinstance(new_rows, pd.DataFrame):
-        for _, row in new_rows.iterrows():
-            formatted_rows.append([row.get("Texte", ""), row.get("Début (s)", None), row.get("Fin (s)", None), ""])
-
-    # Conversion en DataFrame et fusion avec l'état actuel
-    if formatted_rows:
-        new_data = pd.DataFrame(formatted_rows, columns=["Texte", "Début (s)", "Fin (s)", "ID"])
-        metadata_state = pd.concat([metadata_state, new_data], ignore_index=True)
-
-        print(f"[LOG] {len(new_rows)} nouvelles lignes ajoutées.")
-
-    return metadata_state
-
-def save_segments(metadata_table):
-    """Sauvegarde les modifications apportées par l'utilisateur."""
-    metadata_state = pd.DataFrame(metadata_table, columns=["Texte", "Début (s)", "Fin (s)", "ID"])
-
+def save_segments(table_data):
     print("[LOG] Enregistrement des segments définis par l'utilisateur...")
-    for index, row in metadata_state.iterrows():
+    formatted_data = []
+    
+    for i, row in table_data.iterrows():
+        text, start_time, end_time = row["Texte"], row["Début (s)"], row["Fin (s)"]
+        segment_id = f"seg_{i+1:02d}"
+        
         try:
-            row["Début (s)"] = float(row["Début (s)"]) if row["Début (s)"] else None
-            row["Fin (s)"] = float(row["Fin (s)"]) if row["Fin (s)"] else None
-            row["ID"] = f"seg_{index+1:02d}"
-            print(f"[LOG] Segment enregistré : {row['Texte']} | Début: {row['Début (s)']}s, Fin: {row['Fin (s)']}s, ID: {row['ID']}")
+            start_time = str(start_time).replace(",", ".")
+            end_time = str(end_time).replace(",", ".")
+            
+            if not start_time.replace(".", "").isdigit() or not end_time.replace(".", "").isdigit():
+                raise ValueError("Valeurs de timestamps invalides")
+            
+            start_time = float(start_time)
+            end_time = float(end_time)
+            
+            if start_time < 0 or end_time <= start_time:
+                raise ValueError("Valeurs incohérentes")
+            
+            formatted_data.append([text, start_time, end_time, segment_id])
+            print(f"[LOG] Segment ajouté : {text} | Début: {start_time:.2f}s, Fin: {end_time:.2f}s, ID: {segment_id}")
+        
         except ValueError as e:
             print(f"[LOG ERROR] Erreur de conversion des timestamps : {e}")
-
-    return metadata_state
-
-# -------------------------------------------------
-# 4. Validation et découpage des extraits audio
-# -------------------------------------------------
-def validate_segments(audio_path, metadata_state):
-    """Découpe les extraits audio en fonction des segments définis."""
-    print("[LOG] Début de la validation des segments...")
-
-    if not audio_path or metadata_state.empty:
-        print("[LOG ERROR] Aucun segment 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)
-
-    for index, row in metadata_state.iterrows():
-        if row["Début (s)"] is None or row["Fin (s)"] is None:
-            print(f"[LOG ERROR] Timestamp manquant pour : {row['Texte']}")
-            continue
-
-        start_ms = int(float(row["Début (s)"]) * 1000)
-        end_ms = int(float(row["Fin (s)"]) * 1000)
-
-        if start_ms < 0 or end_ms <= start_ms:
-            print(f"[LOG ERROR] Problème de découpage : {row['Texte']} | {row['Début (s)']}s - {row['Fin (s)']}s")
-            continue
-
-        segment_filename = f"{Path(audio_path).stem}_{row['ID']}.wav"
-        segment_path = os.path.join(TEMP_DIR, segment_filename)
-
-        extract = original_audio[start_ms:end_ms]
-        extract.export(segment_path, format="wav")
-
-        metadata_state.at[index, "audio_file"] = segment_filename
-        print(f"[LOG] Extrait généré : {segment_filename}")
-
-    return metadata_state
+            return pd.DataFrame(), "Erreur : Vérifiez que les valeurs sont bien des nombres valides."
+    
+    return pd.DataFrame(formatted_data, columns=["Texte", "Début (s)", "Fin (s)", "ID"]), ""
 
 # -------------------------------------------------
-# 5. Génération du fichier ZIP
+# 4. Génération du fichier ZIP
 # -------------------------------------------------
-def generate_zip(metadata_state):
-    """Génère un fichier ZIP contenant les extraits audio et les métadonnées."""
-    if metadata_state.empty:
+def generate_zip(metadata_state, audio_path):
+    if isinstance(metadata_state, tuple): 
+        metadata_state = metadata_state[0]  # Extraire le DataFrame si c'est un tuple
+    
+    if metadata_state is None or metadata_state.empty:
         print("[LOG ERROR] Aucun segment valide trouvé pour la génération du ZIP.")
         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")
     metadata_state.to_csv(metadata_csv_path, sep="|", index=False)
-
+    
     with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
         zf.write(metadata_csv_path, "metadata.csv")
-        for index, row in metadata_state.iterrows():
-            file_path = os.path.join(TEMP_DIR, row["audio_file"])
-            if os.path.exists(file_path):
-                zf.write(file_path, row["audio_file"])
-
+        original_audio = AudioSegment.from_file(audio_path)
+        
+        for _, row in metadata_state.iterrows():
+            start_ms, end_ms = int(row["Début (s)"] * 1000), int(row["Fin (s)"] * 1000)
+            segment_audio = original_audio[start_ms:end_ms]
+            segment_filename = f"{Path(audio_path).stem}_{row['ID']}.wav"
+            segment_path = os.path.join(TEMP_DIR, segment_filename)
+            segment_audio.export(segment_path, format="wav")
+            zf.write(segment_path, segment_filename)
+    
     print("[LOG] Fichier ZIP généré avec succès.")
     return zip_path
 
 # -------------------------------------------------
-# 6. Interface utilisateur Gradio
+# 5. Interface utilisateur Gradio
 # -------------------------------------------------
 with gr.Blocks() as demo:
     gr.Markdown("# Application de Découpe Audio")
-
     metadata_state = gr.State(init_metadata_state())
-
+    
     audio_input = gr.Audio(type="filepath", label="Fichier audio")
-    table = gr.Dataframe(label="Segments", headers=["Texte", "Début (s)", "Fin (s)", "ID"], datatype=["str", "number", "number", "str"], interactive=True)
-
-    add_row_button = gr.Button("Ajouter des lignes")
-    save_button = gr.Button("Enregistrer")
-    validate_button = gr.Button("Valider")
+    raw_transcription = gr.Textbox(label="Transcription", interactive=False)
+    transcription_timestamps = gr.Textbox(label="Transcription avec Timestamps", interactive=False)
+    table = gr.Dataframe(headers=["Texte", "Début (s)", "Fin (s)"], datatype=["str", "str", "str"], row_count=(1, "dynamic"))
+    save_button = gr.Button("Enregistrer les segments")
     generate_button = gr.Button("Générer ZIP")
+    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, word_timestamps, transcription_timestamps, audio_input])
+    save_button.click(save_segments, inputs=table, outputs=[metadata_state])
+    generate_button.click(generate_zip, inputs=[metadata_state, audio_input], outputs=zip_file)
 
-    add_row_button.click(add_row, inputs=[metadata_state, table], outputs=metadata_state)
-    save_button.click(save_segments, inputs=table, outputs=metadata_state)
-
-demo.queue().launch()
+demo.queue().launch()