#!/usr/bin/env python3 """Учебное демо: PCM16 WAV → удалённая Realtime transcription → текст. Один переносимый файл. Нужен websocket-client, но не Desktop, CLI или app-server. Это WebSocket transcription, НЕ копия голосового WebRTC v3 в Codex CLI. ChatGPT OAuth здесь экспериментален; доступ конкретного аккаунта не обещан. """ import argparse import base64 import json import math import os import sys import time import wave from pathlib import Path ENDPOINT = "wss://api.openai.com/v1/realtime?intent=transcription" RATE = 24000 MAX_SECONDS = 60 CHUNK_BYTES = 48000 # Одна секунда mono PCM16: 24 000 отсчётов по 2 байта. def read_audio(path): # WAV — контейнер; по сети отправляем только исходные PCM-сэмплы, без RIFF. with wave.open(str(path), "rb") as audio: if (audio.getnchannels(), audio.getsampwidth(), audio.getframerate(), audio.getcomptype()) != (1, 2, RATE, "NONE"): raise RuntimeError("Нужен WAV: PCM16, mono, 24000 Гц; конвертируйте файл через ffmpeg") frames = audio.getnframes() # Нижняя граница даёт небольшой запас относительно минимального буфера commit. if not RATE // 5 <= frames <= RATE * MAX_SECONDS: raise RuntimeError("Демо принимает записи длиной от 0,2 до 60 секунд") pcm = audio.readframes(frames) if len(pcm) != frames * 2: raise RuntimeError("WAV обрезан: аудиобайтов меньше, чем заявлено в заголовке") return pcm def read_credential(auth_file, api_key_file): # Секрет читается прямо из файла, а не из argv и не из строк исходного кода. if api_key_file: token = api_key_file.expanduser().read_text(encoding="utf-8").strip() else: auth = json.loads(auth_file.expanduser().read_text(encoding="utf-8")) if auth.get("auth_mode") != "chatgpt": raise RuntimeError("--auth-file должен содержать именно ChatGPT-вход") token = (auth.get("tokens") or {}).get("access_token") if not isinstance(token, str) or not token or token == "0000" or any(c in token for c in "\r\n"): raise RuntimeError("Файл не содержит корректного credential; gateway-маркер 0000 не подходит") return token def session_update(model): # Только распознавание входа: нет response.create, выходного голоса или ответа модели. return {"type": "session.update", "session": {"type": "transcription", "audio": { "input": {"format": {"type": "audio/pcm", "rate": RATE}, "transcription": {"model": model}, "turn_detection": None}}}} def transcribe(pcm, token, model, timeout): import websocket # Единственная внешняя Python-зависимость. deadline = time.monotonic() + timeout connection = None try: # WSS = WebSocket поверх TLS. Токен — только в Authorization handshake. # HTTP proxy окружения отключён; конфигурацию acodex не используем. connection = websocket.create_connection(ENDPOINT, header=["Authorization: Bearer " + token], timeout=timeout, http_no_proxy=["*"]) def remaining(): seconds = deadline - time.monotonic() if seconds <= 0: raise RuntimeError("Истёк общий timeout транскрипции") connection.settimeout(seconds) def receive(): remaining() event = json.loads(connection.recv()) # Сырые события/тела ошибок не печатаем: там могут оказаться личные данные. if event.get("type") in ("error", "conversation.item.input_audio_transcription.failed"): raise RuntimeError("Сервис отклонил сеанс или распознавание; тело ошибки скрыто") return event def send(event): remaining() connection.send(json.dumps(event)) def wait_for(kind): while receive().get("type") != kind: pass wait_for("session.created") send(session_update(model)) wait_for("session.updated") for offset in range(0, len(pcm), CHUNK_BYTES): # Чанки отправляются подряд, без искусственного ожидания реального времени. # Base64 — кодирование для JSON, не шифрование; шифрование даёт TLS. send({"type": "input_audio_buffer.append", "audio": base64.b64encode(pcm[offset:offset + CHUNK_BYTES]).decode("ascii")}) send({"type": "input_audio_buffer.commit"}) committed = None completed = {} while True: event = receive() kind = event.get("type") item = event.get("item_id") # Проверяем ID, чтобы не принять чужой/старый текст за наш committed turn. # Completed может прийти раньше ack: сохраняем его до появления нужного ID. if kind == "input_audio_buffer.committed" and isinstance(item, str) and item: committed = item elif kind == "conversation.item.input_audio_transcription.completed" and isinstance(item, str) and item: text = event.get("transcript") if not isinstance(text, str): raise RuntimeError("Сервис вернул некорректную транскрипцию") if len(text) > 100000: raise RuntimeError("Превышен локальный лимит текста") completed[item] = text if len(completed) > 8: raise RuntimeError("Получено слишком много посторонних результатов") if committed in completed: text = completed[committed].strip() if not text: raise RuntimeError("Сервис не распознал речь: итоговая строка пуста") return text except websocket.WebSocketBadStatusException as error: raise RuntimeError(f"WebSocket handshake: HTTP {error.status_code}; тело ответа скрыто") from None except websocket.WebSocketException: raise RuntimeError("Сбой соединения Realtime; подробности скрыты") from None finally: if connection is not None: connection.close() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("input_wav", type=Path) credentials = parser.add_mutually_exclusive_group() credentials.add_argument("--auth-file", type=Path, default=Path(os.getenv("CODEX_HOME", str(Path.home() / ".codex"))) / "auth.json") credentials.add_argument("--api-key-file", type=Path, help="Файл своего Platform key вместо экспериментального OAuth") parser.add_argument("--model", choices=["gpt-transcribe", "gpt-live-transcribe"], default="gpt-transcribe") parser.add_argument("--timeout", type=float, default=90) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() if not math.isfinite(args.timeout) or args.timeout <= 0: parser.error("Нужен конечный положительный --timeout") pcm = read_audio(args.input_wav.expanduser()) if args.dry_run: # Этот режим не импортирует websocket, не читает credential и не делает сеть. print(json.dumps({"endpoint": ENDPOINT, "model": args.model, "pcm_bytes": len(pcm), "duration_seconds": len(pcm) / 48000, "chunks": math.ceil(len(pcm) / CHUNK_BYTES), "credentials_read": False, "network": False, "app_server": False})) return token = read_credential(args.auth_file, args.api_key_file) # stdout содержит только полезный текст. Перенаправление файла контролирует shell. print(transcribe(pcm, token, args.model, args.timeout)) if __name__ == "__main__": try: main() except (OSError, ValueError, RuntimeError, ImportError, KeyError, TypeError, wave.Error): error = sys.exc_info()[1] # Не выводим traceback, credential, заголовки, тела HTTP или серверный JSON. print("Realtime STT: " + (str(error) if isinstance(error, RuntimeError) else "Проверьте WAV, файл входа, сеть и websocket-client"), file=sys.stderr) raise SystemExit(1)