import os, json, sys from uuid import uuid4 from typing import Literal import urllib.parse import hashlib from pydantic import BaseModel, Field, model_validator from typing import Literal, List, Optional, Dict, Any, Union from uuid import uuid4 import hashlib def convert_value_type(value: Any, type_: str): if value is None: return None if isinstance(value, str): value = urllib.parse.unquote(value) if type(value).__name__ == type_: # 如果值的类型和参数的类型一致，直接返回值 return value if type_ == "int": return int(value) elif type_ == "float": if isinstance(value, str) and value[-1] == "%": return float(value[:-1]) / 100 else: return float(value) elif type_ == "bool": if isinstance(value, bool): return value return str(value).lower() in ("true", "1", "t", "y", "yes", "allow", "allowed") else: # 默认为字符串 return str(value) class ParamItem(BaseModel): """ Represents a parameter item for a TTS task. Attributes: type (str): The data type of the parameter. default (Any): The default value of the parameter. alias (List[str]): The list of aliases for the parameter. label (Optional[str]): The label for the parameter. name (Optional[str]): The name of the parameter. description (Optional[str]): The description of the parameter. min_value (Optional[float]): The minimum value of the parameter. max_value (Optional[float]): The maximum value of the parameter. step (Optional[float]): The step value for the parameter. choices (Optional[List[str]]): The list of choices for the parameter. """ type: str component_type: Optional[str] = None default: Any alias: List[str] label: Optional[str] = None name: Optional[str] description: Optional[str] = None min_value: Optional[float] = None max_value: Optional[float] = None step: Optional[float] = None choices: Optional[List[str]] = None def __init__(self, **data): if not data.get("type"): data.update({"type": "str"}) super().__init__(**data) self.default = convert_value_type(self.default, self.type) def init_params_config(res: dict): result = {} for key, value in res.items(): if value.get("label") is None: value.update({"label": key}) value.update({"name": key}) result[key] = ParamItem(**value) return result class Base_TTS_Task(BaseModel): """ Base class for TTS (Text-to-Speech) tasks. Attributes: uuid (str): Unique identifier for the task. params_config (Dict[str, ParamItem]): Configuration parameters for the task. task_type (Literal["text", "ssml", "audio"]): Type of the task. Can be "text", "ssml", or "audio". audio_path (Optional[str]): Path to the audio file. src (Optional[str]): Source url. ssml (Optional[str]): SSML (Speech Synthesis Markup Language) text. text (Optional[str]): Text content. format (Optional[str]): Audio format. sample_rate (Optional[int]): Sample rate of the audio. loudness (Optional[float]): Loudness of the audio. speed (Optional[float]): Speed of the audio. stream (Optional[bool]): Flag indicating if the audio should be streamed. save_temp (Optional[bool]): Flag indicating if the temporary files should be saved. disabled_features (Optional[List[str]]): List of disabled features. """ uuid: str = None params_config: Dict[str, ParamItem] task_type: Literal["text", "ssml", "audio"] = Field(default="text") audio_path: Optional[str] = None src: Optional[str] = None ssml: Optional[str] = None text: Optional[str] = None character: Optional[str] = None emotion: Optional[str] = None format: Optional[str] = None sample_rate: Optional[int] = None loudness: Optional[float] = None speed: Optional[float] = None stream: Optional[bool] = None save_temp: Optional[bool] = False disabled_features: Optional[List[str]] = None class Config: populate_by_name = True extra = "ignore" def __init__(self, other_task: Union[BaseModel, dict, None] = None, **data): if isinstance(other_task, BaseModel): # 如果 task 是 Base_TTS_Task 实例，从该实例复制属性 data = other_task.model_dump() super().__init__(**data) else: # 如果 task 是字典，直接使用这个字典 if isinstance(other_task, dict): data = other_task assert "params_config" in data, "params_config is not defined." super().__init__(params_config=data.get("params_config")) self.set_default_values() self.set_values(**data) self.uuid = str(uuid4()) def update_value(self, key: str, value: Any, allow_none: bool = False): if not allow_none and value is None: return assert self.params_config is not None, "params_config is not defined." for param_key, param_value in self.params_config.items(): if key in param_value.alias: if hasattr(self, param_key): value = convert_value_type(value, param_value.type) setattr(self, param_key, value) else: pass # raise ValueError(f"Attribute {param_key} not found. Something went wrong in params_config.json.") def set_values(self, **data): assert self.params_config is not None, "params_config is not defined." for key, value in data.items(): if hasattr(self, key): value = convert_value_type(value, type(getattr(self, key)).__name__) setattr(self, key, value) else: self.update_value(key, value) def set_default_values(self): assert self.params_config is not None, "params_config is not defined." for key, value in self.params_config.items(): if ( hasattr(self, key) and getattr(self, key) is None and value.default is not None ): setattr(self, key, value.default) @property def md5(self): m = hashlib.md5() if self.task_type == "text": m.update(self.text.encode()) elif self.task_type == "ssml": m.update(self.ssml.encode()) elif self.task_type == "audio": m.update(self.src.encode()) return m.hexdigest() def __str__(self): dict_content: dict = self.model_dump(exclude={"params_config"}) # 收集所有值为None的键 keys_to_remove = [key for key, value in dict_content.items() if value is None] # 弹出这些键 for key in keys_to_remove: dict_content.pop(key) return json.dumps(dict_content, indent=4, ensure_ascii=False) def copy(self, update: Dict[str, Any] = {}, deep: bool = False): update["uuid"] = str(uuid4()) return super().model_copy(update=update, deep=deep)