1#!/usr/bin/env python
 2# -*- coding: utf-8 -*-
 3from pathlib import Path
 4
 5import anyio
 6from loguru import logger
 7
 8from asr.utils import audio_duration
 9from config import DOWNLOAD_DIR, PROXY, TTS
10from networking import hx_req
11from utils import markdown_to_text, rand_string
12
13
14async def edge_tts(texts: str, model: str = "", voice_name: str = "") -> dict:
15    """Edge TTS.
16
17    https://github.com/wangwangit/tts
18
19    Returns:
20        {"voice": str, "duration": int, "voice_name":str, "model":str}
21    """
22    model = model or TTS.EDGE_MODEL
23    voice_name = voice_name or TTS.EDGE_VOICE
24    raw_texts = markdown_to_text(texts)
25    logger.debug(f"TTS via {model}, voice: {voice_name}, texts: {texts}")
26    response = await hx_req(
27        f"{TTS.EDGE_DOMAIN}/v1/audio/speech",
28        "POST",
29        headers={"Content-Type": "application/json"},
30        json_data={"input": raw_texts, "voice": model, "speed": 1.0, "pitch": "0", "style": "general"},
31        proxy=PROXY.EDGE,
32        rformat="content",
33    )
34    if not isinstance(response.get("content"), bytes):
35        return {}
36    save_path = Path(DOWNLOAD_DIR) / f"{rand_string(8)}.mp3"
37    async with await anyio.open_file(save_path, "wb") as f:
38        await f.write(response["content"])
39    return {"voice": save_path, "duration": audio_duration(save_path), "voice_name": voice_name, "model": model}