Update app.py

app.py

@@ -29,7 +29,6 @@ def get_bad_words():
     except Exception as e:
         print(f"⚠️ Failed to fetch list: {e}")
 
-    # fallback local list
     fallback = {"fuck", "shit", "bitch", "ass", "nigga", "nigger", "pussy", "cunt"}
     print(f"⚠️ Using fallback list ({len(fallback)} words).")
     return fallback
@@ -43,30 +42,7 @@ print(f"🚀 Loading LARGE Whisper model: {MODEL_NAME} ({COMPUTE_TYPE}) on {DEVI
 large_model = WhisperModel(MODEL_NAME, device=DEVICE, compute_type=COMPUTE_TYPE)
 print("✅ Models ready!")
 
-# === HELPERS ===
-def merge_intervals(intervals, padding=0.15):
-    """Merge overlapping intervals; also expand each interval by padding seconds."""
-    if not intervals:
-        return []
-    # apply padding
-    intervals = [(max(0, s - padding), e + padding) for s, e in intervals]
-    intervals.sort(key=lambda x: x[0])
-    merged = []
-    cur_s, cur_e = intervals[0]
-    for s, e in intervals[1:]:
-        if s <= cur_e:
-            cur_e = max(cur_e, e)
-        else:
-            merged.append((cur_s, cur_e))
-            cur_s, cur_e = s, e
-    merged.append((cur_s, cur_e))
-    return merged
-
-def replace_range_in_list(lst, start_idx, end_idx, new_items):
-    """Replace lst[start_idx:end_idx] with new_items (in-place)."""
-    return lst[:start_idx] + new_items + lst[end_idx:]
-
-# === TRANSCRIBE FUNCTION (HYBRID) ===
+# === TRANSCRIBE FUNCTION (HYBRID WORD-LEVEL) ===
 def transcribe(file_path):
     # --- Ensure proper audio format (mono, 16k) ---
     wav, sr = torchaudio.load(file_path)
@@ -78,8 +54,8 @@ def transcribe(file_path):
     fixed_path = "input_fixed.wav"
     torchaudio.save(fixed_path, wav, target_sr)
 
-    # --- FAST PASS (cheap, detect possible explicit words) ---
-    print("⚡ Running fast (cheap) pass to detect candidate explicit words…")
+    # --- FAST PASS (cheap) ---
+    print("⚡ Running fast pass to detect candidate explicit words…")
     fast_segments, fast_info = fast_model.transcribe(
         fixed_path,
         beam_size=1,
@@ -88,7 +64,7 @@ def transcribe(file_path):
     )
     sample_rate = getattr(fast_info, "sample_rate", target_sr)
 
-    # Build initial transcript from fast pass
+    # Build initial transcript
     transcript = []
     for seg in fast_segments:
         if hasattr(seg, "words") and seg.words:
@@ -103,7 +79,6 @@ def transcribe(file_path):
                     "explicit": word_text.lower() in BAD_WORDS
                 })
         else:
-            # fallback: segment-level entry
             transcript.append({
                 "text": seg.text,
                 "start": float(seg.start),
@@ -111,40 +86,34 @@ def transcribe(file_path):
                 "explicit": False
             })
 
-    # --- Determine flagged intervals to re-run with large model ---
-    flagged_intervals = [(w["start"], w["end"]) for w in transcript if w.get("explicit")]
-    merged_intervals = merge_intervals(flagged_intervals, padding=0.15)
-    print(f"🔎 Fast pass flagged {len(flagged_intervals)} words -> {len(merged_intervals)} merged intervals")
-
-    # --- SECOND PASS (large model) on flagged words only ---
-    if flagged_intervals:
+    # --- SECOND PASS: large model on explicit words only ---
+    flagged_words = [t for t in transcript if t.get("explicit")]
+    if flagged_words:
+        print(f"🔎 Fast pass flagged {len(flagged_words)} explicit words — refining with large model…")
         refined_entries = []
-        for idx, w in enumerate([t for t in transcript if t.get("explicit")]):
+
+        for idx, w in enumerate(flagged_words):
             s, e = w["start"], w["end"]
-            print(f"⏱️ Refining explicit word {idx+1}/{len(flagged_intervals)}: {s:.2f}s -> {e:.2f}s")
-    
-            # extract the single word chunk
+            print(f"⏱️ Refining word {idx+1}/{len(flagged_words)}: {s:.2f}s -> {e:.2f}s")
+
             start_sample = int(max(0, s * sample_rate))
             end_sample = int(min(wav.shape[-1], e * sample_rate))
             num_frames = max(0, end_sample - start_sample)
             if num_frames == 0:
                 continue
             chunk = wav[:, start_sample:end_sample]
-    
-            # write temp file
+
             with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
                 temp_path = tmp.name
             torchaudio.save(temp_path, chunk, sample_rate)
-    
-            # transcribe chunk with large model
+
             segs, _ = large_model.transcribe(
                 temp_path,
                 beam_size=5,
                 word_timestamps=True,
                 vad_filter=True
             )
-    
-            # adjust chunk-relative timestamps to original
+
             for seg in segs:
                 if hasattr(seg, "words") and seg.words:
                     for word_obj in seg.words:
@@ -161,58 +130,24 @@ def transcribe(file_path):
                         "end": float(seg.end) + s,
                         "explicit": False
                     })
-    
+
             try:
                 os.remove(temp_path)
             except Exception:
                 pass
 
-    # Merge refined words back into transcript
-    final_transcript = []
-    i = 0
-    for t in transcript:
-        if t.get("explicit"):
-            # replace flagged word with refined version
-            refined_word = refined_entries.pop(0)
-            final_transcript.append(refined_word)
-        else:
-            final_transcript.append(t)
-    transcript = final_transcript
-
-
-        # --- Merge refined entries into the original fast transcript ---
-        # For each merged interval, replace overlapping fast-pass items with refined items for that interval.
+        # Merge refined words back into transcript
         final_transcript = []
-        i = 0
-        n = len(transcript)
-        for interval in merged_intervals:
-            int_s, int_e = interval
-            # append all fast-pass items before this interval
-            while i < n and transcript[i]["end"] <= int_s:
-                final_transcript.append(transcript[i])
-                i += 1
-            # skip fast-pass items that overlap the interval
-            skip_start = i
-            while i < n and transcript[i]["start"] < int_e:
-                i += 1
-            skip_end = i
-            # collect refined entries for this interval (those whose times fall into interval)
-            refined_for_interval = [r for r in refined_entries if not (r["end"] <= int_s or r["start"] >= int_e)]
-            # sort refined entries
-            refined_for_interval.sort(key=lambda x: x.get("start", 0))
-            # append refined entries instead of the skipped fast-pass items
-            final_transcript.extend(refined_for_interval)
-        # append any remaining fast-pass items after all intervals
-        while i < n:
-            final_transcript.append(transcript[i])
-            i += 1
-
-        # If no merged_intervals matched anything, fall back to initial transcript
-        transcript = final_transcript if final_transcript else transcript
+        for t in transcript:
+            if t.get("explicit") and refined_entries:
+                final_transcript.append(refined_entries.pop(0))
+            else:
+                final_transcript.append(t)
+        transcript = final_transcript
     else:
-        print("✅ No flagged intervals — skipping large-model refinement.")
+        print("✅ No flagged words — skipping large-model refinement.")
 
-    # --- final housekeeping: if transcript is empty, build segment-level fallback from fast pass segments ---
+    # --- fallback if transcript empty ---
     if not transcript:
         transcript = [{
             "text": seg.text,
@@ -231,7 +166,7 @@ iface = gr.Interface(
     inputs=gr.Audio(type="filepath", label="Upload Vocals"),
     outputs=gr.JSON(label="Transcript with Explicit Flags"),
     title="CleanSong AI — Whisper Transcriber (Hybrid Fast→Accurate)",
-    description="Two-pass transcription: fast model to detect explicit words, large model to refine only flagged intervals."
+    description="Two-pass transcription: fast model to detect explicit words, large model to refine only flagged words."
 )
 
 if __name__ == "__main__":