Python

自定義TTS Server代碼示例如下：

from aiohttp import web


async def stream_audio(request):
    data = await request.json()
    text = data.get('Text', "")
    token = data.get('Token', None)
    sample_rate = data.get('SampleRate', 48000)
    extend_data = data.get('ExtendData', "")
    print(f"text:{text}, token:{token}, sample_rate:{sample_rate}, extend_data:{extend_data}")
    # TODO:校驗Token是否合法

    response = web.StreamResponse(
        status=200,
        reason='OK',
        headers={'Content-Type': 'audio/mpeg'}
    )

    # 開始響應
    await response.prepare(request)

    # generate_tts_data 是一個協程，用于生成音頻數據
    async for chunk in generate_tts_data(text, sample_rate):
        await response.write(chunk)

    # 完成響應
    await response.write_eof()

    return response


async def generate_tts_data(text: str, sample_rate: int):
    # TODO: 調用TTS服務，生成對應音頻對應采樣率的音頻數據
    # 示例：從文件讀取音頻數據
    file_path = '/your_dir/sample.pcm'
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(4096)  # 每次讀取4KB的數據
            if not chunk:
                break
            yield chunk

app = web.Application()
app.add_routes([web.post('/stream-audio', stream_audio)])

if __name__ == '__main__':
    web.run_app(app)