feat(transcribe): add BW Labs STT as optional provider alongside ElevenLabs

.env.example

@@ -1 +1,10 @@
 ELEVENLABS_API_KEY=
+
+# Optional: BW Labs STT — get a key at https://labs.bandwidth.com/
+# API docs: https://labs.bandwidth.com/docs/speech-to-text
+# Install:  pip install websockets  (or: pip install "video-use[bw-stt]")
+BW_STT_API_KEY=
+
+# Optional: default transcription provider ("elevenlabs" or "bw_stt").
+# If unset, auto-detects from whichever key is present (ElevenLabs wins if both).
+# TRANSCRIBE_PROVIDER=elevenlabs

helpers/transcribe.py

@@ -1,16 +1,37 @@
-"""Transcribe a video with ElevenLabs Scribe.
+"""Transcribe a video with ElevenLabs Scribe (default) or BW Labs STT.
 
-Extracts mono 16kHz audio via ffmpeg, uploads to Scribe with verbatim +
-diarize + audio events + word-level timestamps, writes the full response
-to <edit_dir>/transcripts/<video_stem>.json.
+ElevenLabs Scribe: HTTP multipart upload, word-level timestamps, speaker
+diarization, and audio events.  Requires ELEVENLABS_API_KEY.
 
-Cached: if the output file already exists, the upload is skipped.
+BW Labs STT: WebSocket streaming, no duration limit, no speaker diarization,
+no language selection.  Requires BW_STT_API_KEY and the websockets library.
+  Get a key:  https://labs.bandwidth.com/
+  API docs:   https://labs.bandwidth.com/docs/speech-to-text
+  Install:    pip install websockets
+         or:  pip install "video-use[bw-stt]"
+
+Provider selection (highest priority first):
+  1. --provider {elevenlabs,bw_stt} CLI flag
+  2. TRANSCRIBE_PROVIDER environment variable or .env entry
+  3. Auto-detect: whichever API key is present (ElevenLabs wins if both)
+
+Output JSON schema (both providers):
+  {"words": [{"type": "word", "text": "...", "start": 0.0, "end": 0.0,
+              "speaker_id": "speaker_0"}, ...],
+   "text": "...", "audio_duration_seconds": 0.0}
+
+ElevenLabs responses also include "spacing" and "audio_event" word entries.
+pack_transcripts.py handles all three types.
+
+Cached: if the output file already exists, the API call is skipped.
 
 Usage:
     python helpers/transcribe.py <video_path>
     python helpers/transcribe.py <video_path> --edit-dir /custom/edit
     python helpers/transcribe.py <video_path> --language en
     python helpers/transcribe.py <video_path> --num-speakers 2
+    python helpers/transcribe.py <video_path> --provider bw_stt
+    python helpers/transcribe.py <video_path> --audio-track 1
 """
 
 from __future__ import annotations
@@ -31,24 +52,75 @@
 
 
 SCRIBE_URL = "https://api.elevenlabs.io/v1/speech-to-text"
+PROVIDERS = ("elevenlabs", "bw_stt")
 
 
-def load_api_key() -> str:
+def _find_env_value(key_name: str) -> str:
+    """Return the value of *key_name* from .env or environment, or '' if absent."""
     for candidate in [Path(__file__).resolve().parent.parent / ".env", Path(".env")]:
         if candidate.exists():
             for line in candidate.read_text().splitlines():
                 line = line.strip()
                 if not line or line.startswith("#") or "=" not in line:
                     continue
                 k, v = line.split("=", 1)
-                if k.strip() == "ELEVENLABS_API_KEY":
+                if k.strip() == key_name:
                     return v.strip().strip('"').strip("'")
-    v = os.environ.get("ELEVENLABS_API_KEY", "")
+    return os.environ.get(key_name, "")
+
+
+def load_api_key(provider: str = "elevenlabs") -> str:
+    env_var = "ELEVENLABS_API_KEY" if provider == "elevenlabs" else "BW_STT_API_KEY"
+    v = _find_env_value(env_var)
     if not v:
-        sys.exit("ELEVENLABS_API_KEY not found in .env or environment")
+        sys.exit(f"{env_var} not found in .env or environment")
     return v
 
 
+def detect_provider() -> str:
+    """Return the provider whose API key is available; ElevenLabs wins if both are set."""
+    if _find_env_value("ELEVENLABS_API_KEY"):
+        return "elevenlabs"
+    if _find_env_value("BW_STT_API_KEY"):
+        return "bw_stt"
+    sys.exit(
+        "No transcription API key found. "
+        "Set ELEVENLABS_API_KEY (ElevenLabs Scribe) or BW_STT_API_KEY (BW Labs STT) "
+        "in .env or the environment."
+    )
+
+
+def explicit_provider(provider: str | None = None) -> str | None:
+    """Provider explicitly requested via CLI flag or TRANSCRIBE_PROVIDER, or None.
+
+    Unlike resolve_provider(), never falls back to key auto-detection, so it
+    can be answered without any API key configured.
+    """
+    if provider is not None:
+        if provider not in PROVIDERS:
+            sys.exit(f"Unknown provider '{provider}'. Choose from: {', '.join(PROVIDERS)}")
+        return provider
+    env_prov = _find_env_value("TRANSCRIBE_PROVIDER")
+    if env_prov:
+        if env_prov not in PROVIDERS:
+            sys.exit(
+                f"TRANSCRIBE_PROVIDER='{env_prov}' is not valid. "
+                f"Choose from: {', '.join(PROVIDERS)}"
+            )
+        return env_prov
+    return None
+
+
+def resolve_provider(provider: str | None = None) -> str:
+    """Resolve the provider using the three-tier priority order.
+
+    1. *provider* argument (from --provider CLI flag)
+    2. TRANSCRIBE_PROVIDER in .env or environment
+    3. Auto-detection from available API keys
+    """
+    return explicit_provider(provider) or detect_provider()
+
+
 def count_audio_tracks(video_path: Path) -> int:
     """How many audio streams the container holds."""
     out = subprocess.run(
@@ -113,6 +185,99 @@ def call_scribe(
     return resp.json()
 
 
+def _call_bw_stt(audio: Path, api_key: str) -> dict:
+    """Stream *audio* to BW Labs STT via WebSocket; return a normalised transcript dict.
+
+    Uses the public WebSocket API directly — no SDK required, only websockets (PyPI).
+    API docs: https://labs.bandwidth.com/docs/speech-to-text
+    """
+    try:
+        import websockets.sync.client as ws_sync
+    except ImportError:
+        sys.exit(
+            "websockets is required for BW Labs STT.\n"
+            "Install it: pip install websockets\n"
+            '       or: pip install "video-use[bw-stt]"'
+        )
+
+    import threading
+
+    ENDPOINT = (
+        "wss://api.labs.bandwidth.com/audio/v1/listen"
+        "?encoding=linear16&sample_rate=16000&channels=1&mode=instant"
+    )
+    FRAMES_PER_CHUNK = 2560  # 160ms × 16 kHz mono (2560 samples × 2 bytes = 5120 bytes)
+    MIN_FRAME_BYTES = 640    # API minimum frame duration is 20ms = 320 samples × 2 bytes
+
+    all_words: list[dict] = []
+    text_parts: list[str] = []
+    audio_duration = 0.0
+
+    with ws_sync.connect(ENDPOINT, additional_headers={"X-BW-LABS-API-KEY": api_key}) as ws:
+        msg = json.loads(ws.recv())
+        if msg["type"] != "SessionOpened":
+            raise RuntimeError(f"unexpected opening message: {msg['type']}")
+
+        # Send from a separate thread so the main thread drains Segment messages
+        # as they stream in. The server transcribes while audio uploads; reading
+        # only after sending everything would stall long files once the client's
+        # receive buffer fills.
+        send_error: list[BaseException] = []
+
+        def _send_audio() -> None:
+            try:
+                # wave.open strips the WAV header; readframes() returns raw PCM
+                with wave.open(str(audio), "rb") as wf:
+                    while frames := wf.readframes(FRAMES_PER_CHUNK):
+                        if len(frames) < MIN_FRAME_BYTES:
+                            # Pad a short tail with silence to the 20ms API minimum
+                            frames += b"\x00" * (MIN_FRAME_BYTES - len(frames))
+                        ws.send(frames)
+                ws.send(json.dumps({"type": "CloseStream"}))
+            except BaseException as e:  # surfaced after the recv loop ends
+                send_error.append(e)
+
+        sender = threading.Thread(target=_send_audio, daemon=True)
+        sender.start()
+
+        session_closed = False
+        for raw in ws:
+            msg = json.loads(raw)
+            if msg["type"] == "Segment":
+                text_parts.append(msg["text"])
+                for w in msg.get("words", []):
+                    all_words.append({
+                        "type": "word",
+                        "text": w["word"],
+                        "start": w["start"],
+                        "end": w["end"],
+                        "speaker_id": "speaker_0",
+                    })
+            elif msg["type"] == "SessionClosed":
+                audio_duration = msg.get("audio_duration_seconds", 0.0)
+                session_closed = True
+                break
+            elif msg["type"] == "Error":
+                raise RuntimeError(f"BW STT error {msg['code']}: {msg['message']}")
+
+        sender.join(timeout=10)
+        if send_error:
+            raise RuntimeError(f"BW STT send failed: {send_error[0]}") from send_error[0]
+        if not session_closed:
+            # A clean socket close ends the iterator without raising; returning
+            # here would cache a truncated transcript as final output.
+            raise RuntimeError(
+                f"BW STT session ended without SessionClosed — transcript incomplete "
+                f"({len(all_words)} words received). Not caching; retry the transcription."
+            )
+
+    return {
+        "words": all_words,
+        "text": "".join(text_parts),
+        "audio_duration_seconds": audio_duration,
+    }
+
+
 def transcript_path(edit_dir: Path, video: Path, audio_track: int = 0) -> Path:
     """Where a video's transcript lands.
 
@@ -125,6 +290,25 @@ def transcript_path(edit_dir: Path, video: Path, audio_track: int = 0) -> Path:
     return edit_dir / "transcripts" / f"{video.stem}{suffix}.json"
 
 
+def cached_provider(path: Path) -> str:
+    """Which provider produced the transcript at *path*.
+
+    Files written before the provider field existed came from ElevenLabs
+    (the only backend at the time), so a missing field means "elevenlabs".
+    An unreadable or corrupt file — including a provider value that is not a
+    known provider name — returns "" so it never matches a provider and gets
+    re-transcribed.
+    """
+    try:
+        payload = json.loads(path.read_text())
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
+        return ""
+    if not isinstance(payload, dict):
+        return ""
+    prov = payload.get("provider", "elevenlabs")
+    return prov if prov in PROVIDERS else ""
+
+
 def transcribe_one(
     video: Path,
     edit_dir: Path,
@@ -133,19 +317,28 @@ def transcribe_one(
     num_speakers: int | None = None,
     verbose: bool = True,
     audio_track: int = 0,
+    provider: str = "elevenlabs",
 ) -> Path:
     """Transcribe a single video. Returns path to transcript JSON.
 
-    Cached: returns existing path immediately if the transcript already exists.
+    Cached: returns existing path immediately if a transcript from the same
+    provider already exists. A cached file from a different provider is
+    re-transcribed and overwritten.
     """
     transcripts_dir = edit_dir / "transcripts"
     transcripts_dir.mkdir(parents=True, exist_ok=True)
     out_path = transcript_path(edit_dir, video, audio_track)
 
     if out_path.exists():
+        prev = cached_provider(out_path)
+        if prev == provider:
+            if verbose:
+                print(f"cached: {out_path.name}")
+            return out_path
         if verbose:
-            print(f"cached: {out_path.name}")
-        return out_path
+            print(f"  cached {out_path.name} is from "
+                  f"{prev or 'an unreadable file'}; re-transcribing with {provider}",
+                  flush=True)
 
     if verbose:
         print(f"  extracting audio from {video.name}", flush=True)
@@ -173,9 +366,26 @@ def transcribe_one(
             )
 
         size_mb = audio.stat().st_size / (1024 * 1024)
-        if verbose:
-            print(f"  uploading {video.stem}.wav ({size_mb:.1f} MB)", flush=True)
-        payload = call_scribe(audio, api_key, language, num_speakers)
+
+        if provider == "elevenlabs":
+            if verbose:
+                print(f"  uploading {video.stem}.wav ({size_mb:.1f} MB) to ElevenLabs Scribe",
+                      flush=True)
+            payload = call_scribe(audio, api_key, language, num_speakers)
+        else:  # bw_stt
+            if verbose:
+                if language:
+                    print("  note: BW Labs STT auto-detects language; --language is ignored",
+                          flush=True)
+                if num_speakers:
+                    print("  note: BW Labs STT is single-channel; --num-speakers is ignored",
+                          flush=True)
+                print(f"  streaming {video.stem}.wav ({size_mb:.1f} MB) to BW Labs STT",
+                      flush=True)
+            payload = _call_bw_stt(audio, api_key)
+
+    if isinstance(payload, dict):
+        payload["provider"] = provider  # cache identity — see cached_provider()
 
     out_path.write_text(json.dumps(payload, indent=2))
     dt = time.time() - t0
@@ -190,7 +400,9 @@ def transcribe_one(
 
 
 def main() -> None:
-    ap = argparse.ArgumentParser(description="Transcribe a video with ElevenLabs Scribe")
+    ap = argparse.ArgumentParser(
+        description="Transcribe a video with ElevenLabs Scribe or BW Labs STT"
+    )
     ap.add_argument("video", type=Path, help="Path to video file")
     ap.add_argument(
         "--edit-dir",
@@ -202,13 +414,13 @@ def main() -> None:
         "--language",
         type=str,
         default=None,
-        help="Optional ISO language code (e.g., 'en'). Omit to auto-detect.",
+        help="ISO language code (e.g., 'en'). ElevenLabs only; omit to auto-detect.",
     )
     ap.add_argument(
         "--num-speakers",
         type=int,
         default=None,
-        help="Optional number of speakers when known. Improves diarization accuracy.",
+        help="Number of speakers when known. ElevenLabs only; improves diarization.",
     )
     ap.add_argument(
         "--audio-track",
@@ -218,14 +430,25 @@ def main() -> None:
              "and the mic on track 1; without this ffmpeg applies its default audio "
              "stream selection, which picks the track with the most channels.",
     )
+    ap.add_argument(
+        "--provider",
+        choices=PROVIDERS,
+        default=None,
+        help=(
+            "Transcription backend (default: auto-detect from available API keys; "
+            "ElevenLabs wins if both are set). Override globally with "
+            "TRANSCRIBE_PROVIDER env var."
+        ),
+    )
     args = ap.parse_args()
 
     video = args.video.resolve()
     if not video.exists():
         sys.exit(f"video not found: {video}")
 
     edit_dir = (args.edit_dir or (video.parent / "edit")).resolve()
-    api_key = load_api_key()
+    provider = resolve_provider(args.provider)
+    api_key = load_api_key(provider)
 
     transcribe_one(
         video=video,
@@ -234,6 +457,7 @@ def main() -> None:
         language=args.language,
         num_speakers=args.num_speakers,
         audio_track=args.audio_track,
+        provider=provider,
     )