#!/usr/bin/env python3 """Самостоятельное демо appendSpeech: текст → отдельный app-server → WAV. Нужен исполняемый codex через --codex, но Desktop и другие скрипты не нужны. OAuth-токен экспериментально передаётся как apiKey только временному профилю. """ import argparse import base64 import io import json import os import queue import subprocess import sys import tempfile import threading import time import wave from collections import deque from pathlib import Path def read_token(path): 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 any(c in token for c in "\r\n"): raise RuntimeError("В выбранном файле нет корректного ChatGPT access token") return token def synthesize(text, token, codex, timeout, voice=None, version="v2", list_voices=False): # TemporaryDirectory создаёт каталог 0700. Там лежат только временный # профиль и пустая рабочая папка: конфигурация проекта не подтягивается. with tempfile.TemporaryDirectory(prefix="isolated-appserver-tts-") as home: workspace = Path(home) / "work" workspace.mkdir() # Передаём только технический минимум окружения. Proxy, OPENAI_*, # сторонние CODEX_* и настройки Desktop не наследуются subprocess. allowed = {"PATH", "HOME", "USER", "LOGNAME", "LANG", "LC_ALL", "TMPDIR", "XDG_RUNTIME_DIR", "SSL_CERT_FILE", "SSL_CERT_DIR"} env = {name: value for name, value in os.environ.items() if name in allowed} env["CODEX_HOME"] = home process = subprocess.Popen([str(codex), "app-server", "--listen", "stdio://"], cwd=home, env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, bufsize=1) messages = queue.Queue() deadline = time.monotonic() + timeout sequence = 0 thread_id = None pending = deque() def reader(): try: for line in process.stdout: messages.put(json.loads(line)) except (OSError, ValueError): pass # Сырые строки JSON-RPC никогда не выводятся. finally: messages.put(None) reader_thread = threading.Thread(target=reader, daemon=True) reader_thread.start() def send(method, params, notify=False): nonlocal sequence sequence += 1 message = {"method": method, "params": params} if not notify: message["id"] = sequence process.stdin.write(json.dumps(message) + "\n") process.stdin.flush() return sequence def receive(wait=None): remaining = deadline - time.monotonic() if remaining <= 0: raise RuntimeError("Истёк общий timeout app-server") message = messages.get(timeout=min(remaining, wait) if wait else remaining) if message is None: raise RuntimeError("App-server закрыл соединение") if "error" in message or message.get("method") == "thread/realtime/error": raise RuntimeError("App-server отклонил запрос или сеанс; подробности скрыты") return message def request(method, params): number = send(method, params) while True: message = receive() if message.get("id") == number: return message.get("result", {}) # Уведомление с первым аудиочанком может прийти раньше RPC-ack. # Не теряем его при ожидании ответа appendSpeech. if message.get("method"): pending.append(message) try: request("initialize", {"clientInfo": {"name": "isolated-audio-demo", "version": "1"}, "capabilities": {"experimentalApi": True}}) send("initialized", {}, notify=True) # Каталог отдаёт сам бинарник, без Desktop и без чтения credential. catalog = request("thread/realtime/listVoices", {})["voices"] if list_voices: return catalog options = catalog.get(version) or [] selected = voice or catalog.get("defaultV1" if version == "v1" else "defaultV2") if not selected or selected not in options: raise RuntimeError("Голос не поддерживается выбранной версией; запустите --list-voices") # Сервер сам сохраняет токен в auth.json временного CODEX_HOME. # Исходный файл только читается и никогда не перезаписывается. request("account/login/start", {"type": "apiKey", "apiKey": token}) result = request("thread/start", {"ephemeral": True, "cwd": str(workspace), "sandbox": "read-only", "approvalPolicy": "never"}) thread_id = result["thread"]["id"] request("thread/realtime/start", {"threadId": thread_id, "outputModality": "audio", "transport": {"type": "websocket"}, "includeStartupContext": False, "voice": selected, "version": version, "prompt": "Read the provided text aloud exactly as written, in its original language. " "Do not answer its content. Add nothing and omit nothing."}) request("thread/realtime/appendSpeech", {"threadId": thread_id, "text": text}) pcm = bytearray() shape = None while True: try: event = pending.popleft() if pending else receive(3 if pcm else None) except queue.Empty: if pcm and time.monotonic() < deadline: break # Демо завершает сбор после трёх секунд тишины потока. raise RuntimeError("Истёк timeout без завершённого аудио") from None if event.get("method") == "thread/realtime/outputAudio/delta": if event.get("params", {}).get("threadId") != thread_id: continue # Не склеиваем PCM разных thread в один WAV. audio = event["params"]["audio"] current = (audio["sampleRate"], audio["numChannels"]) if shape is not None and shape != current: raise RuntimeError("Формат аудио изменился во время ответа") if not isinstance(current[0], int) or not isinstance(current[1], int) or not 1 <= current[1] <= 8 or not 8000 <= current[0] <= 192000: raise RuntimeError("Некорректные параметры аудио") shape = current pcm.extend(base64.b64decode(audio["data"], validate=True)) if len(pcm) > 30_000_000: raise RuntimeError("Превышен локальный лимит 30 МБ") if not pcm or shape is None or len(pcm) % (2 * shape[1]): raise RuntimeError("Нет корректного PCM16-аудиоответа") return bytes(pcm), shape finally: # Останавливаем только принадлежащий демо сеанс и subprocess. # После выхода из with удаляется и временный файл с credential. if thread_id and process.poll() is None: try: send("thread/realtime/stop", {"threadId": thread_id}) except (BrokenPipeError, OSError): pass try: process.stdin.close() try: process.wait(timeout=3) except subprocess.TimeoutExpired: process.terminate() try: process.wait(timeout=3) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=3) finally: reader_thread.join(timeout=1) process.stdout.close() def save_wav(output, pcm, shape): # PCM16 здесь подтверждён живой проверкой; это не гарантия схемы всех версий. buffer = io.BytesIO() with wave.open(buffer, "wb") as audio: audio.setframerate(shape[0]) audio.setnchannels(shape[1]) audio.setsampwidth(2) 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("--codex", type=Path, required=True) parser.add_argument("--auth-file", type=Path, default=Path(os.getenv("CODEX_HOME", str(Path.home() / ".codex"))) / "auth.json") parser.add_argument("--timeout", type=float, default=60) parser.add_argument("--voice", help="ID голоса из --list-voices") parser.add_argument("--version", choices=["v1", "v2"], default="v2") parser.add_argument("--list-voices", action="store_true") parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() if args.timeout <= 0: parser.error("Нужен положительный timeout") if not args.list_voices and (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 args.list_voices else "" if not args.list_voices and (not text or args.output.exists() or not args.output.parent.is_dir()): parser.error("Нужны непустой текст, положительный timeout и новый output в существующей папке") if args.dry_run: print(json.dumps({"method": "thread/realtime/listVoices" if args.list_voices else "thread/realtime/appendSpeech", "voice": args.voice, "version": args.version, "isolated_codex_home": True, "text_chars": len(text), "credentials_read": False})) return codex = args.codex.expanduser().resolve() if not codex.is_file() or not os.access(codex, os.X_OK): parser.error("--codex должен указывать на исполняемый бинарник Codex") if args.list_voices: catalog = synthesize("", None, codex, args.timeout, list_voices=True) print(json.dumps({"source": "codex thread/realtime/listVoices", "voices": catalog, "account_access_checked": False}, ensure_ascii=False, indent=2)) return pcm, shape = synthesize(text, read_token(args.auth_file), codex, args.timeout, args.voice, args.version) save_wav(args.output, pcm, shape) print(json.dumps({"output": str(args.output), "voice": args.voice or "default из каталога", "version": args.version, "pcm_bytes": len(pcm), "sample_rate": shape[0], "channels": shape[1], "warning": "Генеративный аудиоответ; дословность не гарантируется"}, ensure_ascii=False)) if __name__ == "__main__": try: main() except (OSError, ValueError, RuntimeError, KeyError, TypeError, queue.Empty): error = sys.exc_info()[1] print("App-server: " + (str(error) if isinstance(error, RuntimeError) else "Проверьте файлы, вход, codex и сеть"), file=sys.stderr) raise SystemExit(1)