Quick local run transcription service CLI mini app for the Mac.

transcript.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "pyannote.audio>=4.0",
#     "mlx-whisper",
# ]
# ///
"""
Transcribe audio with speaker diarization (macOS Apple Silicon only).

Usage:
    ./transcribe.py <audio_file> [options]

Requirements:
    - uv (https://docs.astral.sh/uv/)
    - ffmpeg: brew install ffmpeg
    - HuggingFace token with gated repo access (HF_TOKEN env var or --hf-token)

    Before first run, accept the model terms at:
    https://huggingface.co/pyannote/speaker-diarization-community-1
"""

import argparse
import os
import shutil
import subprocess
import sys
from enum import Enum
from pathlib import Path

DIARIZATION_MODEL = "pyannote/speaker-diarization-community-1"


class Model(str, Enum):
    tiny = "mlx-community/whisper-tiny"
    base = "mlx-community/whisper-base"
    small = "mlx-community/whisper-small"
    medium = "mlx-community/whisper-medium"
    large = "mlx-community/whisper-large-v3-mlx"
    turbo = "mlx-community/whisper-large-v3-turbo"

    def __str__(self):
        return self.name


def check_prerequisites(hf_token):
    """Check that all prerequisites are met before downloading heavy models."""
    ok = True

    if not shutil.which("ffmpeg"):
        print("ERROR: ffmpeg is not installed.")
        print("       pyannote.audio 4.0 uses torchcodec which requires ffmpeg.")
        print()
        print("  Install with:  brew install ffmpeg")
        print()
        ok = False

    if not hf_token:
        print("ERROR: No HuggingFace token provided.")
        print("       The diarization model requires authentication.")
        print()
        print("  1. Create a free account at https://huggingface.co")
        print(f"  2. Accept model terms at https://huggingface.co/{DIARIZATION_MODEL}")
        print("  3. Create a token at https://huggingface.co/settings/tokens")
        print("     - Enable 'Access public gated repos' in token permissions")
        print("  4. Run with: HF_TOKEN=hf_... ./transcribe.py <audio_file>")
        print()
        ok = False

    return ok


def get_audio_duration(audio_file):
    """Get audio duration in seconds using ffprobe."""
    try:
        result = subprocess.run(
            ["ffprobe", "-v", "quiet", "-show_entries", "format=duration",
             "-of", "default=noprint_wrappers=1:nokey=1", audio_file],
            capture_output=True, text=True,
        )
        return float(result.stdout.strip())
    except Exception:
        return None


def main():
    parser = argparse.ArgumentParser(
        description="Transcribe audio with speaker diarization (macOS Apple Silicon)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""examples:
  ./transcribe.py meeting.m4a
  ./transcribe.py meeting.m4a --model large --max-speakers 6
  HF_TOKEN=hf_... ./transcribe.py meeting.m4a""",
    )
    parser.add_argument("audio_file", help="Path to audio file")
    parser.add_argument("--model", type=lambda s: Model[s], default="turbo",
                        choices=list(Model),
                        help="Whisper model size (default: turbo)")
    parser.add_argument("--language", default="en", help="Language code (default: en)")
    parser.add_argument("--min-speakers", type=int, default=1)
    parser.add_argument("--max-speakers", type=int, default=4)
    parser.add_argument("--hf-token", default=os.environ.get("HF_TOKEN"),
                        help="HuggingFace token (or set HF_TOKEN env var)")
    parser.add_argument("--output", help="Output file path (default: <audio>_transcript.txt)")
    args = parser.parse_args()

    if not os.path.isfile(args.audio_file):
        print(f"ERROR: Audio file not found: {args.audio_file}")
        sys.exit(1)

    if not check_prerequisites(args.hf_token):
        sys.exit(1)

    duration = get_audio_duration(args.audio_file)
    if duration:
        print(f">> Audio: {args.audio_file} ({duration / 60:.1f} min)")
    else:
        print(f">> Audio: {args.audio_file}")

    # Import heavy dependencies only after prerequisites pass
    import torch
    import mlx_whisper
    from pyannote.audio import Pipeline
    from pyannote.audio.pipelines.utils.hook import ProgressHook
    from pyannote.core import Segment

    print(">> Step 1: Loading diarization pipeline...")
    diarization_pipeline = Pipeline.from_pretrained(
        DIARIZATION_MODEL,
        token=args.hf_token,
    )

    device = torch.device("mps") if torch.backends.mps.is_available() else torch.device("cpu")
    print(f">> Using device: {device}")
    diarization_pipeline.to(device)

    print(">> Running diarization...")
    with ProgressHook() as hook:
        diarization = diarization_pipeline(
            args.audio_file,
            min_speakers=args.min_speakers,
            max_speakers=args.max_speakers,
            hook=hook,
        )

    # DiarizeOutput has .speaker_diarization and .exclusive_speaker_diarization
    # Build a simple lookup list from the exclusive (non-overlapping) segments
    diarization_segments = [
        (turn.start, turn.end, speaker)
        for turn, speaker in diarization.exclusive_speaker_diarization
    ]
    speakers = set(s for _, _, s in diarization_segments)
    print(f">> Found {len(speakers)} speaker(s)")

    # Free diarization model before transcription
    del diarization_pipeline, diarization
    if device.type == "mps":
        torch.mps.empty_cache()

    print(f">> Step 2: Transcribing with {args.model.name} (MLX)...")
    transcription = mlx_whisper.transcribe(
        args.audio_file,
        path_or_hf_repo=args.model.value,
        language=args.language,
        verbose=True,
    )

    print("\n>> Merging results...\n")
    print("=" * 80)

    def find_speaker(seg_start, seg_end):
        best_speaker, best_overlap = "UNKNOWN", 0.0
        for start, end, speaker in diarization_segments:
            overlap = min(end, seg_end) - max(start, seg_start)
            if overlap > best_overlap:
                best_overlap = overlap
                best_speaker = speaker
        return best_speaker

    output_lines = []
    for segment in transcription["segments"]:
        speaker = find_speaker(segment["start"], segment["end"])
        line = f"[{segment['start']:.1f}s - {segment['end']:.1f}s] {speaker}: {segment['text'].strip()}"
        print(line)
        output_lines.append(line)

    audio_path = Path(args.audio_file)
    output_file = args.output or str(audio_path.with_name(audio_path.stem + "_transcript.txt"))
    with open(output_file, "w", encoding="utf-8") as f:
        f.write("\n".join(output_lines) + "\n")

    print("=" * 80)
    print(f"\n>> Transcript saved to {output_file}")


if __name__ == "__main__":
    main()