stt_openai.py

#!/usr/bin/env -S uv run
# /// script
# dependencies = [
#   "openai>=1.0",
# ]
# requires-python = ">=3.11"
# ///
"""
OpenAI Speech-to-Text (STT) Transcription Tool

Transcribe audio files using the OpenAI Whisper API with support for
multiple languages, timestamps, and various output formats.

Usage:
    ./stt_openai.py audio.mp3
    ./stt_openai.py -l en audio.mp3  # English transcription
    ./stt_openai.py --timestamps audio.mp3  # With word timestamps

Environment:
    OPENAI_API_KEY - Your OpenAI API key (required)
"""

import os
import argparse
import sys
import json
from openai import OpenAI

# ----------------------------------------------------------------------
# Simple verbosity-aware logger.
#   -v          -> INFO messages
#   -vvvvv      -> DEBUG messages
# All logs are written to STDERR and are suppressed by --quiet
# ----------------------------------------------------------------------
def _should_log(args, level_threshold):
    return getattr(args, "verbose", 0) >= level_threshold and not getattr(args, "quiet", False)

def log_error(args, message):
    print(f"ERROR: {message}", file=sys.stderr)

def log_warning(args, message):
    if _should_log(args, 0):
        print(f"WARNING: {message}", file=sys.stderr)

def log_info(args, message):
    if _should_log(args, 1):
        print(f"INFO: {message}", file=sys.stderr)

def log_debug(args, message):
    if _should_log(args, 5):
        print(f"DEBUG: {message}", file=sys.stderr)


# ----------------------------------------------------------------------
# Filename helpers for long filenames
# ----------------------------------------------------------------------

KNOWN_EXTENSIONS = {
    "mp3", "mp4", "wav", "flac", "m4a", "ogg", "webm", "mpeg", "mpga",
    "json", "txt", "srt", "vtt", "md",
    "openai"
}


def _split_known_suffix(filename, known_extensions=KNOWN_EXTENSIONS):
    """
    Return (basename, chained_suffix_with_dot)
    Example:
        >>> _split_known_suffix("foo.bar.mp4.openai.json")
        ('foo.bar', '.mp4.openai.json')
    """
    parts = filename.split(".")
    if len(parts) == 1:
        return filename, ""

    suffix_parts = []
    for part in reversed(parts[1:]):
        if part.lower() in known_extensions:
            suffix_parts.insert(0, part)
        else:
            break

    if suffix_parts:
        suffix = "." + ".".join(suffix_parts)
        base_parts_count = len(parts) - len(suffix_parts)
        basename = ".".join(parts[:base_parts_count])
        return basename, suffix
    else:
        basename, ext = os.path.splitext(filename)
        return basename, ext


def make_safe_filename(path, max_component_length=255):
    """
    Ensure the final component of *path* is <= max_component_length bytes.
    If it is longer, truncate the basename until it fits.
    """
    dir_name, file_name = os.path.split(path)
    if len(file_name.encode()) <= max_component_length:
        return path

    base, suffix = _split_known_suffix(file_name)
    allowed = max(1, max_component_length - len(suffix.encode()))
    truncated_base = base.encode()[:allowed].decode(errors="ignore")
    safe_name = truncated_base + suffix
    return os.path.join(dir_name, safe_name)


# ----------------------------------------------------------------------
# META message helper
# ----------------------------------------------------------------------

def get_meta_message(args):
    """
    Get META warning message for STT transcripts.

    Returns empty string if disabled via flag or environment variable.
    Returns custom message if STT_META_MESSAGE env var is set.
    Returns default message otherwise.
    """
    if getattr(args, 'no_meta_message', False) or getattr(args, 'disable_meta_message', False):
        return ""

    if os.environ.get('STT_META_MESSAGE_DISABLE', '').lower() in ('1', 'true', 'yes'):
        return ""

    custom_message = os.environ.get('STT_META_MESSAGE', '').strip()
    if custom_message:
        return f"---\nmeta: {custom_message}\n---\n"

    default_message = (
        "THIS IS AN AUTOMATED SPEECH-TO-TEXT (STT) TRANSCRIPT AND MAY CONTAIN TRANSCRIPTION ERRORS. "
        "This transcript was generated by automated speech recognition technology and should be treated "
        "as a rough transcription for reference purposes. Common types of errors include: incorrect word "
        "recognition (especially homophones, proper nouns, technical terminology, or words in noisy audio "
        "conditions), missing or incorrect punctuation, speaker misidentification in multi-speaker scenarios, "
        "and timing inaccuracies. For best comprehension and to mentally correct potential errors, please consider: "
        "the broader conversational context, relevant domain knowledge, technical background of the subject matter, "
        "and any supplementary information about the speakers or topic. This transcript is intended to convey "
        "the general content and flow of the conversation rather than serving as a verbatim, word-perfect record. "
        "When critical accuracy is required, please verify important details against the original audio source."
    )

    return f"---\nmeta: {default_message}\n---\n"


# ----------------------------------------------------------------------
# API functions
# ----------------------------------------------------------------------

def transcribe_audio(client, audio_input, args):
    """
    Transcribe audio file using OpenAI Whisper API.

    Args:
        client: OpenAI client
        audio_input: Path to audio file
        args: Parsed arguments

    Returns:
        Transcription result (dict for verbose_json, string for others)
    """
    log_info(args, f"Transcribing {audio_input}...")

    # Determine response format
    response_format = "verbose_json" if args.timestamps else "json"
    if args.response_format:
        response_format = args.response_format

    log_debug(args, f"Using response format: {response_format}")

    with open(audio_input, "rb") as audio_file:
        kwargs = {
            "model": args.model,
            "file": audio_file,
            "response_format": response_format,
        }

        # Add optional parameters
        if args.language and args.language != 'auto':
            kwargs["language"] = args.language
            log_debug(args, f"Language: {args.language}")

        if args.prompt:
            kwargs["prompt"] = args.prompt
            log_debug(args, f"Prompt: {args.prompt}")

        if args.temperature is not None:
            kwargs["temperature"] = args.temperature
            log_debug(args, f"Temperature: {args.temperature}")

        log_debug(args, f"API request kwargs: {kwargs}")

        transcript = client.audio.transcriptions.create(**kwargs)

    log_info(args, "Transcription complete.")

    # Handle different response formats
    if response_format in ("verbose_json", "json"):
        if hasattr(transcript, 'model_dump'):
            return transcript.model_dump()
        return transcript
    else:
        return transcript


def translate_audio(client, audio_input, args):
    """
    Translate audio file to English using OpenAI Whisper API.

    Args:
        client: OpenAI client
        audio_input: Path to audio file
        args: Parsed arguments

    Returns:
        Translation result
    """
    log_info(args, f"Translating {audio_input} to English...")

    response_format = "verbose_json" if args.timestamps else "json"
    if args.response_format:
        response_format = args.response_format

    with open(audio_input, "rb") as audio_file:
        kwargs = {
            "model": args.model,
            "file": audio_file,
            "response_format": response_format,
        }

        if args.prompt:
            kwargs["prompt"] = args.prompt

        if args.temperature is not None:
            kwargs["temperature"] = args.temperature

        translation = client.audio.translations.create(**kwargs)

    log_info(args, "Translation complete.")

    if response_format in ("verbose_json", "json"):
        if hasattr(translation, 'model_dump'):
            return translation.model_dump()
        return translation
    else:
        return translation


# ----------------------------------------------------------------------
# Output helpers
# ----------------------------------------------------------------------

def format_transcript_txt(transcript_data, args):
    """
    Format transcript data as human-readable text.

    Args:
        transcript_data: Transcript from API (dict or string)
        args: Parsed arguments

    Returns:
        Formatted text string
    """
    if isinstance(transcript_data, str):
        return transcript_data.strip() + '\n'

    if isinstance(transcript_data, dict):
        text = transcript_data.get('text', '')
        return text.strip() + '\n' if text else ""

    return str(transcript_data).strip() + '\n'


def write_str(args, output, string, mode='w'):
    if output != '-':
        with open(output, mode) as f:
            f.write(string)
    if output == '-' or not args.quiet:
        print(string, end='')


def write_transcript_to_file(args, output, transcript_data, audio_input):
    """Write transcript to output files."""
    import copy
    args_force_quiet = copy.deepcopy(args)
    args_force_quiet.quiet = True
    json_path = make_safe_filename(audio_input + '.openai.json')

    # Prepare JSON data
    if isinstance(transcript_data, dict):
        json_data = transcript_data
    else:
        json_data = {"text": str(transcript_data)}

    # Add META note to JSON if enabled
    meta_message_text = get_meta_message(args).replace("---\nmeta: ", "").replace("\n---\n", "").strip()
    if meta_message_text:
        json_data = {
            "_meta_note": meta_message_text,
            **json_data
        }

    write_str(args_force_quiet, json_path, json.dumps(json_data, indent=2, ensure_ascii=False))

    if not args.quiet:
        log_info(args, f"Server response written to {json_path}")

    # Prepend META message to TXT output if enabled
    meta_message = get_meta_message(args)
    if meta_message:
        write_str(args, output, meta_message)

    # Format and write text transcript
    txt_content = format_transcript_txt(transcript_data, args)
    write_str(args, output, txt_content, 'a' if meta_message else 'w')

    if output != '-' and not args.quiet:
        log_info(args, f"Output written to {output}")


# ----------------------------------------------------------------------
# Main
# ----------------------------------------------------------------------

def stt_openai_main(args, client):
    audio_input = args.audio_input

    try:
        log_info(args, "Processing audio input...")

        # Determine the output file
        if args.output == '-':
            potential_output = audio_input + '.txt'
            output = potential_output if os.path.exists(potential_output) else '-'
        else:
            output = args.output if args.output is not None else audio_input + '.txt'
            output = make_safe_filename(output)
        log_info(args, f"output filename: {output}")

        # Check if output file exists before making the transcript
        if os.path.exists(output):
            if not args.quiet and args.verbose:
                sys.stderr.write(f'SKIPPING: transcription of {audio_input} as {output} already exists\n')
            if (not args.quiet) or args.output == '-':
                with open(output, 'r') as f:
                    print(f.read())
            sys.exit(0)

        # Transcribe or translate
        if args.translate:
            transcript_data = translate_audio(client, audio_input, args)
        else:
            transcript_data = transcribe_audio(client, audio_input, args)

        # Write output
        log_info(args, "Writing output files...")
        write_transcript_to_file(args, output, transcript_data, audio_input)
        log_info(args, "Done.")

    except Exception as e:
        log_error(args, f'Error: {e}')
        sys.exit(1)


def make_arg_parser():
    parser = argparse.ArgumentParser(
        description='Transcribe audio file using OpenAI Whisper API.',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s audio.mp3
  %(prog)s -l en audio.mp3                # English transcription
  %(prog)s --timestamps audio.mp3         # With word timestamps
  %(prog)s --translate audio.mp3          # Translate to English
  %(prog)s --model whisper-1 audio.mp3    # Specify model

Environment:
  OPENAI_API_KEY            Your OpenAI API key (required)
  STT_META_MESSAGE_DISABLE=1  Disable META warning message
  STT_META_MESSAGE="..."      Custom META message
"""
    )
    parser.add_argument('audio_input', type=str,
                        help='Path to audio file to transcribe (mp3, mp4, mpeg, mpga, m4a, wav, webm)')
    parser.add_argument('-o', '--output', type=str, default=None,
                        help='Output file path. Default: {audio}.txt. Use "-" for stdout only.')
    parser.add_argument('-q', '--quiet', action='store_true',
                        help='Suppress all status messages')
    parser.add_argument('-l', '--language', type=str, default=None,
                        help='Language code (ISO 639-1). Default: auto-detect. Examples: en, de, fr, es, ja, zh')
    parser.add_argument('--model', type=str, default='whisper-1',
                        help='Whisper model to use. Default: whisper-1')
    parser.add_argument('--timestamps', action='store_true',
                        help='Include word-level timestamps in output (uses verbose_json format)')
    parser.add_argument('--response-format', choices=['json', 'text', 'srt', 'verbose_json', 'vtt'],
                        default=None,
                        help='Response format. Default: json (or verbose_json with --timestamps)')
    parser.add_argument('--translate', action='store_true',
                        help='Translate audio to English instead of transcribing')
    parser.add_argument('--prompt', type=str, default=None,
                        help='Optional prompt to guide transcription style or provide context')
    parser.add_argument('--temperature', type=float, default=None,
                        help='Sampling temperature (0-1). Lower = more deterministic.')
    parser.add_argument('-v', '--verbose', action='count', default=0,
                        help='Increase verbosity. -v for INFO, -vvvvv for DEBUG.')
    parser.add_argument('--no-meta-message', '--disable-meta-message', action='store_true',
                        dest='no_meta_message',
                        help='Disable META warning message about transcription errors')
    return parser


if __name__ == "__main__":
    try:
        api_key = os.environ["OPENAI_API_KEY"]
    except KeyError:
        print("Error: OPENAI_API_KEY environment variable not set.")
        print("Get your API key at: https://platform.openai.com/api-keys")
        sys.exit(1)

    parser = make_arg_parser()
    args = parser.parse_args()

    client = OpenAI(api_key=api_key)
    stt_openai_main(args, client)