Digital_Human/tts_ceshi.py

from pathlib import Path
from typing import Optional
import time  # 新增：导入时间模块
import requests


URL = "http://127.0.0.1:1024/tts"


def tts_request(
    text: str,
    voice: Optional[str] = None,
    rate: Optional[str] = None,
    volume: Optional[str] = None,
    pitch: Optional[str] = None,
    output_path: str = "speech.wav",
):
    """Send text to Edge TTS server and save returned audio file.
    新增：记录请求开始/结束时间，计算总耗时
    """
    # ========== 新增：记录请求开始时间 ==========
    start_time = time.time()  # 开始时间（时间戳）
    start_time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(start_time))  # 格式化开始时间
    print(f"📅 请求开始时间: {start_time_str}")

    headers = {"Content-Type": "application/json"}
    data = {"text": text}
    if voice:
        data["voice"] = voice
    if rate:
        data["rate"] = rate
    if volume:
        data["volume"] = volume
    if pitch:
        data["pitch"] = pitch

    try:
        resp = requests.post(URL, json=data, headers=headers, timeout=30)
        
        # ========== 新增：记录请求结束时间 & 计算耗时 ==========
        end_time = time.time()
        end_time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(end_time))
        total_duration = round((end_time - start_time) * 1000, 2)  # 总耗时（毫秒）
        print(f"📅 请求结束时间: {end_time_str}")
        print(f"⏱️  请求总耗时: {total_duration} 毫秒")

        print("服务器响应状态码:", resp.status_code)
        print("服务器响应内容类型:", resp.headers.get("Content-Type"))

        # 成功返回音频流
        if resp.status_code == 200 and str(resp.headers.get("Content-Type", "")).startswith("audio/"):
            Path(output_path).write_bytes(resp.content)
            print(f"✅ 已保存音频到 {output_path}")
            # ========== 新增：返回结果中包含时间信息 ==========
            return {
                "file": output_path,
                "start_time": start_time_str,
                "end_time": end_time_str,
                "duration_ms": total_duration  # 耗时（毫秒）
            }

        # 非 200 时解析错误并抛出
        try:
            detail = resp.json()
        except Exception:
            detail = resp.text
        print("⚠️ 接口返回错误:", detail)
        resp.raise_for_status()
        return None

    except requests.exceptions.HTTPError as e:
        print(f"HTTP 错误: {e}")
        return None
    except requests.exceptions.ConnectionError:
        print("❌ 无法连接到服务器，请检查 IP/端口 或服务器是否在线")
        return None
    except requests.exceptions.Timeout:
        print("❌ 请求超时，请检查服务器网络")
        return None
    except Exception as e:
        print(f"其他错误: {str(e)}")
        return None


if __name__ == "__main__":
    # 示例：直接克隆文字
    result = tts_request("克隆文字", voice="zh-CN-YunxiaNeural", rate="+0%", volume="+0%", pitch="+0Hz")
    if result:
        print("请求成功:", result)
        # 示例：单独提取时间信息
        print(f"音频生成耗时: {result['duration_ms']} 毫秒")
    else:
        print("请求失败")