Update app.py

app.py

@@ -288,53 +288,41 @@ def segment_audio_from_video(video_path):
 
     segment_result, speech_audio_path = segment_background_audio(audio_path)
     print(f"Saved non-speech (background) audio to local")
-    
-    # Set up device
+
     device = "cuda" if torch.cuda.is_available() else "cpu"
     logger.info(f"Using device: {device}")
-    
+
     try:
-        # Load a medium model with float32 for broader compatibility
         model = whisperx.load_model("large-v3", device=device, compute_type="float32")
         logger.info("WhisperX model loaded")
-    
-        # Transcribe
-        result = model.transcribe(speech_audio_path, chunk_size=4, print_progress = True)
+        result = model.transcribe(speech_audio_path, chunk_size=4, print_progress=True)
         logger.info("Audio transcription completed")
-
     except Exception as e:
         logger.error(f"❌ WhisperX pipeline failed: {e}")
+        return audio_path, []
 
-    # Extract timestamps, text, and speaker IDs
+    # Return segment boundaries (only timestamps, not text)
     transcript_with_speakers = [
         {
             "start": segment["start"],
             "end": segment["end"]
         }
         for segment in result["segments"]
+        if segment["end"] > segment["start"]
     ]
 
     return audio_path, transcript_with_speakers
 
-def transcribe_segments_with_scribe(full_audio_path, segments):
-    """
-    Transcribes pre-defined audio segments using the ElevenLabs Scribe API.
-    Diarization is explicitly turned off as per requirements.
+def clean_transcribed_text(text: str) -> str:
+    """Remove repetitive symbols and artifacts from text."""
+    # Remove only-punctuation or repeated tokens
+    cleaned = re.sub(r"[_,.~`^•·。！？!?,，\.\/\\\-–—=+]+", " ", text)
+    cleaned = re.sub(r"\s+", " ", cleaned).strip()
+    return cleaned
 
-    Args:
-        full_audio_path (str): The path to the full extracted audio file.
-        segments (list): A list of dictionaries, where each dictionary
-                         represents a segment with 'start' and 'end' timestamps in seconds.
-
-    Returns:
-        tuple: A tuple containing:
-            - transcribed_segments (list): A list of dictionaries, where each dictionary
-              represents a transcribed segment with 'start', 'end', and 'text'.
-            - detected_language (str): The language detected by the API (e.g., "en", "es").
-            - error_message (str, optional): An error message if transcription fails.
-    """
+def transcribe_segments_with_scribe(full_audio_path, segments):
     transcribed_segments = []
-    detected_language = "unknown" # Default
+    detected_language = "unknown"
     error_message = None
 
     if not os.path.exists(full_audio_path):
@@ -342,94 +330,63 @@ def transcribe_segments_with_scribe(full_audio_path, segments):
 
     try:
         audio_clip = AudioFileClip(full_audio_path)
-        
-        headers = {
-            "xi-api-key": ELEVENLABS_API_KEY,
-        }
-        data = {
-            "model_id": "scribe_v1",
-        }
-        # Explicitly set diarize to false, as it's not needed.
-        params = {
-            "diarize": "false",
-        }
+        headers = {"xi-api-key": ELEVENLABS_API_KEY}
+        data = {"model_id": "scribe_v1"}
+        params = {"diarize": "false"}
 
         logger.info(f"Starting transcription of {len(segments)} segments with ElevenLabs Scribe...")
 
         for i, segment in enumerate(segments):
-            segment_start = segment["start"]
-            segment_end = segment["end"]
-            
-            # Ensure segment duration is positive
-            if segment_end <= segment_start:
-                logger.warning(f"Skipping segment {i} due to invalid duration: {segment_start:.2f}s -> {segment_end:.2f}s")
+            start, end = segment["start"], segment["end"]
+            if end <= start:
+                logger.warning(f"Skipping invalid segment {i}: {start:.2f}s → {end:.2f}s")
                 continue
 
             temp_segment_audio_path = f"temp_segment_{i}.wav"
             try:
-                # Subclip the audio and save it temporarily
-                sub_clip = audio_clip.subclip(segment_start, segment_end)
-                # Save as 16-bit PCM WAV for Scribe API compatibility
+                sub_clip = audio_clip.subclip(start, end)
                 sub_clip.write_audiofile(temp_segment_audio_path, codec='pcm_s16le')
-                
-                logger.info(f"Transcribing segment {i+1}/{len(segments)}: {segment_start:.2f}s - {segment_end:.2f}s")
 
                 with open(temp_segment_audio_path, "rb") as audio_file:
-                    files = {
-                        "file": (os.path.basename(temp_segment_audio_path), audio_file, "audio/wav")
-                    }
+                    files = {"file": (os.path.basename(temp_segment_audio_path), audio_file, "audio/wav")}
                     response = requests.post(ELEVENLABS_SCRIBE_API_URL, headers=headers, files=files, data=data, params=params)
                     response.raise_for_status()
                     scribe_result = response.json()
 
-                    segment_text = ""
-                    if "text" in scribe_result:
-                        segment_text = scribe_result["text"].strip()
-                    elif "words" in scribe_result and scribe_result["words"]:
-                        # Fallback if 'text' field is not directly available, reconstruct from words
-                        segment_text = " ".join([w.get("text", "") for w in scribe_result["words"] if w.get("type") == "word"]).strip()
-                    
-                    if segment_text:
+                    raw_text = scribe_result.get("text") or " ".join(
+                        [w.get("text", "") for w in scribe_result.get("words", []) if w.get("type") == "word"]
+                    )
+
+                    cleaned_text = clean_transcribed_text(raw_text)
+                    if cleaned_text:
                         transcribed_segments.append({
-                            "start": segment_start,
-                            "end": segment_end,
-                            "text": segment_text,
+                            "start": start,
+                            "end": end,
+                            "text": cleaned_text,
                             "speaker": "SPEAKER_00"
                         })
                     else:
-                        logger.warning(f"No transcription text found for segment {i+1}.")
+                        logger.info(f"Segment {i+1} discarded: cleaned text is empty.")
 
-                    # Update detected language from the first successful transcription
                     if "language_code" in scribe_result and detected_language == "unknown":
                         detected_language = scribe_result["language_code"]
 
-            except requests.exceptions.HTTPError as http_err:
-                error_message = f"HTTP error for segment {i+1}: {http_err} - {response.text}"
-                logger.error(error_message)
-                # Continue to next segment even if one fails
-            except requests.exceptions.RequestException as req_err:
-                error_message = f"Request error for segment {i+1}: {req_err}"
-                logger.error(error_message)
-                # Continue to next segment
             except Exception as e:
-                error_message = f"Error processing segment {i+1}: {e}"
-                logger.error(error_message)
-                # Continue to next segment
+                logger.error(f"Error processing segment {i+1}: {e}")
             finally:
                 if os.path.exists(temp_segment_audio_path):
                     os.remove(temp_segment_audio_path)
-                    
+
         logger.info("All segments processed by ElevenLabs Scribe.")
 
     except Exception as e:
-        error_message = f"An error occurred during overall transcription process: {e}"
+        error_message = f"An error occurred: {e}"
         logger.error(error_message)
     finally:
-        if 'audio_clip' in locals() and audio_clip is not None:
+        if 'audio_clip' in locals():
             audio_clip.close()
 
     return transcribed_segments, detected_language, error_message
-
     
 # Function to get the appropriate translation model based on target language
 def get_translation_model(source_language, target_language):