audio_codebuddy/hardware-integration/pc-simulator/tts_client.py

# TTS 客户端（使用新增的同步合成端点）⭐

import requests
import tempfile
import time
from config import Config


def synthesize(text: str, voice_id: str = None, speed: float = 1.0, emotion: str = None) -> str:
    """
    调用 TTS 服务同步合成语音（立即返回音频 URL）
    限制：<= 500 字
    """
    voice_id = voice_id or Config.TTS_VOICE_ID

    url = f'{Config.TTS_BASE_URL}/api/tts/synthesize'

    headers = {
        'Authorization': f'Bearer {Config.TTS_API_KEY}',
        'Content-Type': 'application/json',
    }

    payload = {
        'text': text,
        'voiceId': voice_id,
        'voiceParams': {'speed': speed, 'pitch': 0, 'volume': 50},
    }

    if emotion:
        payload['emotion'] = emotion

    preview = text[:30] + ('...' if len(text) > 30 else '')
    print(f'TTS 同步合成: "{preview}"')

    response = requests.post(url, json=payload, headers=headers, timeout=15)

    if response.status_code != 200:
        raise Exception(f'TTS 失败: {response.status_code} {response.text}')

    data = response.json()

    if data.get('code') != 0:
        raise Exception(f'TTS 错误: {data.get("message")}')

    audio_url = data['data']['audioUrl']
    duration = data['data'].get('duration', 0)
    print(f'TTS 完成: 时长 {duration:.1f}s')
    return audio_url


def download_audio(audio_url: str) -> str:
    """下载音频到本地临时文件"""
    print(f'下载音频: {audio_url}')
    response = requests.get(audio_url, timeout=30)
    response.raise_for_status()

    suffix = '.mp3' if '.mp3' in audio_url.lower() else '.wav'
    tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix=suffix)
    tmp_file.write(response.content)
    tmp_file.close()

    return tmp_file.name


def synthesize_and_download(text: str, voice_id: str = None, emotion: str = None) -> str:
    """同步合成 + 下载，返回本地文件路径（一步到位）"""
    audio_url = synthesize(text, voice_id, emotion=emotion)
    local_path = download_audio(audio_url)
    return local_path


def synthesize_async(text: str, voice_id: str = None, max_wait: int = 30) -> str:
    """
    长文本异步合成（> 500 字）
    回退方案：调 /tts/generate + 轮询 /tts/chapter-status
    """
    voice_id = voice_id or Config.TTS_VOICE_ID

    headers = {
        'Authorization': f'Bearer {Config.TTS_API_KEY}',
        'Content-Type': 'application/json',
    }

    # 1. 提交任务
    print(f'提交异步任务: "{text[:30]}..."')
    response = requests.post(
        f'{Config.TTS_BASE_URL}/api/tts/generate',
        json={
            'text': text,
            'voiceId': voice_id,
            'voiceParams': {'speed': 1.0, 'pitch': 0, 'volume': 50},
        },
        headers=headers,
        timeout=10,
    )
    data = response.json()

    if data.get('code') != 0:
        raise Exception(f'任务提交失败: {data.get("message")}')

    chapter_id = data['data']['chapterId']

    # 2. 轮询
    status_url = f'{Config.TTS_BASE_URL}/api/tts/chapter-status/{chapter_id}'
    for i in range(max_wait):
        time.sleep(1)
        status_resp = requests.get(status_url, headers=headers, timeout=10)
        status_data = status_resp.json()['data']

        if status_data.get('status') == 'completed':
            return status_data['audioUrl']
        elif status_data.get('status') == 'failed':
            raise Exception('异步任务失败')

        print(f'  进度 {i+1}/{max_wait}s...')

    raise Exception(f'异步任务超时（{max_wait}秒）')


def synthesize_auto(text: str, voice_id: str = None) -> str:
    """
    自动选择同步/异步合成
    - 文本 <= 500 字：同步（~800ms）
    - 文本 > 500 字：异步（3-5s）
    """
    if len(text) <= 500:
        return synthesize(text, voice_id)
    else:
        return synthesize_async(text, voice_id)


if __name__ == '__main__':
    # 测试
    test_text = '你好，我是用 TTS 同步端点生成的语音。'
    local_path = synthesize_and_download(test_text)
    print(f'音频已保存到: {local_path}')