lenML/ChatTTS-Forge
301
1import json2import logging3import os4 5import pandas as pd6 7logger = logging.getLogger(__name__)8 9 10class DataExistsError(Exception):11 pass12 13 14class DataNotFoundError(Exception):15 pass16 17 18# FIXME: 😓这个东西写的比较拉跨,最好找个什么csv库替代掉...19class BaseManager:20 def __init__(self, csv_file):21 self.csv_file = csv_file22 self.columns = ["id", "name", "desc", "params"]23 if not os.path.exists(csv_file):24 df = pd.DataFrame(columns=self.columns)25 df.to_csv(self.csv_file, index=False)26 27 def _load_data(self):28 return pd.read_csv(self.csv_file)29 30 def _save_data(self, df):31 df.to_csv(self.csv_file, index=False)32 33 def add_item(self, item_id, name, desc, params):34 df = self._load_data()35 if item_id in df["id"].values:36 raise DataExistsError(f"Item ID {item_id} already exists.")37 new_row = pd.DataFrame(38 [39 {40 "id": item_id,41 "name": name,42 "desc": desc,43 "params": json.dumps(params, ensure_ascii=False),44 }45 ]46 )47 df = pd.concat([df, new_row], ignore_index=True)48 self._save_data(df)49 50 def delete_item(self, item_id):51 df = self._load_data()52 if item_id not in df["id"].values:53 raise DataNotFoundError(f"Item ID {item_id} not found.")54 df = df[df["id"] != item_id]55 self._save_data(df)56 57 def update_item(self, item_id, name=None, desc=None, params=None):58 df = self._load_data()59 if item_id not in df["id"].values:60 raise DataNotFoundError(f"Item ID {item_id} not found.")61 if name:62 df.loc[df["id"] == item_id, "name"] = name63 if desc:64 df.loc[df["id"] == item_id, "desc"] = desc65 if params:66 df.loc[df["id"] == item_id, "params"] = params67 self._save_data(df)68 69 def get_item(self, item_id):70 df = self._load_data()71 if item_id not in df["id"].values:72 raise DataNotFoundError(f"Item ID {item_id} not found.")73 item = df[df["id"] == item_id].to_dict("records")[0]74 item["params"] = json.loads(item["params"])75 return item76 77 def list_items(self):78 df = self._load_data()79 items = df.to_dict("records")80 for item in items:81 item["params"] = json.loads(item["params"])82 return items83 84 def find_item_by_name(self, name):85 df = self._load_data()86 if name not in df["name"].values:87 raise DataNotFoundError(f"Name {name} not found.")88 item = df[df["name"] == name].to_dict("records")[0]89 item["params"] = json.loads(item["params"])90 return item91 92 def find_params_by_name(self, name):93 try:94 return self.find_item_by_name(name)["params"]95 except Exception as e:96 logger.error(e)97 return {}98 99 def find_params_by_id(self, id):100 try:101 return self.get_item(id)["params"]102 except Exception as e:103 logger.error(e)104 return {}105 106 107# Usage example108if __name__ == "__main__":109 110 class SpeakerManager(BaseManager):111 def __init__(self, csv_file):112 super().__init__(csv_file)113 114 manager = SpeakerManager("speakers.test.csv")115 116 try:117 # Add speaker118 manager.add_item(119 1, "Speaker1", "Description for speaker 1", '{"param1": "value1"}'120 )121 except DataExistsError as e:122 print(e)123 124 # List all speakers125 speakers = manager.list_items()126 print(speakers)127 128 try:129 # Get specific speaker130 speaker = manager.get_item(1)131 print(speaker)132 except DataNotFoundError as e:133 print(e)134 135 try:136 # Update speaker137 manager.update_item(138 1, name="Updated Speaker1", desc="Updated description for speaker 1"139 )140 except DataNotFoundError as e:141 print(e)142 143 try:144 # Delete speaker145 manager.delete_item(1)146 except DataNotFoundError as e:147 print(e)148 149 try:150 # Find speaker by name151 speaker_by_name = manager.find_item_by_name("Updated Speaker1")152 print(speaker_by_name)153 except DataNotFoundError as e:154 print(e)155 156 # Find speakers by params157 speakers_by_params = manager.find_items_by_params('{"param1": "value1"}')158 print(speakers_by_params)159 