Something went wrong. Try again.
Something went wrong. Try again.
2.1 kB · 61 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162"""ElevenLabs Scribe v2 transcription with diarization.
Sends the original MP3 in one request so Scribe keeps unbroken voiceprintcontinuity across the whole episode. Cost comes back in response headers(fiat-cost-before-overages) -- we record it per episode."""import os
import httpx
import config
API = "https://api.elevenlabs.io/v1/speech-to-text"
def transcribe(mp3_path, num_speakers=None): """Returns (scribe_json, cost_usd).""" num_speakers = num_speakers or config.NUM_SPEAKERS key = os.environ["ELEVEN_LABS_API_TOKEN"] with open(mp3_path, "rb") as fh: r = httpx.post( API, headers={"xi-api-key": key}, files={"file": (os.path.basename(mp3_path), fh, "audio/mpeg")}, data={"model_id": "scribe_v2", "diarize": "true", "num_speakers": str(num_speakers), "timestamps_granularity": "word", "tag_audio_events": "true", "language_code": "eng"}, timeout=3600, ) r.raise_for_status() cost = float(r.headers.get("fiat-cost-before-overages", 0) or 0) return r.json(), cost
def words_to_turns(scribe_json): """Collapse the word stream into speaker turns.
Keeps [audio events] inline. A turn boundary is a speaker change. Returns [{t, end, spk, text}] with t/end in float seconds. """ turns, cur = [], None for w in scribe_json.get("words", []): kind, txt, spk = w.get("type"), w.get("text", ""), w.get("speaker_id") if kind == "spacing": if cur: cur["text"] += txt continue if kind not in ("word", "audio_event"): continue if cur is None or cur["spk"] != spk: cur = {"spk": spk, "t": w["start"], "end": w["end"], "text": txt} turns.append(cur) else: if not cur["text"].endswith(" "): cur["text"] += " " cur["text"] += txt cur["end"] = w["end"] for t in turns: t["text"] = " ".join(t["text"].split()) return [t for t in turns if t["text"]]