IT狗
9a93cb6ffd
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
347 lines
15 KiB
Python
347 lines
15 KiB
Python
#!/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",
|
||
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,
|
||
# ✅ 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:
|
||
# 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
|
||
|
||
|
||
@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()
|
||
# 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, 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):
|
||
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:
|
||
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({
|
||
"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)
|