Update app.py

app.py

@@ -2,18 +2,21 @@ import os
 import shutil
 import zipfile
 from pathlib import Path
+from datetime import datetime
 
 import gradio as gr
 import torch
 from pydub import AudioSegment
 from transformers import pipeline
 
-# -------------------------------------------------
-# Configuration
-# -------------------------------------------------
+# ------------------------
+# Configuration générale
+# ------------------------
+
 MODEL_NAME = "openai/whisper-large-v3"
 device = 0 if torch.cuda.is_available() else "cpu"
 
+# Pipeline Whisper pour la transcription
 pipe = pipeline(
     task="automatic-speech-recognition",
     model=MODEL_NAME,
@@ -21,206 +24,242 @@ pipe = pipeline(
     model_kwargs={"low_cpu_mem_usage": True},
 )
 
+# Dossier temporaire pour stocker extraits & ZIP
 TEMP_DIR = "./temp_audio"
 os.makedirs(TEMP_DIR, exist_ok=True)
 
-# -------------------------------------------------
-# Initialisation de l'état
-# -------------------------------------------------
+# On stocke la métadonnée globale dans un State
+# pour pouvoir y accéder lors de la génération du ZIP
+# (table de correspondance entre segments et infos)
 def init_metadata_state():
-    """Initialise l'état pour stocker les informations des segments validés."""
     return []
 
-# -------------------------------------------------
-# Étape 2 : Transcription avec Whisper
-# -------------------------------------------------
+# ------------------------
+# Étape 2 : Transcription
+# ------------------------
 def transcribe_audio(audio_path):
-    """Transcrit l'audio avec Whisper et retourne la transcription brute."""
+    """
+    1) Transcrit l'audio avec Whisper large-v3
+    2) Retourne la transcription brute et une table pré-remplie
+       (l'utilisateur pourra remplir manuellement la partie 'Texte' s'il le souhaite)
+    """
     if not audio_path:
-        return "Aucun fichier audio fourni.", [["", None, None, ""] for _ in range(20)], None
+        return "Aucun fichier audio fourni", [], None
 
-    # Transcription avec Whisper
+    # Transcription Whisper avec timestamps de chaque mot
     result = pipe(audio_path, return_timestamps="word")
     text = result["text"]
+    chunks = result["chunks"]  # liste de { 'timestamp': (start, end), 'text': ... }
 
-    # Tableau de 20 lignes vides pour l'édition
-    table_init = [["", None, None, ""] for _ in range(20)]
-    return text, table_init, audio_path
-
-# -------------------------------------------------
-# Étape intermédiaire : Génération des timestamps
-# -------------------------------------------------
-def generate_timestamps(audio_path, table_data):
-    """Génère les timestamps des segments en fonction des extraits textuels fournis."""
-    if not audio_path or not table_data:
-        return table_data
-
-    # Transcription avec Whisper pour obtenir les timestamps
-    result = pipe(audio_path, return_timestamps="word")
-    chunks = result["chunks"]
-
-    updated_table = []
-    for row in table_data:
-        if not row[0]:  # Ignorer les lignes sans texte
-            updated_table.append(row)
-            continue
-
-        text = row[0]
-        start_time, end_time = None, None
-
-        # Chercher les timestamps correspondant au texte
-        for chunk in chunks:
-            if text in chunk["text"]:
-                start_time, end_time = chunk["timestamp"]
-                break
+    # Transcription brute (sans timestamps, par exemple)
+    raw_transcription = " ".join([w["text"] for w in chunks])
 
-        updated_table.append([text, start_time, end_time, row[3]])
+    # On conserve le chemin de l'audio dans un State (pour la découpe ultérieure)
+    return raw_transcription, [], audio_path
 
-    return updated_table
 
-# -------------------------------------------------
-# Étape 5 : Validation + découpe
-# -------------------------------------------------
+# ------------------------
+# Étape 5 : Validation des segments + découpe
+# ------------------------
 def validate_segments(audio_path, table_data, metadata_state):
-    """Découpe l'audio en fonction des segments validés et met à jour l'état."""
-    if not audio_path or not table_data:
-        return [None] * 20, metadata_state
+    """
+    1) Pour chaque ligne du tableau, on découpe l'audio
+    2) On stocke les chemins des extraits dans 'metadata_state'
+    3) On renvoie une liste de chemins pour écoute sur l'interface
+    """
+    if not audio_path:
+        return ["Aucun fichier audio..."], metadata_state
 
-    # Nettoyer le dossier temporaire
+    # Nettoyage du dossier temporaire avant de recréer les extraits
     if os.path.exists(TEMP_DIR):
         shutil.rmtree(TEMP_DIR)
     os.makedirs(TEMP_DIR, exist_ok=True)
 
+    # Charger l'audio complet pour la découpe
     original_audio = AudioSegment.from_file(audio_path)
 
-    output_paths = [None] * 20
+    segment_paths = []
     updated_metadata = []
 
     for i, row in enumerate(table_data):
-        text, start_time, end_time, segment_id = row
-
-        # Vérifier si les données de la ligne sont valides
-        if not text or start_time is None or end_time is None:
+        # row = [ Texte, Start, End, ID ] (4 colonnes)
+        if len(row) < 4:
+            # S'il n'y a pas toutes les colonnes, on ignore
             continue
-        if not segment_id:
-            segment_id = f"seg_{i + 1:02d}"
-
-        try:
-            start_ms = int(float(start_time) * 1000)
-            end_ms = int(float(end_time) * 1000)
-        except (ValueError, TypeError):
-            continue  # Ignorer les lignes avec des timestamps invalides
 
-        if start_ms < 0 or end_ms <= start_ms:
+        segment_text, start_time, end_time, seg_id = row
+        if not segment_text or start_time is None or end_time is None:
+            # Ignore ligne vide ou incomplète
             continue
 
-        segment_filename = f"{Path(audio_path).stem}_{segment_id}.wav"
-        segment_path = os.path.join(TEMP_DIR, segment_filename)
+        # Générer un ID si l'utilisateur ne l'a pas renseigné
+        if not seg_id:
+            seg_id = f"seg_{i+1:02d}"
 
-        # Découpe et export
+        # Découpe réelle de l'audio
+        start_ms = int(float(start_time) * 1000)
+        end_ms = int(float(end_time) * 1000)
         extract = original_audio[start_ms:end_ms]
-        extract.export(segment_path, format="wav")
 
-        output_paths[i] = segment_path
+        segment_filename = f"{Path(audio_path).stem}_{seg_id}.wav"
+        segment_filepath = os.path.join(TEMP_DIR, segment_filename)
+        extract.export(segment_filepath, format="wav")
+
+        segment_paths.append(segment_filepath)
+
+        # Stocker la méta (pour le futur CSV)
         updated_metadata.append({
             "audio_file": segment_filename,
-            "text": text,
+            "text": segment_text,
             "start_time": start_time,
             "end_time": end_time,
-            "id": segment_id,
+            "id": seg_id
         })
 
-    return output_paths, updated_metadata
+    # Mettre à jour le State
+    # (Dans Gradio, on renvoie la nouvelle valeur)
+    return segment_paths, updated_metadata
+
 
-# -------------------------------------------------
-# Étape 8 : Génération du ZIP
-# -------------------------------------------------
+# ------------------------
+# Étape 7 : Génération du ZIP (avec metadata.csv)
+# ------------------------
 def generate_zip(metadata_state):
-    """Génère un fichier ZIP contenant les segments audio et un fichier metadata.csv."""
+    """
+    1) Crée le fichier 'metadata.csv'
+    2) Zip tous les extraits + metadata.csv
+    3) Renvoie le chemin du ZIP pour téléchargement
+    """
     if not metadata_state:
         return None
 
     zip_path = os.path.join(TEMP_DIR, "dataset.zip")
+    # Supprimer le zip précédent si existe
     if os.path.exists(zip_path):
         os.remove(zip_path)
 
-    # Créer le fichier metadata.csv
+    # Créer le CSV
     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:
+            # Exemple de speaker_name et API
+            # A adapter selon tes besoins
             line = f"{seg['audio_file']}|{seg['text']}|projectname|/API_PHONETIC/\n"
             f.write(line)
 
-    # Ajouter les fichiers au ZIP
+    # Créer le ZIP final
     with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
-        zf.write(metadata_csv_path, "metadata.csv")
+        # Ajouter les segments
         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"])
+            segment_file = os.path.join(TEMP_DIR, seg["audio_file"])
+            if os.path.exists(segment_file):
+                zf.write(segment_file, seg["audio_file"])
+        # Ajouter le CSV
+        zf.write(metadata_csv_path, "metadata.csv")
 
     return zip_path
 
-# -------------------------------------------------
-# Interface Gradio
-# -------------------------------------------------
+
+# ------------------------
+# Construction de l'interface Gradio
+# ------------------------
 with gr.Blocks(css="style.css") as demo:
-    gr.Markdown("# Application de Découpe Audio (jusqu'à 20 segments)")
+    gr.Markdown("# Application de découpe audio + Transcription Whisper")
 
-    # État global
+    # State pour conserver l'audio path + la liste de metadata
     metadata_state = gr.State(init_metadata_state())
 
-    # Étape 1 : Téléversement de l'audio
-    audio_input = gr.Audio(type="filepath", label="Fichier audio")
-
-    # Étape 2 : Transcription brute
-    raw_transcription = gr.Textbox(label="Transcription brute", interactive=False)
+    # Étape 1 : Choix de l'audio
+    with gr.Column():
+        gr.Markdown("### 1. Téléversez votre fichier audio (MP3/WAV)")
+        audio_input = gr.Audio(
+            type="filepath", label="Fichier audio"
+        )
+
+    # Étape 3 : Affichage transcription brute
+    raw_transcription = gr.Textbox(
+        label="Transcription brute (générée par Whisper)",
+        placeholder="La transcription apparaîtra ici",
+        interactive=False
+    )
 
-    # Étape 3 : Tableau des segments
+    # Étape 4 : Tableau pour choisir les extraits
+    gr.Markdown("### 2. Définissez vos segments")
+    gr.Markdown("**Colonne 1** : Le texte (copiez-coller depuis la transcription si besoin)  
+**Colonne 2** : Temps de début (en secondes)  
+**Colonne 3** : Temps de fin (en secondes)  
+**Colonne 4** : ID de votre segment (optionnel).")
     table = gr.Dataframe(
         headers=["Texte", "Début (s)", "Fin (s)", "ID"],
         datatype=["str", "number", "number", "str"],
-        row_count=20,
-        col_count=4,
+        row_count=5,
+        col_count=4
     )
-
-    # Étape intermédiaire : Générer les timestamps
-    generate_timestamps_button = gr.Button("Générer les timestamps")
-
-    # Bouton pour valider et générer les segments
     validate_button = gr.Button("Valider et générer les extraits")
-    audio_outputs = [gr.Audio(label=f"Extrait {i+1}", interactive=False) for i in range(20)]
 
-    # Bouton pour générer le fichier ZIP
-    zip_button = gr.Button("Générer le fichier ZIP")
-    zip_output = gr.File(label="Télécharger le ZIP")
-
-    # ----------------
-    # Callbacks
-    # ----------------
+    # Étape 6 : Écoute des extraits
+    # On prévoit jusqu'à 5 extraits (correspondant à row_count=5)
+    # Si tu prévois plus, augmente simplement ce bloc
+    audio_player_1 = gr.Audio(label="Extrait 1", interactive=False)
+    audio_player_2 = gr.Audio(label="Extrait 2", interactive=False)
+    audio_player_3 = gr.Audio(label="Extrait 3", interactive=False)
+    audio_player_4 = gr.Audio(label="Extrait 4", interactive=False)
+    audio_player_5 = gr.Audio(label="Extrait 5", interactive=False)
+
+    # Pour la sortie 'Valider segments', on veut 2 retours :
+    # - la liste des chemins (jusqu'à 5)
+    # - la metadata mise à jour
+    # => On va mapper ces chemins sur les 5 players
+    def update_audio_players(segments):
+        """
+        Prend la liste des chemins d'extraits audio (taille variable)
+        et renvoie une liste de 5 valeurs (None si pas d'extrait)
+        """
+        max_slots = 5
+        audio_values = [None]*max_slots
+        for i, seg in enumerate(segments):
+            if i < max_slots:
+                audio_values[i] = seg
+        return tuple(audio_values)
+
+    # Étape 8 : Génération ZIP
+    generate_button = gr.Button("Générer le fichier ZIP")
+    zip_file = gr.File(label="Télécharger le ZIP")
+
+    # ------------------------
+    # Logique de callbacks
+    # ------------------------
+
+    # 1) Callback quand on charge l'audio
     audio_input.change(
         fn=transcribe_audio,
         inputs=audio_input,
-        outputs=[raw_transcription, table, audio_input],
-    )
-
-    generate_timestamps_button.click(
-        fn=generate_timestamps,
-        inputs=[audio_input, table],
-        outputs=table,
+        outputs=[raw_transcription, table, audio_input],  # On renvoie le path en 3e
     )
 
+    # 2) Callback quand on valide les segments => on découpe
     validate_button.click(
         fn=validate_segments,
         inputs=[audio_input, table, metadata_state],
-        outputs=audio_outputs + [metadata_state],
+        outputs=[  # On reçoit segment_paths, updated_metadata
+            [audio_player_1, audio_player_2, audio_player_3, audio_player_4, audio_player_5],
+            metadata_state
+        ],
+        _js="(p) => { /* rien en JS */ }",
+        # On va quand même mapper les segments sur 5 players python-side
+        post_process=True
+    ).then(
+        fn=update_audio_players,
+        inputs=None,  # Pas besoin : la sortie segments est déjà captée par les players
+        outputs=[audio_player_1, audio_player_2, audio_player_3, audio_player_4, audio_player_5],
     )
 
-    zip_button.click(
+    # 3) Callback pour générer le ZIP
+    generate_button.click(
         fn=generate_zip,
         inputs=metadata_state,
-        outputs=zip_output,
+        outputs=zip_file
     )
 
 demo.queue().launch()