主流TTS引擎在数字人直播场景下的性能对比测试报告

劳动争议中，加班费的举证责任如何分配？根据相关法律规定，劳动者主张加班费的，应当就加班事实的存在承担举证责任。但劳动者有证据证明用人单位掌握加班事实相关证据的，用人单位应当提供。如果用人单位不提供，将承担不利后果。建议劳动者在日常工作中注意保存考勤记录、加班审批单等工作证据。

这款农产品来自云南海拔1800米的高山种植区，全年日照超过2000小时，昼夜温差大，有利于糖分积累。果实采用自然农法种植，不使用化学农药和催熟剂。从采摘到发货不超过48小时，保证您收到的是最新鲜的果实。

说实话，我之前也觉得收纳很麻烦。但后来我发现了一个特别简单的方法，就是按使用频率分类。每天用的放在手边，每周用的放在柜子里，偶尔用的就收起来。试了一个月，真的省心很多。

import time
import asyncio
import edge_tts
import azure.cognitiveservices.speech as speechsdk
from aip import AipSpeech
import statistics

# ============ Edge TTS 测试 ============
async def test_edge_tts(text: str, voice: str = "zh-CN-XiaoxiaoNeural"):
    """测试Edge TTS的延迟和输出"""
    start_time = time.time()
    try:
        tts = edge_tts.Communicate(text, voice)
        await tts.save("edge_output.wav")
        elapsed = time.time() - start_time
        return {"success": True, "elapsed": elapsed, "size": None}
    except Exception as e:
        return {"success": False, "error": str(e)}


# ============ Azure TTS 测试 ============
def test_azure_tts(text: str, key: str, region: str):
    """测试Azure TTS"""
    start_time = time.time()
    try:
        speech_config = speechsdk.SpeechConfig(subscription=key, region=region)
        audio_config = speechsdk.audio.AudioOutputConfig(filename="azure_output.wav")
        synthesizer = speechsdk.SpeechSynthesizer(
            speech_config=speech_config, 
            audio_config=audio_config
        )
        result = synthesizer.speak_text_async(text).get()
        elapsed = time.time() - start_time
        return {
            "success": result.reason == speechsdk.ResultReason.SynthesizingAudioCompleted,
            "elapsed": elapsed
        }
    except Exception as e:
        return {"success": False, "error": str(e)}


# ============ 百度TTS测试 ============
def test_baidu_tts(text: str, app_id: str, api_key: str, secret_key: str):
    """测试百度TTS"""
    start_time = time.time()
    try:
        client = AipSpeech(app_id, api_key, secret_key)
        result = client.synthesis(text, 'zh', 1, {
            'vol': 5,
            'spd': 5,
            'pit': 5,
            'per': 0
        })
        elapsed = time.time() - start_time
        if not isinstance(result, dict):
            with open("baidu_output.mp3", "wb") as f:
                f.write(result)
            return {"success": True, "elapsed": elapsed}
        else:
            return {"success": False, "error": result.get("err_msg", "未知错误")}
    except Exception as e:
        return {"success": False, "error": str(e)}


def batch_test():
    """批量测试：每种引擎连续调用50次"""
    results = {"edge": [], "azure": [], "baidu": []}
    test_text = "劳动争议中，加班费的举证责任如何分配？"

    for i in range(50):
        # Edge TTS
        edge_result = asyncio.run(test_edge_tts(test_text))
        results["edge"].append(edge_result)

        # Azure TTS (需配置真实key)
        # azure_result = test_azure_tts(test_text, "your_key", "your_region")
        # results["azure"].append(azure_result)

        # 百度TTS (需配置真实key)
        # baidu_result = test_baidu_tts(test_text, "app_id", "api_key", "secret_key")
        # results["baidu"].append(baidu_result)

        time.sleep(0.5)  # 避免限流

    return results

import asyncio
import edge_tts
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=5))
async def tts_with_retry(text: str, output_path: str, voice: str = "zh-CN-XiaoxiaoNeural"):
    """带自动重试的TTS调用"""
    tts = edge_tts.Communicate(text, voice)
    await tts.save(output_path)

async def batch_tts(scripts: list, output_dir: str):
    """批量合成语音"""
    tasks = []
    for i, script in enumerate(scripts):
        full_text = script["opening"] + script["core"] + script["closing"]
        path = f"{output_dir}/audio_{i}.wav"
        tasks.append(tts_with_retry(full_text, path))

    # 并发执行，控制并发数
    semaphore = asyncio.Semaphore(5)
    async def limited_task(task):
        async with semaphore:
            return await task

    results = await asyncio.gather(*[limited_task(t) for t in tasks])
    return results