#!/usr/bin/env python3 """Самостоятельное демо: текстовый файл → удалённый Realtime → WAV. Нужен websocket-client. Других файлов репозитория и запущенного Desktop не нужно. ChatGPT OAuth-вход используется экспериментально, его доступность не гарантируется. """ import argparse import base64 import io import json import os import sys import tempfile import time import wave from pathlib import Path from urllib.parse import urlencode VOICES = ("alloy", "ash", "ballad", "coral", "echo", "sage", "shimmer", "verse", "marin", "cedar") VOICE_SOURCE = "https://developers.openai.com/api/docs/guides/realtime-conversations" def read_token(path): # Читаем актуальный access token прямо из auth.json, не копируем его в код. auth = json.loads(path.read_text(encoding="utf-8")) token = (auth.get("tokens") or {}).get("access_token") if auth.get("auth_mode") != "chatgpt" or not isinstance(token, str) or not token or token == "0000" or any(c in token for c in "\r\n"): raise RuntimeError("В выбранном auth.json нет корректного ChatGPT access token") return token def synthesize(text, token, model, voice, timeout): import websocket # Фиксированный сервер OpenAI. Credential находится только в заголовке, # не в URL, argv, журнале или отдельном файле с ключом. url = "wss://api.openai.com/v1/realtime?" + urlencode({"model": model}) deadline = time.monotonic() + timeout connection = None try: connection = websocket.create_connection(url, header=["Authorization: Bearer " + token], timeout=timeout, http_no_proxy=["*"]) def receive(): # Общий срок ограничивает и поток посторонних событий, а не только # время ожидания одного чанка. Не печатаем сырые события сервера. remaining = deadline - time.monotonic() if remaining <= 0: raise RuntimeError("Истёк общий timeout Realtime") connection.settimeout(remaining) event = json.loads(connection.recv()) if event.get("type") == "error": raise RuntimeError("Realtime вернул error; подробности ответа скрыты") return event def wait_for(kind): while receive().get("type") != kind: pass wait_for("session.created") # Сначала настраиваем голос, PCM16 и инструкцию чтения. Это генеративный # ответ: инструкция просит дословность, но не обеспечивает её всегда. connection.send(json.dumps({"type": "session.update", "session": { "type": "realtime", "model": model, "output_modalities": ["audio"], "audio": {"input": {"turn_detection": None}, "output": {"format": {"type": "audio/pcm", "rate": 24000}, "voice": voice}}, "instructions": "Read the user's text aloud exactly as written. Add nothing."}})) wait_for("session.updated") connection.send(json.dumps({"type": "conversation.item.create", "item": { "type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}})) connection.send(json.dumps({"type": "response.create"})) pcm = bytearray() response_id = None while True: event = receive() if event.get("type") == "response.created": created_id = event.get("response", {}).get("id") if not isinstance(created_id, str) or not created_id: raise RuntimeError("Realtime не вернул ID созданного ответа") if response_id is not None and response_id != created_id: raise RuntimeError("Realtime создал второй ответ вместо одного ожидаемого") response_id = created_id if event.get("type") == "response.output_audio.delta": # Не склеиваем звук разных response. Для одного нового сеанса # created должен предшествовать delta; нарушение — отказ, не WAV. if response_id is None: raise RuntimeError("Аудио пришло до response.created") if event.get("response_id") != response_id: continue pcm.extend(base64.b64decode(event["delta"], validate=True)) if len(pcm) > 24000 * 2 * 120: raise RuntimeError("Аудио превышает локальный лимит 120 секунд") if event.get("type") == "response.done": if response_id is None or event.get("response", {}).get("id") != response_id: raise RuntimeError("Итоговый ответ не соответствует response.created") if event.get("response", {}).get("status") != "completed" or not pcm or len(pcm) % 2: raise RuntimeError("Нет завершённого корректного PCM16-аудиоответа") return bytes(pcm) 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 save_wav(output, pcm): # Формируем WAV целиком, затем публикуем готовый файл. Права 0600, # os.link не перезаписывает существующий output даже при гонке процессов. buffer = io.BytesIO() with wave.open(buffer, "wb") as audio: audio.setnchannels(1) audio.setsampwidth(2) audio.setframerate(24000) audio.writeframes(pcm) temporary = tempfile.NamedTemporaryFile(dir=output.parent, prefix=".remote-tts-", delete=False) try: with temporary: temporary.write(buffer.getvalue()) os.link(temporary.name, output) finally: Path(temporary.name).unlink(missing_ok=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("text_file", type=Path, nargs="?") parser.add_argument("output", type=Path, nargs="?") parser.add_argument("--auth-file", type=Path, default=Path(os.getenv("CODEX_HOME", str(Path.home() / ".codex"))) / "auth.json") parser.add_argument("--model", default="gpt-realtime") parser.add_argument("--voice", default="marin", choices=VOICES) parser.add_argument("--list-voices", action="store_true") parser.add_argument("--timeout", type=float, default=60) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() if args.list_voices: # Документированный набор, не выдуманный REST endpoint с каталогом. # Этот режим не читает credential и не утверждает доступность аккаунту. print(json.dumps({"source": VOICE_SOURCE, "voices": VOICES, "default": "marin", "account_access_checked": False}, ensure_ascii=False, indent=2)) return if args.text_file is None or args.output is None: parser.error("Для синтеза нужны text_file и output") text = args.text_file.read_text(encoding="utf-8").strip() if not text or args.timeout <= 0 or args.output.exists() or not args.output.parent.is_dir(): parser.error("Нужны непустой текст, положительный timeout и новый output в существующей папке") if args.dry_run: print(json.dumps({"endpoint": "wss://api.openai.com/v1/realtime", "model": args.model, "voice": args.voice, "text_chars": len(text), "credentials_read": False})) return pcm = synthesize(text, read_token(args.auth_file), args.model, args.voice, args.timeout) save_wav(args.output, pcm) print(json.dumps({"output": str(args.output), "voice": args.voice, "model": args.model, "pcm_bytes": len(pcm), "duration_seconds": len(pcm) / 48000})) if __name__ == "__main__": try: main() except (OSError, ValueError, RuntimeError, ImportError, KeyError): error = sys.exc_info()[1] print("Realtime: " + (str(error) if isinstance(error, RuntimeError) else "Проверьте файлы, вход, сеть и websocket-client"), file=sys.stderr) raise SystemExit(1)