fix: integrate Azure TTS with shared selector

This commit is contained in:
calesthio
2026-08-13 09:27:32 -07:00
parent a3c45aa3bf
commit 078eb7eecd
3 changed files with 88 additions and 11 deletions

View File

@@ -6,6 +6,7 @@ construction, and execute() guardrails.
"""
import sys
import xml.etree.ElementTree as ET
from pathlib import Path
import pytest
@@ -148,6 +149,20 @@ class TestSSML:
ssml = t._build_ssml({"text": "Hallo", "locale": "de-DE"}, "de-DE-KatjaNeural")
assert 'xml:lang="de-DE"' in ssml
def test_ssml_quotes_in_attributes_remain_well_formed(self):
t = AzureTTS()
ssml = t._build_ssml(
{
"text": 'She said "hello" & waved',
"locale": 'en-US" data-bad="yes',
"rate": '0%" data-bad="yes',
"style": 'calm" data-bad="yes',
},
'en-US-GuyNeural" data-bad="yes',
)
root = ET.fromstring(ssml)
assert all("data-bad" not in element.attrib for element in root.iter())
def test_host_prefers_explicit_endpoint(self, monkeypatch):
monkeypatch.setenv("AZURE_TTS_ENDPOINT", "https://custom.tts.example.com/")
assert AzureTTS()._host() == "https://custom.tts.example.com"
@@ -201,6 +216,39 @@ class TestExecute:
assert captured["headers"]["X-Microsoft-OutputFormat"] == "audio-48khz-192kbitrate-mono-mp3"
assert b"Hello world" in captured["body"]
def test_selector_adapts_shared_controls_for_azure(self, azure_env, tmp_path, monkeypatch):
import requests
from tools.audio.tts_selector import TTSSelector
captured = {}
def fake_post(url, headers=None, data=None, timeout=None):
captured["body"] = data.decode("utf-8")
return _FakeResponse()
monkeypatch.setattr(requests, "post", fake_post)
monkeypatch.setattr(TTSSelector, "_providers", lambda self: [AzureTTS()])
result = TTSSelector().execute(
{
"text": "Selector narration",
"preferred_provider": "azure",
"voice_id": "jenny",
"speaking_rate": 1.1,
"pitch": 2,
"style": 0.8,
"output_format": "mp3_44100_128",
"output_path": str(tmp_path / "selector.mp3"),
}
)
assert result.success
assert result.data["selected_tool"] == "azure_tts"
assert result.data["voice"] == "en-US-JennyNeural"
assert 'rate="+10%"' in captured["body"]
assert 'pitch="+2st"' in captured["body"]
assert "express-as" not in captured["body"]
def test_wav_output_format(self, azure_env, tmp_path, monkeypatch):
import requests

View File

@@ -19,7 +19,7 @@ import os
import time
from pathlib import Path
from typing import Any
from xml.sax.saxutils import escape
from xml.sax.saxutils import escape, quoteattr
from tools.base_tool import (
BaseTool,
@@ -201,23 +201,21 @@ class AzureTTS(BaseTool):
return self.RECOMMENDED_VOICES.get(voice.lower(), voice)
def _build_ssml(self, inputs: dict[str, Any], voice: str) -> str:
locale = inputs.get("locale", "en-US")
rate = inputs.get("rate", "0%")
pitch = inputs.get("pitch", "0%")
locale = str(inputs.get("locale", "en-US"))
rate = str(inputs.get("rate", "0%"))
pitch = str(inputs.get("pitch", "0%"))
style = inputs.get("style")
text = escape(inputs["text"])
inner = f'<prosody rate="{escape(rate)}" pitch="{escape(pitch)}">{text}</prosody>'
inner = f"<prosody rate={quoteattr(rate)} pitch={quoteattr(pitch)}>{text}</prosody>"
if style:
inner = (
f'<mstts:express-as style="{escape(style)}">{inner}</mstts:express-as>'
)
inner = f"<mstts:express-as style={quoteattr(str(style))}>{inner}</mstts:express-as>"
return (
f'<speak version="1.0" '
f'xmlns="http://www.w3.org/2001/10/synthesis" '
f'xmlns:mstts="https://www.w3.org/2001/mstts" '
f'xml:lang="{locale}">'
f'<voice name="{voice}">{inner}</voice></speak>'
f"xml:lang={quoteattr(locale)}>"
f"<voice name={quoteattr(voice)}>{inner}</voice></speak>"
)
def execute(self, inputs: dict[str, Any]) -> ToolResult:

View File

@@ -188,7 +188,7 @@ class TTSSelector(BaseTool):
if tool is None:
return ToolResult(success=False, error="No TTS provider available.")
result = tool.execute(inputs)
result = tool.execute(self._adapt_inputs(tool, inputs))
if result.success:
result.data.setdefault("selected_tool", tool.name)
result.data["selected_provider"] = tool.provider
@@ -202,6 +202,37 @@ class TTSSelector(BaseTool):
]
return result
@staticmethod
def _adapt_inputs(tool: BaseTool, inputs: dict[str, Any]) -> dict[str, Any]:
"""Translate capability-level controls to provider-native inputs."""
adapted = dict(inputs)
if tool.name != "azure_tts":
return adapted
if inputs.get("voice_id") and not inputs.get("voice"):
adapted["voice"] = inputs["voice_id"]
speed = inputs.get("speaking_rate", inputs.get("speed"))
if speed is not None and "rate" not in inputs:
percent = round((float(speed) - 1.0) * 100)
adapted["rate"] = f"{percent:+d}%" if percent else "0%"
pitch = inputs.get("pitch")
if isinstance(pitch, (int, float)):
adapted["pitch"] = f"{pitch:+g}st" if pitch else "0%"
# The selector's numeric style is ElevenLabs-specific. Azure's style
# is a named express-as value such as "calm" or "newscast".
if not isinstance(inputs.get("style"), str):
adapted.pop("style", None)
output_format = str(inputs.get("output_format", ""))
if output_format.startswith("mp3"):
adapted["output_format"] = "mp3"
elif output_format.startswith(("wav", "riff", "pcm")):
adapted["output_format"] = "wav"
return adapted
def _select_best_tool(
self,
inputs: dict[str, Any],