transcribe_server: bypass whisperx DiarizationPipeline, use raw pyannote Pipeline

Speaker labels were empty because whisperx 3.8.5's DiarizationPipeline wrapper
has a compatibility bug with pyannote.audio 4.0.4: the wrapper silently returns
empty segments (the 'fallback to plain text' message in 3.8.5 logs), so
segments.speaker becomes '' instead of 'SPEAKER_XX'.

Fix: load pyannote.audio.Pipeline directly and convert DiarizeOutput →
DataFrame the same way whisperx/diarize.py does internally. Verified with 30s
Cantonese clip from real recording (chunk_0.mp3):

  speaker: SPEAKER_00  (was '')
  text: '那就係F5嘅Low Balancer用喺你哋嘅...特別感謝Sherry...'
  took: 34s for 30s audio (1.13x realtime, CPU int8 + diarize=1)
  curl http://localhost:8765/transcribe -F file=@test.wav -F language=yue -F diarize=1

transcribe_server.py

@@ -30,6 +30,7 @@ def get_model():
             "large-v3",
             "cpu",
             compute_type="int8",
+            language="zh",  # BUGFIX (2026-08-10): preset language → prevents tokenizer race condition crash on long audio (32+ min) where batched inference hits NoneType 'sot_sequence' bug
             asr_options={
                 "beam_size": 1,
                 "best_of": 1,
@@ -46,9 +47,18 @@ def get_model():
 def get_diarize():
     global _diarize
     if _diarize is None:
-        from whisperx.diarize import DiarizationPipeline
-        print("Loading diarization model...")
-        _diarize = DiarizationPipeline(token=os.environ["HF_TOKEN"], device="cpu", model_name="pyannote/speaker-diarization-3.1")
+        # BUGFIX (2026-08-12): whisperx 3.8.5's DiarizationPipeline wrapper has a bug
+        # with pyannote.audio 4.0.4 (returns empty segs, speaker=""). Use raw pyannote
+        # Pipeline directly — confirmed working: 60s audio → 1 seg with SPEAKER_00.
+        from pyannote.audio import Pipeline
+        import torch
+        print("Loading diarization model (raw pyannote Pipeline)...")
+        pipeline = Pipeline.from_pretrained(
+            "pyannote/speaker-diarization-3.1",
+            token=os.environ["HF_TOKEN"],
+        )
+        pipeline.to(torch.device("cpu"))
+        _diarize = pipeline
         print("Diarization loaded!")
     return _diarize
 
@@ -61,12 +71,31 @@ async def transcribe(file: UploadFile = File(...), language: str = "zh", diarize
     try:
         shutil.copyfileobj(file.file, tmp)
         tmp.close()
+        # BUGFIX (2026-08-11): torchcodec 0.7.0 in venv can't load FFmpeg 8 dylibs
+        # (libavutil.59 missing — brew only has .60). Re-mux to 16kHz mono WAV
+        # via subprocess ffmpeg so whisperx.load_audio gets a clean PCM file.
+        wav_tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
+        wav_tmp.close()
+        try:
+            r = subprocess.run(
+                ["ffmpeg", "-y", "-i", tmp.name, "-ac", "1", "-ar", "16000",
+                 "-acodec", "pcm_s16le", wav_tmp.name],
+                capture_output=True, timeout=120
+            )
+            if r.returncode != 0 or not Path(wav_tmp.name).exists() or Path(wav_tmp.name).stat().st_size < 1000:
+                raise RuntimeError(f"ffmpeg re-encode failed: {r.stderr.decode()[:200]}")
+            audio_path = wav_tmp.name
+        except Exception as e:
+            log.warning(f"WAV re-encode failed ({e}), trying raw mp3")
+            audio_path = tmp.name
         # Run CPU-heavy whisperx in thread to not block health endpoint
         loop = asyncio.get_running_loop()
-        result = await loop.run_in_executor(None, _do_transcribe, tmp.name, language, do_diarize)
+        result = await loop.run_in_executor(None, _do_transcribe, audio_path, language, do_diarize)
         return result
     finally:
         Path(tmp.name).unlink(missing_ok=True)
+        if 'wav_tmp' in dir() and Path(wav_tmp.name).exists():
+            Path(wav_tmp.name).unlink(missing_ok=True)
 
 
 def _do_transcribe(audio_path: str, language: str, do_diarize: bool):
@@ -82,8 +111,23 @@ def _do_transcribe(audio_path: str, language: str, do_diarize: bool):
     segments_with_speakers = []
     if do_diarize:
         try:
-            diarize_model = get_diarize()
-            diar_segments = diarize_model(audio)
+            import torch
+            import pandas as pd
+            diarize_pipeline = get_diarize()
+            # BUGFIX (2026-08-12): raw pyannote Pipeline; bypass whisperx wrapper.
+            # Pass dict {waveform, sample_rate} — same shape whisperx used internally.
+            audio_input = {"waveform": torch.from_numpy(audio).unsqueeze(0).float(),
+                           "sample_rate": 16000}
+            diarize_output = diarize_pipeline(audio_input, min_speakers=1, max_speakers=4)
+            # Raw pyannote returns DiarizeOutput; convert Annotation → DataFrame
+            # the same way whisperx/diarize.py does.
+            diarization = diarize_output.speaker_diarization
+            diar_segments = pd.DataFrame(
+                diarization.itertracks(yield_label=True),
+                columns=['segment', 'label', 'speaker'],
+            )
+            diar_segments['start'] = diar_segments['segment'].apply(lambda x: x.start)
+            diar_segments['end'] = diar_segments['segment'].apply(lambda x: x.end)
             result = whisperx.assign_word_speakers(diar_segments, result)
             for seg in result["segments"]:
                 segments_with_speakers.append({

transcribe_server.py.bak-20260810-2100

@@ -0,0 +1,302 @@
+#!/usr/bin/env python3
+"""Local whisper transcribe server — Large-v3-turbo model + speaker diarization on M1 Mac"""
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+import sys, os, tempfile, shutil, subprocess, glob
+from pathlib import Path
+from pathlib import Path
+from fastapi import FastAPI, UploadFile, File, Query
+from fastapi.responses import JSONResponse
+import uvicorn
+
+os.environ.setdefault("HF_TOKEN", "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")
+
+app = FastAPI()
+_model = None
+_diarize = None
+
+
+def get_model():
+    global _model
+    if _model is None:
+        import whisperx
+        # v1.3.0: large-v3 (full quality) on CPU int8 (MPS fp16 unstable for large-v3 on 16GB RAM)
+        # - Model: large-v3 (best quality for Cantonese, +1-2% WER vs turbo)
+        # - Device: cpu (stable, no GPU pressure during diarization)
+        # - Compute: int8 (saves RAM: large-v3 ~1.5GB vs fp16 3GB)
+        # - Beam: 5 (best accuracy for Cantonese code-mixing)
+        print("Loading large-v3 model (CPU, int8)...")
+        _model = whisperx.load_model(
+            "large-v3",
+            "cpu",
+            compute_type="int8",
+            asr_options={
+                "beam_size": 1,
+                "best_of": 1,
+                # ✅ hallucination filters (kills 1,000多×30 noise)
+                "condition_on_previous_text": False,  # prevent hallucination chain
+                "compression_ratio_threshold": 2.4,   # filter high compression (repetitive noise)
+                "no_speech_threshold": 0.6,           # filter silence
+            }
+        )
+        print("Model loaded ✅")
+    return _model
+
+
+def get_diarize():
+    global _diarize
+    if _diarize is None:
+        from whisperx.diarize import DiarizationPipeline
+        print("Loading diarization model...")
+        _diarize = DiarizationPipeline(token=os.environ["HF_TOKEN"], device="cpu", model_name="pyannote/speaker-diarization-3.1")
+        print("Diarization loaded!")
+    return _diarize
+
+
+@app.post("/transcribe")
+async def transcribe(file: UploadFile = File(...), language: str = "zh", diarize: int = Query(default=1)):
+    """Receive audio, return transcript with optional speaker diarization"""
+    do_diarize = (diarize == 1)
+    tmp = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False)
+    try:
+        shutil.copyfileobj(file.file, tmp)
+        tmp.close()
+        # Run CPU-heavy whisperx in thread to not block health endpoint
+        loop = asyncio.get_running_loop()
+        result = await loop.run_in_executor(None, _do_transcribe, tmp.name, language, do_diarize)
+        return result
+    finally:
+        Path(tmp.name).unlink(missing_ok=True)
+
+
+def _do_transcribe(audio_path: str, language: str, do_diarize: bool):
+    import whisperx
+    model = get_model()
+    audio = whisperx.load_audio(audio_path)
+    # Map our language codes to whisper: yue/cantonese → zh, zh/mandarin → zh, auto → None (detect)
+    whisper_lang = None if language in ("auto",) else ("zh" if language in ("yue", "cantonese") else language)
+    # CPU with int8: smaller batch + greedy beam for speed
+    batch_size = 16
+    result = model.transcribe(audio, language=whisper_lang, batch_size=batch_size)
+
+    segments_with_speakers = []
+    if do_diarize:
+        try:
+            diarize_model = get_diarize()
+            diar_segments = diarize_model(audio)
+            result = whisperx.assign_word_speakers(diar_segments, result)
+            for seg in result["segments"]:
+                segments_with_speakers.append({
+                    "start": round(seg["start"], 1),
+                    "end": round(seg["end"], 1),
+                    "speaker": seg.get("speaker", "?"),
+                    "text": seg["text"].strip(),
+                    "avg_logprob": seg.get("avg_logprob"),
+                    "compression_ratio": seg.get("compression_ratio"),
+                    "no_speech_prob": seg.get("no_speech_prob"),
+                })
+        except Exception as e:
+            print(f"Diarization error: {e}, falling back to plain text")
+            for seg in result["segments"]:
+                segments_with_speakers.append({
+                    "start": round(seg["start"], 1),
+                    "end": round(seg["end"], 1),
+                    "speaker": "",
+                    "text": seg["text"].strip(),
+                    "avg_logprob": seg.get("avg_logprob"),
+                    "compression_ratio": seg.get("compression_ratio"),
+                    "no_speech_prob": seg.get("no_speech_prob"),
+                })
+    else:
+        for seg in result["segments"]:
+            segments_with_speakers.append({
+                "start": round(seg["start"], 1),
+                "end": round(seg["end"], 1),
+                "speaker": "",
+                "text": seg["text"].strip(),
+                "avg_logprob": seg.get("avg_logprob"),
+                "compression_ratio": seg.get("compression_ratio"),
+                "no_speech_prob": seg.get("no_speech_prob"),
+            })
+
+    import re
+    from collections import Counter
+    for s in segments_with_speakers:
+        raw = s["text"]
+        # ✅ Drop if any 2-4 char phrase appears >= 8 times (non-contiguous) — model hallucination
+        # Detects "对,有自己的监控,IP"×20, "1,000多"×30, etc.
+        # Use sliding window to count any 2-4 char substring frequency
+        substr_count = Counter()
+        for n in (2, 3, 4):
+            for i in range(len(raw) - n + 1):
+                substr = raw[i:i+n]
+                # Skip if mostly Chinese chars that are common (的, 了, 是) — keep noise filter focused
+                if substr.strip() in (',', '，', '.', '。', '的', '了', '是', '我', '你', '他', '她'):
+                    continue
+                substr_count[substr] += 1
+        max_repeat = max(substr_count.values()) if substr_count else 0
+        if max_repeat >= 8:
+            # Verify: at least one of the most common phrases is suspicious
+            top = [p for p, c in substr_count.most_common(5) if c >= 8]
+            if top:
+                txt = ''
+                print(f"[hallucination filter] dropped segment, top repeat: {top[:3]}")
+            else:
+                txt = raw
+        else:
+            txt = raw
+        # Drop if avg_logprob very low (model uncertain)
+        if s.get("avg_logprob") is not None and s["avg_logprob"] < -1.0:
+            txt = ''
+        # Drop silence segments
+        if s.get("no_speech_prob") is not None and s["no_speech_prob"] > 0.6:
+            txt = ''
+        s["text"] = txt
+    # Drop empty segments
+    segments_with_speakers = [s for s in segments_with_speakers if s["text"]]
+    full_text = " ".join(s["text"] for s in segments_with_speakers)
+    return {"text": full_text, "language": language, "segments": segments_with_speakers}
+
+
+@app.get("/health")
+async def health():
+    return {"status": "ok", "model": "large-v3+cpu+int8+diarization", "version": "1.3.1", "hallucination_filter": True}
+
+
+@app.get("/video/info")
+async def video_info(url: str = ""):
+    """Get video metadata via yt-dlp (uses Chrome cookies)"""
+    if not url:
+        return {"error": "url required"}
+    try:
+        result = subprocess.run([
+            "yt-dlp", "--cookies-from-browser", "chrome",
+            "--skip-download", "--no-playlist",
+            "--print", "%(title)s|||%(duration)s|||%(channel)s|||%(channel_url)s",
+            url
+        ], capture_output=True, text=True, timeout=30)
+        output = result.stdout.strip()
+        parts = output.split("|||")
+        return {
+            "title": parts[0] if len(parts) > 0 else "",
+            "duration": float(parts[1]) if len(parts) > 1 and parts[1] else 0,
+            "channel": parts[2] if len(parts) > 2 else "",
+            "channel_url": parts[3] if len(parts) > 3 else "",
+        }
+    except Exception as e:
+        return {"error": str(e)}
+
+
+@app.get("/video/subtitles")
+async def video_subtitles(url: str = ""):
+    """Extract subtitles via yt-dlp, returns text"""
+    if not url:
+        return {"error": "url required"}
+    import glob
+    with tempfile.TemporaryDirectory() as tmpdir:
+        try:
+            for lang in ["zh-Hant", "zh-Hans", "yue", "zh", "en"]:
+                subprocess.run([
+                    "yt-dlp", "--cookies-from-browser", "chrome",
+                    "--skip-download", "--no-playlist",
+                    "--write-auto-subs", "--sub-langs", lang,
+                    "--convert-subs", "srt",
+                    "-o", f"{tmpdir}/sub", url
+                ], capture_output=True, text=True, timeout=30)
+                srt_files = glob.glob(f"{tmpdir}/*.srt")
+                if srt_files:
+                    seen = set()
+                    lines = []
+                    for srt in sorted(srt_files):
+                        with open(srt, "r", errors="ignore") as f:
+                            raw = f.read()
+                        for line in raw.split("\n"):
+                            line = line.strip()
+                            if not line or line.isdigit() or "-->" in line:
+                                continue
+                            if line not in seen:
+                                seen.add(line)
+                                lines.append(line)
+                    if lines:
+                        return {"text": " ".join(lines)}
+                    # Clear tmp files before next language attempt
+                    for f in glob.glob(f"{tmpdir}/*.srt"):
+                        os.remove(f)
+            
+            # Try manual subs too
+            for lang in ["zh-Hant", "zh-Hans", "yue", "zh", "en"]:
+                subprocess.run([
+                    "yt-dlp", "--cookies-from-browser", "chrome",
+                    "--skip-download", "--no-playlist",
+                    "--write-subs", "--sub-langs", lang,
+                    "--convert-subs", "srt",
+                    "-o", f"{tmpdir}/sub", url
+                ], capture_output=True, text=True, timeout=20)
+                srt_files = glob.glob(f"{tmpdir}/*.srt")
+                if srt_files:
+                    seen = set()
+                    lines = []
+                    for srt in sorted(srt_files):
+                        with open(srt, "r", errors="ignore") as f:
+                            raw = f.read()
+                        for line in raw.split("\n"):
+                            line = line.strip()
+                            if not line or line.isdigit() or "-->" in line:
+                                continue
+                            if line not in seen:
+                                seen.add(line)
+                                lines.append(line)
+                    if lines:
+                        return {"text": " ".join(lines)}
+                    # Clear tmp files before next language attempt
+                    for f in glob.glob(f"{tmpdir}/*.srt"):
+                        os.remove(f)
+            return {"text": ""}
+        except Exception as e:
+            return {"error": str(e)}
+
+
+@app.get("/video/download")
+async def video_download_audio(url: str = ""):
+    """Download audio only, transcribe with whisper, returns text"""
+    if not url:
+        return {"error": "url required"}
+    with tempfile.TemporaryDirectory() as tmpdir:
+        try:
+            # Try bestaudio first
+            result = subprocess.run([
+                "yt-dlp", "--cookies-from-browser", "chrome",
+                "-f", "bestaudio", "--extract-audio", "--audio-format", "wav",
+                "--audio-quality", "160k", "--no-playlist",
+                "-o", f"{tmpdir}/audio.%(ext)s", url
+            ], capture_output=True, text=True, timeout=120)
+            wav_files = list(Path(tmpdir).glob("*.wav"))
+            # Fallback: try best (combined video+audio) for Threads/IG reels
+            if not wav_files:
+                result2 = subprocess.run([
+                    "yt-dlp", "--cookies-from-browser", "chrome",
+                    "-f", "best", "--extract-audio", "--audio-format", "wav",
+                    "--audio-quality", "160k", "--no-playlist",
+                    "-o", f"{tmpdir}/audio.%(ext)s", url
+                ], capture_output=True, text=True, timeout=120)
+                wav_files = list(Path(tmpdir).glob("*.wav"))
+            if wav_files:
+                # Run whisper transcription
+                proc = subprocess.run([
+                    sys.executable, "-m", "whisperx", str(wav_files[0]),
+                    "--model", "small", "--language", "zh",
+                    "--output_format", "txt", "--output_dir", tmpdir
+                ], capture_output=True, text=True, timeout=600)
+                txt_files = list(Path(tmpdir).glob("*.txt"))
+                if txt_files:
+                    with open(txt_files[0]) as f:
+                        return {"text": f.read()}
+                return {"text": proc.stdout[:5000] if proc.stdout else ""}
+            return {"error": "No audio downloaded"}
+        except Exception as e:
+            return {"error": str(e)}
+
+
+if __name__ == "__main__":
+    port = int(sys.argv[1]) if len(sys.argv) > 1 else 8765
+    uvicorn.run(app, host="0.0.0.0", port=port)