sundynix-agentix/sundynix-gateway/internal/voice/tts.go

package voice

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"sync"
	"time"

	"github.com/gorilla/websocket"
)

// ttsDebug 打开时（环境变量 VOICE_DEBUG=1）打印每个 TTS 帧，便于联调双向流事件/音频。
var ttsDebug = os.Getenv("VOICE_DEBUG") != ""

func ttsDebugf(format string, a ...any) {
	if ttsDebug {
		log.Printf("[tts-debug] "+format, a...)
	}
}

// preview 取 payload 前 n 字节的可读预览（音频用长度代替）。
func preview(b []byte, isAudio bool) string {
	if isAudio {
		return fmt.Sprintf("<audio %d bytes>", len(b))
	}
	if len(b) > 160 {
		return string(b[:160]) + "…"
	}
	return string(b)
}

// 火山「双向流式 TTS V3」客户端（seed-tts-2.0）。边推文字边收音频，配 LLM token 流做连续朗读。
// 帧协议见 tts_frame.go；鉴权走**新版控制台 API Key**（单个 X-Api-Key，见 frame.go）。
//
// 一轮朗读的生命周期：StartTTS(连接+StartSession) → Speak(逐句推文字/事件200) → Finish(FinishSession)
// → 客户端 range Audio() 收音频直到 channel 关闭 → Close()。一个 TTSSession = 一次 Agent 回答。

const (
	ttsEndpoint   = "wss://openspeech.bytedance.com/api/v3/tts/bidirection"
	TTSSampleRate = 24000 // 双向 TTS 回 PCM 24kHz 单声道（客户端按此播放）
)

// TTSSession 是一路双向 TTS 会话。Speak 推文字、Audio 出 PCM、Finish 收尾、Close 关闭。
type TTSSession struct {
	conn      *websocket.Conn
	sessionID string
	reqBase   map[string]any // req_params 基底（speaker+audio_params），Speak 时加 text 复用
	audio     chan []byte
	closed    chan struct{}
	closeOnce sync.Once
	mu        sync.Mutex
	writeMu   sync.Mutex // 串行化对火山连接的写（Speak 与 Finish 可能并发）
	err       error
}

// ttsReqBase 组双向流式 TTS 的 req_params 基底（音色 + 音频参数 PCM 24k）。
// 对齐官方 python demo：StartSession 与 TaskRequest 都带这个 req_params，TaskRequest 再往里加 text。
func ttsReqBase(cfg Config) map[string]any {
	return map[string]any{
		"speaker": cfg.TTSVoiceType,
		"audio_params": map[string]any{
			"format":      "pcm",
			"sample_rate": TTSSampleRate,
		},
	}
}

// buildTTSStartSession 组 StartSession payload：{req_params:{speaker,audio_params}}（无 namespace/user）。
func buildTTSStartSession(base map[string]any) []byte {
	b, _ := json.Marshal(map[string]any{"req_params": base})
	return b
}

// StartTTS 连火山双向 TTS，握手（StartConnection→StartSession）成功后返回会话。
func StartTTS(ctx context.Context, cfg Config) (*TTSSession, error) {
	if !cfg.TTSEnabled() {
		return nil, fmt.Errorf("TTS 未配置")
	}
	hdr := http.Header{}
	setVolcAuthHeaders(hdr, cfg.APIKey, cfg.TTSResourceID)

	dialer := websocket.Dialer{HandshakeTimeout: 10 * time.Second}
	conn, resp, err := dialer.DialContext(ctx, ttsEndpoint, hdr)
	if err != nil {
		return nil, fmt.Errorf("连接火山 TTS 失败: %w%s", err, handshakeDetail(resp))
	}
	sid := newConnectID()
	base := ttsReqBase(cfg)

	// StartConnection → 期待 ConnectionStarted。
	if err := conn.WriteMessage(websocket.BinaryMessage, connEventFrame(evStartConnection, []byte("{}"))); err != nil {
		_ = conn.Close()
		return nil, fmt.Errorf("发送 StartConnection 失败: %w", err)
	}
	if err := expectTTSEvent(conn, evConnectionStarted); err != nil {
		_ = conn.Close()
		return nil, err
	}
	// StartSession → 期待 SessionStarted。
	if err := conn.WriteMessage(websocket.BinaryMessage, sessionEventFrame(evStartSession, sid, buildTTSStartSession(base))); err != nil {
		_ = conn.Close()
		return nil, fmt.Errorf("发送 StartSession 失败: %w", err)
	}
	if err := expectTTSEvent(conn, evSessionStarted); err != nil {
		_ = conn.Close()
		return nil, err
	}

	s := &TTSSession{
		conn: conn, sessionID: sid, reqBase: base,
		audio: make(chan []byte, 64), closed: make(chan struct{}),
	}
	go s.readLoop()
	return s, nil
}

// expectTTSEvent 同步读一帧，校验是期望的事件（握手阶段用，此时还没起 readLoop）。
func expectTTSEvent(conn *websocket.Conn, want int32) error {
	_ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
	_, data, err := conn.ReadMessage()
	conn.SetReadDeadline(time.Time{}) // 清除
	if err != nil {
		return fmt.Errorf("读 TTS 握手响应失败: %w", err)
	}
	r, perr := parseTTSFrame(data)
	if perr != nil {
		return fmt.Errorf("解析 TTS 握手响应失败: %w", perr)
	}
	ttsDebugf("握手帧: event=%d msgType=%d payload=%s", r.Event, r.MsgType, preview(r.Payload, r.IsAudio))
	if r.MsgType == ttsServerErr {
		return fmt.Errorf("TTS 握手被拒 code=%d: %s", r.Code, string(r.Payload))
	}
	if r.Event != want && r.Event != 0 { // 0=未带事件号的容错
		return fmt.Errorf("TTS 握手期待事件 %d，得 %d", want, r.Event)
	}
	return nil
}

// Speak 推一段文字给 TTS（TaskRequest / 事件 200）。可多次调用逐句推。
// payload 必须带完整 req_params（speaker+audio_params）再加 text——只发 {text} 火山收不到文字
// （联调实证：会回一个空 text 的 TTSSentenceStart 后直接结束，无音频）。
func (s *TTSSession) Speak(text string) error {
	rp := make(map[string]any, len(s.reqBase)+1)
	for k, v := range s.reqBase {
		rp[k] = v
	}
	rp["text"] = text
	payload, _ := json.Marshal(map[string]any{"req_params": rp})
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	return s.conn.WriteMessage(websocket.BinaryMessage, sessionEventFrame(evTaskRequest, s.sessionID, payload))
}

// Finish 发 FinishSession，告知本轮文字推完；服务端把剩余音频吐完后回 SessionFinished（Audio 随之关闭）。
func (s *TTSSession) Finish() error {
	s.writeMu.Lock()
	defer s.writeMu.Unlock()
	return s.conn.WriteMessage(websocket.BinaryMessage, sessionEventFrame(evFinishSession, s.sessionID, []byte("{}")))
}

// Audio 返回下行音频流（PCM 24k）；会话结束/出错时关闭。
func (s *TTSSession) Audio() <-chan []byte { return s.audio }

// Err 返回会话错误（Audio 关闭后读取）。
func (s *TTSSession) Err() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.err
}

func (s *TTSSession) setErr(err error) {
	s.mu.Lock()
	if s.err == nil {
		s.err = err
	}
	s.mu.Unlock()
}

// Close 关闭底层连接（读 goroutine 随之退出，Audio 关闭）。幂等。
func (s *TTSSession) Close() {
	s.closeOnce.Do(func() {
		close(s.closed)
		_ = s.conn.Close()
	})
}

func (s *TTSSession) readLoop() {
	defer close(s.audio)
	for {
		_, data, err := s.conn.ReadMessage()
		if err != nil {
			return // 连接关闭 / 读错误
		}
		r, perr := parseTTSFrame(data)
		if perr != nil {
			ttsDebugf("解析失败(%d 字节): %v", len(data), perr)
			continue
		}
		ttsDebugf("帧: event=%d msgType=%d isAudio=%v payload=%s", r.Event, r.MsgType, r.IsAudio, preview(r.Payload, r.IsAudio))
		switch {
		case r.MsgType == ttsServerErr:
			s.setErr(fmt.Errorf("火山 TTS 错误 code=%d: %s", r.Code, string(r.Payload)))
			return
		case r.Event == evSessionFailed:
			s.setErr(fmt.Errorf("火山 TTS 会话失败: %s", string(r.Payload)))
			return
		case r.Event == evSessionFinished || r.Event == evConnectionFinished:
			return // 本轮朗读完毕
		case r.IsAudio && len(r.Payload) > 0:
			select {
			case s.audio <- r.Payload:
			case <-s.closed:
				return
			}
		}
	}
}