#!/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)