✨ 适配 GPT-Sovits v4 版本 (#56)

* ✨ 适配 GPT-Sovits v4版本

* 🚚 修改css文件命名，支持配置超时时间

Author: FrostN0v0

nonebot_plugin_deepseek/__init__.py

@@ -1,3 +1,6 @@
+import itertools
+from pathlib import Path
+from importlib import import_module
 from importlib.util import find_spec
 
 from nonebot import require
@@ -20,6 +23,7 @@
     MultiVar,
     Namespace,
     Subcommand,
+    UniMessage,
     CommandMeta,
     on_alconna,
 )
@@ -29,6 +33,8 @@
 if find_spec("nonebot_plugin_htmlrender"):
     require("nonebot_plugin_htmlrender")
     htmlrender_enable = True
+    text_to_pic = import_module("nonebot_plugin_htmlrender").text_to_pic
+
 else:
     htmlrender_enable = False
 
@@ -101,15 +107,15 @@
         ),
         Subcommand(
             "tts",
-            Option("-l|--list", help_text="支持的 TTS 模型列表"),
+            Option("-l|--list", Args["page?#页码", int], help_text="支持的 TTS 模型列表"),
             Option(
                 "--set-default",
                 Args[
                     "model#模型名称",
                     str,
                     Field(
                         completion=lambda: f"请输入 TTS 模型预设名，预期为："
-                        f"{json_config.available_tts_models[:10]}…… 其中之一\n"
+                        f"{list(json_config.available_tts_models.keys())[:10]}…… 其中之一\n"
                         "输入 `/deepseek tts -l` 查看所有 TTS 模型及角色"
                     ),
                 ],
@@ -213,29 +219,69 @@ async def _(
 
 
 @deepseek.assign("tts.list")
-async def _():
+async def _(
+    page: Query[int] = Query("tts.list.page"),
+):
     if not tts_config.enable_models:
         await deepseek.finish("当前未启用 TTS 功能")
-    if json_config.tts_model_dict:
-        model_list = "".join(
-            f"{model}\n - "
-            + "|".join(f"{spk}(默认)" if default_model.name == f"{model}-{spk}" else spk for spk in speakers)
-            + "\n"
-            for model, speakers in json_config.tts_model_dict.items()
+
+    def parse_model_dict(model_dict: dict[str, dict[str, list[str]]], start_index: int) -> str:
+        return "\n".join(
+            (f"{'✅️ ' if model_name == default_model.model_name else '⏹️'}{start_index + index + 1}.{model_name}")
+            for index, model_name in enumerate(model_dict.keys())
             if json_config.default_tts_model
             and (default_model := tts_config.get_tts_model(json_config.default_tts_model))
         )
-        custom_models = "\n".join(
-            f"- {model}（默认）" if model == json_config.default_tts_model else f"- {model}"
-            for model in tts_config.get_enable_tts()
+
+    if json_config.available_tts_models:
+        page_size = 200
+        page_num = page.result if page.available else 1
+        start_index = (page_num - 1) * page_size
+        page_model_dict = dict(
+            itertools.islice(json_config.available_tts_models.items(), start_index, start_index + page_size)
+        )
+        if not page_model_dict:
+            await deepseek.finish(f"页码 {page_num} 超出范围，没有找到任何模型。")
+
+        model_list_msg = parse_model_dict(page_model_dict, start_index)
+        custom_models = (
+            "\n".join(
+                f"{'✅️ ' if model.name == json_config.default_tts_model else '⏹️'}{index + 1}.{model.name}"
+                for index, model in enumerate(tts_config.enable_models)
+            )
+            if isinstance(tts_config.enable_models, list)
+            else ""
         )
-        custom_models_msg = f"\n自定义预设:\n{custom_models}"
     else:
         await deepseek.finish("当前未查找到可用模型")
 
-    message = f"支持的 TTS 模型列表: \n{model_list}"
-    if isinstance(tts_config.enable_models, list):
-        message += custom_models_msg
+    total_models = len(json_config.available_tts_models)
+    total_pages = (total_models + page_size - 1) // page_size
+
+    if page_num > total_pages or page_num < 1:
+        await deepseek.finish("请输入正确的页码")
+
+    header_msg = (
+        f"支持的 TTS 模型列表 \n(第 {page_num}/{total_pages} 页, 共 {total_models} 个):\n\n"
+        f"当前TTS模型:\n✅️ {json_config.default_tts_model}\n\n"
+    )
+    message = (
+        (f"自定义 TTS 模型预设:\n {custom_models}" if isinstance(tts_config.enable_models, list) else "")
+        + f"\n\n{header_msg}"
+        + model_list_msg
+    )
+    if htmlrender_enable:
+        custom_models_html = "".join(f"<div>{line}</div>" for line in custom_models.split("\n") if line)
+        header_html = (
+            f"<header class='custom-header'>"
+            f"<h2 class='header-title'>自定义 TTS 预设</h2>"
+            f"<div class='models-container'>{custom_models_html}</div></header>"
+        )
+        model_lines = "".join(f"<div>{line}</div>" for line in model_list_msg.split("\n") if line)
+        model_html = f"<h2 class='header-title'>{header_msg}</h2><div class='models-container'>{model_lines}</div>"
+        final_html = header_html + model_html
+        css_path = str(Path(__file__).parent / "resources/tts_models.css")
+        await deepseek.finish(UniMessage.image(raw=await text_to_pic(text=final_html, css_path=css_path, width=1440)))
     await deepseek.finish(message)
 
 
@@ -248,10 +294,11 @@ async def _(
         await deepseek.finish("当前未启用 TTS 功能")
     if not is_superuser:
         await deepseek.finish("该指令仅超管可用")
-    if model.result not in json_config.available_tts_models:
+    available_tts_model_names = list(json_config.available_tts_models.keys()) + tts_config.get_enable_tts()
+    if model.result not in available_tts_model_names:
         await deepseek.finish(
             f"请输入 TTS 模型预设名，预期为："
-            f"{json_config.available_tts_models[:10]}…… 其中之一\n"
+            f"{list(json_config.available_tts_models.keys())[:10]}…… 其中之一\n"
             "输入 `/deepseek tts -l` 查看所有 TTS 模型及角色"
         )
     json_config.default_tts_model = model.result

nonebot_plugin_deepseek/apis/request.py

@@ -9,7 +9,7 @@
 from ..log import ds_logger, tts_logger
 from ..exception import RequestException
 from ..config import ds_config, tts_config, json_config, uninfo_enable
-from ..schemas import Balance, TTSResponse, ChatCompletions, StreamChoiceList
+from ..schemas import Balance, TTSModelInfo, ChatCompletions, StreamChoiceList
 
 
 class API:
@@ -63,66 +63,56 @@ async def query_balance(cls, model_name: str) -> Balance:
         return Balance(**response.json())
 
     @classmethod
-    async def get_tts_models(cls) -> list[TTSResponse]:
+    async def get_tts_models(cls) -> list[TTSModelInfo]:
         try:
             async with httpx.AsyncClient() as client:
-                response = await client.get(
+                response = await client.post(
                     f"{tts_config.base_url}/models",
                     headers={**cls._headers},
-                    timeout=30,
+                    json={"version": tts_config.tts_version},
+                    timeout=tts_config.timeout,
                 )
             if response.status_code != 200:
                 raise RequestException(f"获取 TTS 模型列表失败，状态码: {response.status_code}")
-            return [await TTSResponse.create(model=model) for model in response.json()]
+            return [
+                TTSModelInfo(model_name=key, language_emotions=value)
+                for key, value in response.json().get("models", {}).items()
+                if isinstance(value, dict)
+            ]
         except httpx.ConnectError as e:
             raise RequestException(f"连接 TTS 模型服务器失败: {e}")
 
-    @classmethod
-    async def get_tts_speakers(cls, model_name: str) -> list[str]:
-        async with httpx.AsyncClient() as client:
-            response = await client.post(
-                f"{tts_config.base_url}/spks",
-                headers={**cls._headers},
-                json={"model": model_name},
-                timeout=30,
-            )
-        if speakers := response.json().get("speakers"):
-            return list(speakers.keys())
-        else:
-            raise RequestException("获取 TTS 模型讲话人列表失败")
-
     @classmethod
     async def text_to_speach(cls, text: str, model: str) -> bytes:
         model_config = tts_config.get_tts_model(model)
         model_name = model_config.model_name
-        speaker = model_config.speaker_name
         json = {
             "text": text,
             "model_name": model_name,
-            "speaker_name": speaker,
             "app_key": tts_config.access_token,
             "access_token": tts_config.access_token,
-            "audio_dl_url": tts_config.audio_dl_url,
+            "version": tts_config.tts_version,
+            "dl_url": tts_config.dl_url,
             **model_config.to_dict(),
         }
 
-        tts_logger("DEBUG", f"使用模型 {model}，讲话人：{speaker}, 配置：{json}")
+        tts_logger("DEBUG", f"使用模型 {model}，配置：{json}")
         try:
             async with httpx.AsyncClient() as client:
                 response = await client.post(
                     f"{tts_config.base_url}/infer_single",
-                    headers={**cls._headers},
+                    headers={**cls._headers, "Authorization": f"Bearer {tts_config.access_token}"},
                     json=json,
-                    timeout=50,
+                    timeout=tts_config.timeout,
                 )
-            tts_logger("DEBUG", f"Response: {response.text}")
+            tts_logger("DEBUG", f"Response: {response.status_code} {response.text}")
             if audio_url := response.json().get("audio_url"):
                 async with httpx.AsyncClient() as client:
-                    response = await client.get(audio_url)
+                    response = await client.get(audio_url, timeout=tts_config.timeout)
                     return response.content
             else:
                 raise RequestException("语音合成失败")
-        except httpx.ConnectError as e:
+        except (httpx.ConnectError, httpx.ReadTimeout) as e:
             raise RequestException(f"连接 TTS 服务器失败: {e}")
 
 

nonebot_plugin_deepseek/cli/plugins/tts.py

@@ -31,10 +31,7 @@ def dispatch(self, result: Arparma) -> Union[bool, None]:
         if result.find("tts.update"):
             available_models = asyncio.run(tts_config.get_available_tts())
             if available_models:
-                json_config.available_tts_models = [
-                    f"{model}-{spk}" for model, speakers in available_models.items() for spk in speakers
-                ]
-                json_config.tts_model_dict = available_models
+                json_config.available_tts_models = available_models
                 json_config.save()
                 tts_logger("SUCCESS", f"Update available TTS models: {available_models}")
             return

nonebot_plugin_deepseek/config.py

@@ -39,8 +39,7 @@ def __init__(self) -> None:
         self.file: Path = store.get_plugin_config_dir() / "config.json"
         self.default_model: str = ds_config.get_enable_models()[0]
         self.enable_md_to_pic: bool = ds_config.md_to_pic
-        self.tts_model_dict: dict[str, list[str]] = {}
-        self.available_tts_models: list[str] = []
+        self.available_tts_models: dict[str, dict[str, list[str]]] = {}
         self.default_tts_model: Optional[str] = None
 
         self.prompt_func: Optional[Callable[[dict[str, Any]], str]] = None
@@ -58,10 +57,7 @@ def load(self):
             self.enable_md_to_pic = data.get("enable_md_to_pic", self.enable_md_to_pic)
             self.default_tts_model = data.get("default_tts_model")
             if isinstance(data.get("available_tts_models"), dict):
-                self.tts_model_dict = data.get("available_tts_models")
-                self.available_tts_models = [
-                    f"{model}-{spk}" for model, speakers in self.tts_model_dict.items() for spk in speakers
-                ] + (tts_config.get_enable_tts() if tts_config.enable_models else [])
+                self.available_tts_models = data.get("available_tts_models")
 
         enable_models = ds_config.get_enable_models()
         if self.default_model not in enable_models:
@@ -70,8 +66,10 @@ def load(self):
         if self.enable_md_to_pic != ds_config.md_to_pic:
             self.enable_md_to_pic = ds_config.md_to_pic
             self.save()
-        if self.available_tts_models and self.default_tts_model not in self.available_tts_models:
-            self.default_tts_model = self.available_tts_models[0]
+        if self.available_tts_models and self.default_tts_model not in (
+            list(self.available_tts_models.keys()) + tts_config.get_enable_tts()
+        ):
+            self.default_tts_model = list(self.available_tts_models.keys())[0]
             self.save()
         if not self.available_tts_models and self.default_tts_model:
             self.save()
@@ -81,10 +79,10 @@ def save(self):
             "default_model": self.default_model,
             "enable_md_to_pic": self.enable_md_to_pic,
         }
-        if self.default_tts_model in self.available_tts_models:
+        if self.default_tts_model in (list(self.available_tts_models.keys()) + tts_config.get_enable_tts()):
             config_data["default_tts_model"] = self.default_tts_model
         if self.available_tts_models:
-            config_data["available_tts_models"] = self.tts_model_dict
+            config_data["available_tts_models"] = self.available_tts_models
         with open(self.file, "w", encoding="utf-8") as f:
             json.dump(config_data, f, ensure_ascii=False, indent=2)
         self.prompt_func = None
@@ -206,15 +204,15 @@ def to_dict(self):
 class CustomTTS(BaseModel):
     name: str
     """TTS Preset Parameters Name"""
+    version: str = "v4"
+    """GPT-Sovits API Version"""
     model_name: str
     """TTS Model Name"""
-    speaker_name: str
-    """TTS Speaker Name"""
     prompt_text_lang: str = "中文"
     """language of the prompt text for the reference audio"""
-    emotion: str = "随机"
+    emotion: str = "默认"
     """Emotion"""
-    text_lang: str = "中文"
+    text_lang: str = "多语种混合"
     """language of the text to be synthesized"""
     top_k: int = Field(default=10, ge=1, le=100)
     """top k sampling"""
@@ -242,6 +240,10 @@ class CustomTTS(BaseModel):
     """repetition penalty for T2S model."""
     seed: int = -1
     """random seed for reproducibility."""
+    sample_steps: int = 16
+    """Number of steps sampled."""
+    if_sr: bool = False
+    """whether to use super-resolution model."""
 
     if PYDANTIC_V2:
         model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
@@ -306,8 +308,11 @@ class ScopedTTSConfig(BaseModel):
     """Your GPT-Sovits API Url """
     access_token: str = ""
     """Your GPT-Sovits API Access Token"""
-    audio_dl_url: str = ""
+    tts_version: str = "v4"
+    """Your GPT-Sovits API Version"""
+    dl_url: str = ""
     """audio download url"""
+    timeout: int = Field(default=60)
 
     @model_validator(mode="before")
     @classmethod
@@ -321,12 +326,12 @@ def get_enable_tts(self) -> list[str]:
             return []
         return [model.name for model in self.enable_models]
 
-    async def get_available_tts(self) -> dict[str, list[str]]:
+    async def get_available_tts(self) -> dict[str, dict[str, list[str]]]:
         from .apis import API
 
         try:
             tts_models = await API.get_tts_models()
-            preset_dict = {model.model: list(model.speakers) for model in tts_models}
+            preset_dict = {model.model_name: model.language_emotions for model in tts_models}
         except RequestException as e:
             preset_dict = {}
             tts_logger("WARNING", f"获取 TTS 模型列表失败: {e}")
@@ -336,16 +341,10 @@ def get_tts_model(self, preset_name: str) -> CustomTTS:
         """Get TTS model config"""
         if not isinstance(self.enable_models, bool):
             for model in self.enable_models:
-                if (
-                    model.name == preset_name
-                    and f"{model.model_name}-{model.speaker_name}" in json_config.available_tts_models
-                ):
+                if model.name == preset_name and f"{model.model_name}" in json_config.available_tts_models:
                     return model
-        if "-" in preset_name:
-            model_name = preset_name.split("-")[0]
-            speaker_name = preset_name.split("-")[1]
-            if preset_name in json_config.available_tts_models:
-                return CustomTTS(name=preset_name, model_name=model_name, speaker_name=speaker_name)
+        if preset_name in json_config.available_tts_models:
+            return CustomTTS(name=preset_name, model_name=preset_name)
         raise ValueError(f"TTS Model {preset_name} not valid")
 
 

nonebot_plugin_deepseek/hook.py

@@ -14,14 +14,11 @@ async def _() -> None:
     if tts_config.enable_models:
         if not json_config.available_tts_models:
             available_models = await tts_config.get_available_tts()
-            json_config.available_tts_models = [
-                f"{model}-{spk}" for model, speakers in available_models.items() for spk in speakers
-            ]
-            json_config.tts_model_dict = available_models
+            json_config.available_tts_models = available_models
             json_config.save()
-            tts_logger("DEBUG", f"Loaded available TTS models: {available_models}")
+            tts_logger("DEBUG", f"Loaded available TTS models: {len(available_models)}")
         else:
-            tts_logger("DEBUG", f"Loaded available TTS models: {json_config.available_tts_models}")
+            tts_logger("DEBUG", f"Loaded available TTS models: {len(json_config.available_tts_models)}")
     command_manager.load_cache(cach_dir)
     ds_logger("DEBUG", "DeepSeek shortcuts cache loaded")