biosn2 commited on
Commit
653367c
·
verified ·
1 Parent(s): 93aed36

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +213 -228
app.py CHANGED
@@ -1,9 +1,9 @@
1
  # ```python
2
  #!/usr/bin/env python3
3
  """
4
- ChatTTS Gradio 应用 - 改编为使用 ChatterboxTTS
5
- 适配 Hugging Face Spaces 运行环境
6
- 基于 Chatterbox-TTS 实现: https://huggingface.co/spaces/ResembleAI/Chatterbox
7
  """
8
 
9
  import os
@@ -42,25 +42,25 @@ warnings.filterwarnings("ignore", category=FutureWarning)
42
  warnings.filterwarnings("ignore", category=UserWarning)
43
 
44
  import argparse
45
- # ----------------- 命令行参数解析 -----------------
46
  parser = argparse.ArgumentParser(description="IndexTTS WebUI")
47
- parser.add_argument("--verbose", action="store_true", default=False, help="Enable verbose mode") # 是否打印详细日志
48
- parser.add_argument("--port", type=int, default=7860, help="Port to run the web UI on") # WebUI 端口
49
- parser.add_argument("--host", type=str, default="127.0.0.1", help="Host to run the web UI on") # WebUI 主机地址
50
- parser.add_argument("--model_dir", type=str, default="checkpoints", help="Model checkpoints directory") # 模型目录
51
  cmd_args = parser.parse_args()
52
  model_dir="checkpoints"
53
 
54
- # ----------------- 设置模块搜索路径 -----------------
55
  current_dir = os.path.dirname(os.path.abspath(__file__))
56
  sys.path.append(current_dir)
57
  sys.path.append(os.path.join(current_dir, "indextts"))
58
 
59
- # ----------------- 下载模型 -----------------
60
  MODE = 'local'
61
- snapshot_download("IndexTeam/IndexTTS-1.5", local_dir="checkpoints") # Hugging Face 下载模型到本地
62
 
63
- # ----------------- 检查模型文件完整性 -----------------
64
  if not os.path.exists(cmd_args.model_dir):
65
  print(f"Model directory {cmd_args.model_dir} does not exist. Please download the model first.")
66
  sys.exit(1)
@@ -76,23 +76,23 @@ for file in [
76
  print(f"Required file {file_path} does not exist. Please download it.")
77
  sys.exit(1)
78
  def analyze_sentiment(text):
79
- """使用 TextBlob 分析文本情感,返回优化后的 IndexTTS 参数字典"""
80
  try:
81
  blob = TextBlob(text)
82
- polarity = blob.sentiment.polarity # -1 (负面) ~ 1 (正面)
83
- subjectivity = blob.sentiment.subjectivity # 0 (客观) ~ 1 (主观)
84
 
85
- # temperature: 正面高 (生动),负面低 (平淡);范围 0.5~1.5
86
  temperature = 1.0 + math.tanh(polarity) * 0.9 # 0.5 ~ 1.5
87
- # top_p: 主观高 (多样),客观低 (保守);范围 0.7~0.95
88
  top_p = 0.7 + subjectivity * 0.25
89
- # top_k: 主观高 (创意),但固定上限;范围 20~50
90
  top_k = int(20 + subjectivity * 30)
91
- # repetition_penalty: 客观高 (避免重复),主观低 (允许强调);范围 5.0~15.0
92
  repetition_penalty = 10.0 + (1 - subjectivity) * 5.0 - polarity * 2.0
93
- # length_penalty: 正面正 (长表达),负面负 ();范围 -1.0~1.0
94
  length_penalty = math.tanh(polarity) * 1.0
95
- # num_beams: 正面高 (自然),但固定以控制速度;范围 2~5
96
  num_beams = int(2 + (polarity + 1) * 1.5)
97
 
98
  return {
@@ -104,8 +104,8 @@ def analyze_sentiment(text):
104
  "num_beams": max(2, min(5, num_beams))
105
  }
106
  except Exception as e:
107
- print(f"情感分析失败: {str(e)}")
108
- return { # 默认值
109
  "temperature": 1.0,
110
  "top_p": 0.8,
111
  "top_k": 30,
@@ -115,28 +115,28 @@ def analyze_sentiment(text):
115
  }
116
 
117
 
118
- # ----------------- 导入 Gradio 和其他模块 -----------------
119
  import gradio as gr
120
  import pandas as pd
121
 
122
- from indextts.infer import IndexTTS # 核心 TTS 推理类
123
- from tools.i18n.i18n import I18nAuto # 国际化工具
124
 
125
- # ----------------- 初始化 TTS 模型 -----------------
126
- i18n = I18nAuto(language="en") # 设置默认中文
127
- tts = IndexTTS(model_dir=cmd_args.model_dir, cfg_path=os.path.join(cmd_args.model_dir, "config.yaml")) # 加载模型
128
 
129
- # # ----------------- 创建输出目录 -----------------
130
- # os.makedirs("outputs/tasks", exist_ok=True)
131
- # os.makedirs("prompts", exist_ok=True)
132
 
133
- # ----------------- 核心函数 -----------------
134
 
135
  def ensure_wav(file_path):
136
  """
137
- 确保输入音频是 WAV 格式
138
- 如果不是 WAV,使用 ffmpeg 转换
139
- 返回 WAV 文件路径
140
  """
141
  if not file_path.lower().endswith(".wav"):
142
  wav_path = file_path.rsplit(".", 1)[0] + ".wav"
@@ -146,102 +146,102 @@ def ensure_wav(file_path):
146
 
147
  # def progress_print(step, total, info=""):
148
  # """
149
- # 打印生成音频的进度到终端
150
- # step: 当前步骤
151
- # total: 总步骤数
152
- # info: 附加信息
153
  # """
154
  # percent = int(step / total * 100)
155
  # print(f"\r[{percent}%] {info}", end="", flush=True)
156
 
157
 
158
- # 设置日志
159
  logging.basicConfig(level=logging.INFO)
160
  logger = logging.getLogger(__name__)
161
 
162
- # 全局变量用于存储最后上传的文件
163
  last_uploaded_file = None
164
 
165
- # 全局取消标志
166
  cancel_generation = threading.Event()
167
 
168
- # 全局生成状态和队列
169
  is_generating = threading.Event()
170
- generation_queue = [] # 存储 {'role_index': i, 'txt': str, 'start_idx': int, 'spk_emb_seed_id': str, 'ref_wav': str}
171
  queued_roles = set()
172
 
173
- # 存储每个角色的 make_bulk 函数
174
  make_bulk_functions = [None] * 40
175
 
176
- # 临时目录用于缓存音频
177
  audio_cache_dir = tempfile.mkdtemp(prefix='audio_cache_')
178
 
179
- # 在脚本启动时清空缓存文件夹
180
  def clear_audio_cache_dir():
181
- """清空或重新创建 audio_cache_dir"""
182
  global audio_cache_dir
183
  try:
184
  if os.path.exists(audio_cache_dir):
185
  shutil.rmtree(audio_cache_dir)
186
- print(f"\033[92m✅ 已清空缓存文件夹: {audio_cache_dir}\033[0m")
187
  os.makedirs(audio_cache_dir)
188
- print(f"\033[92m✅ 创建新的缓存文件夹: {audio_cache_dir}\033[0m")
189
  except Exception as e:
190
- print(f"\033[91m❌ 清空缓存文件夹失败: {str(e)}\033[0m")
191
 
192
- # Monkey patch torch.load 以处理设备映射
193
  original_torch_load = torch.load
194
 
195
  def patched_torch_load(f, map_location=None, **kwargs):
196
  """
197
- 修补后的 torch.load,自动将 CUDA 张量映射到 CPU/CUDA
198
  """
199
  if map_location is None:
200
  map_location = 'cpu'
201
- logger.info(f"🔧 使用 map_location={map_location} 加载")
202
  return original_torch_load(f, map_location=map_location, **kwargs)
203
 
204
- # torch 导入后立即应用补丁
205
  torch.load = patched_torch_load
206
  if 'torch' in sys.modules:
207
  sys.modules['torch'].load = patched_torch_load
208
 
209
- logger.info("✅ 成功应用 torch.load 设备映射补丁")
210
 
211
- # 设备检测,适配 Hugging Face Spaces
212
  if torch.cuda.is_available():
213
  DEVICE = "cuda"
214
- logger.info("🚀 使用 CUDA GPU 运行")
215
  else:
216
  DEVICE = "cpu"
217
- logger.info("🚀 使用 CPU 运行")
218
 
219
- print(f"🚀 在设备上运行: {DEVICE}")
220
 
221
- # 全局模型变量
222
  MODEL = True
223
 
224
  def get_or_load_model():
225
- """加载 ChatterboxTTS 模型(如果尚未加载),并确保其在正确的设备上运行"""
226
  global MODEL, DEVICE
227
  if MODEL is None:
228
- print("模型未加载,正在初始化...")
229
  try:
230
- # 优先尝试官方导入路径
231
  try:
232
  from chatterbox.src.chatterbox.tts import ChatterboxTTS
233
- logger.info("✅ 使用官方 chatterbox.src 导入路径")
234
  except ImportError:
235
- # 回退到备用导入路径
236
  from chatterbox import ChatterboxTTS
237
- logger.info("✅ 使用 chatterbox 直接导入路径")
238
 
239
- # 先将模型加载到 CPU 以避免设备问题
240
  MODEL = ChatterboxTTS.from_pretrained("cpu")
241
 
242
- # 如果是 CUDA,移动到目标设备
243
  if DEVICE == "cuda":
244
- logger.info(f"将模型组件移动到 {DEVICE}...")
245
  try:
246
  if hasattr(MODEL, 't3'):
247
  MODEL.t3 = MODEL.t3.to(DEVICE)
@@ -251,28 +251,28 @@ def get_or_load_model():
251
  MODEL.ve = MODEL.ve.to(DEVICE)
252
 
253
  MODEL.device = DEVICE
254
- logger.info(f"✅ 所有模型组件已移动到 {DEVICE}")
255
  except Exception as e:
256
- logger.warning(f"⚠️ 无法将某些组件移动到 {DEVICE}: {e}")
257
- logger.info("🔄 为确保稳定性回退到 CPU 模式")
258
  DEVICE = "cpu"
259
  MODEL.device = "cpu"
260
 
261
- logger.info(f"✅ 模型在 {DEVICE} 上加载成功")
262
- return MODEL, "模型加载成功"
263
 
264
  except Exception as e:
265
- logger.error(f"❌ 加载模型失败: {e}")
266
- return None, f"模型加载失败: {str(e)}"
267
- return MODEL, "模型已加载"
268
 
269
  def parse_role_mappings(mapping_text):
270
- """解析角色映射文本,返回 {角色代号: {'display_name': 显示名, 'voice_file': 'xxx.wav', 'seed_id': seed_id}} 字典"""
271
  role_map = {}
272
  if not mapping_text:
273
  return role_map
274
  lines = mapping_text.strip().splitlines()
275
- pattern = re.compile(r'角色代号:\s*(\w+)\s*→\s*显示名:\s*([^→]+?)(?:\s+voice:\s*([\w_-]+))?(?:\s+seed_id:\s*(\w+))?\s*$')
276
 
277
  for line in lines:
278
  match = pattern.match(line.strip())
@@ -291,7 +291,7 @@ def parse_role_mappings(mapping_text):
291
  return role_map
292
 
293
  def filter_lines(lines, filter_text):
294
- """根据筛选条件过滤对话行"""
295
  if not filter_text:
296
  return lines
297
  filtered = []
@@ -308,13 +308,13 @@ def filter_lines(lines, filter_text):
308
  if len(re.sub(r'[^A-Za-z0-9]', '', line)) < min_len:
309
  include = False
310
  except ValueError:
311
- print(f"警告: 无效的 min_len 条件: {condition}")
312
  elif condition.startswith('keyword='):
313
  keyword = condition.split('=')[1].lower()
314
  if keyword not in line.lower():
315
  include = False
316
  else:
317
- print(f"警告: 未知筛选条件: {condition}")
318
  if include:
319
  filtered.append(line)
320
  return filtered
@@ -348,7 +348,7 @@ def parse_roles_from_rpy_file(file_bytes, filter_text=""):
348
 
349
  return role_dict, lines
350
  except Exception as e:
351
- raise ValueError(f"解析 .rpy 文件失败: {str(e)}")
352
 
353
  def save_wav(audio_data, sample_rate, output_path):
354
  try:
@@ -358,30 +358,11 @@ def save_wav(audio_data, sample_rate, output_path):
358
  audio_int16 = audio_data
359
  write(output_path, sample_rate, audio_int16)
360
  except Exception as e:
361
- raise RuntimeError(f"保存音频文件失败: {str(e)}")
362
 
363
  # def analyze_sentiment(text):
364
- # """使用 TextBlob 分析文本情感,返回 temperature, cfg_weight 和 exaggeration 参数"""
365
- # try:
366
- # blob = TextBlob(text)
367
- # polarity = blob.sentiment.polarity
368
- # subjectivity = blob.sentiment.subjectivity
369
-
370
- # temperature = 0.5 + math.tanh(polarity) * 0.75
371
- # temperature = max(0.05, min(2.0, temperature))
372
-
373
- # cfg_weight = 0.5 + (1 - subjectivity) * 0.45
374
- # cfg_weight = max(0.1, min(1.0, cfg_weight))
375
-
376
- # exaggeration = 0.55 + math.tanh(polarity) * 0.45
377
- # exaggeration = max(0.1, min(1.0, exaggeration))
378
-
379
- # return temperature, cfg_weight, exaggeration
380
- # except Exception as e:
381
- # print(f"情感分析失败: {str(e)}")
382
- # return 0.3, 0.5, 0.25
383
  def pre_analyze_lines(texts):
384
- """为每行文本分析情感,返回参数字典列表"""
385
  context_window = 2
386
  params_list = []
387
  for idx, line in enumerate(texts):
@@ -403,7 +384,7 @@ def pre_analyze_lines(texts):
403
  return params_list
404
 
405
  def get_pt_file(seed_id, csv_path=os.path.join(os.path.dirname(__file__), "evaluation_results.csv")):
406
- """根据 seed_id CSV 文件获取 .pt 数据并加载为 PyTorch tensor"""
407
  try:
408
  if seed_id and not seed_id.startswith("seed_"):
409
  seed_id = f"seed_{seed_id}"
@@ -411,16 +392,16 @@ def get_pt_file(seed_id, csv_path=os.path.join(os.path.dirname(__file__), "evalu
411
  df = pd.read_csv(csv_path, encoding="utf-8")
412
  row = df[df["seed_id"] == seed_id]
413
  if row.empty:
414
- return None, f"未找到 seed_id: {seed_id}"
415
 
416
  emb_data = row.iloc[0]["emb_data"]
417
  emb_bytes = base64.b64decode(emb_data)
418
  emb_buffer = io.BytesIO(emb_bytes)
419
  spk_emb = torch.load(emb_buffer)
420
 
421
- return spk_emb, f"成功加载 seed_id: {seed_id} 的 spk_emb 数据"
422
  except Exception as e:
423
- return None, f"加载 spk_emb 数据失败: {str(e)}"
424
 
425
  def zip_outputs_folder():
426
  try:
@@ -435,18 +416,18 @@ def zip_outputs_folder():
435
  file_path = os.path.join(root, file)
436
  arcname = os.path.relpath(file_path, outputs_dir)
437
  zipf.write(file_path, os.path.join("Outputs", arcname))
438
- return zip_path, "Outputs 文件夹已成功打包为 outputs.zip"
439
  else:
440
- return None, "Outputs 文件夹不存在或为空"
441
  except Exception as e:
442
- return None, f"打包 Outputs 文件夹失败: {str(e)}"
443
 
444
  def compress_wav_to_mp3():
445
- """ Outputs 文件夹及其子目录中的 .wav 文件转换为 80kbps .mp3 文件,并替换原文件"""
446
  try:
447
  outputs_dir = "Outputs"
448
  if not os.path.exists(outputs_dir) or not os.path.isdir(outputs_dir):
449
- return "Outputs 文件夹不存在或为空"
450
 
451
  converted_files = 0
452
  for root, _, files in os.walk(outputs_dir):
@@ -455,42 +436,42 @@ def compress_wav_to_mp3():
455
  wav_path = os.path.join(root, file)
456
  mp3_path = os.path.join(root, file[:-4] + ".mp3")
457
  try:
458
- # 读取 WAV 文件
459
  audio = AudioSegment.from_wav(wav_path)
460
- # 转换为 MP3,设置比特率为 80kbps
461
  audio.export(mp3_path, format="mp3", bitrate="80k")
462
- # 删除原 WAV 文件
463
  os.remove(wav_path)
464
- print(f"\033[92m✅ 转换并替换: {wav_path} -> {mp3_path}\033[0m")
465
  converted_files += 1
466
  except Exception as e:
467
- print(f"\033[91m❌ 转换 {wav_path} 失败: {str(e)}\033[0m")
468
  continue
469
 
470
  if converted_files == 0:
471
- return "未找到 .wav 文件进行转换"
472
- return f"成功将 {converted_files} .wav 文件转换为 .mp3 并替换"
473
  except Exception as e:
474
- print(f"\033[91m❌ 压缩 WAV MP3 失败: {str(e)}\033[0m")
475
- return f"压缩 WAV MP3 失败: {str(e)}"
476
 
477
 
478
  def generate_all_lines_audio(role, texts, lines, start_index, role_map, spk_emb_seed_id=None, ref_wav=None, role_index=0, button=None, status_output=None, use_sentiment=True):
479
- """批量生成角色语音,基于情感分析动态调整 IndexTTS 参数"""
480
  try:
481
  display_name = role_map.get(role, {'display_name': role})['display_name']
482
 
483
  if cancel_generation.is_set():
484
- return None, f"{display_name}:生成已取消", button
485
 
486
- # 检查是否提供 spk_emb_seed_id ref_wav
487
  if not spk_emb_seed_id and not ref_wav:
488
- print(f"\033[93m⚠️ 角色 {display_name} 未提供 spk_emb_seed_id 或参考音频,跳过生成\033[0m")
489
- return None, f"{display_name}:未提供 spk_emb_seed_id 或参考音频,跳过生成", gr.Button(interactive=True)
490
 
491
  current_model = get_or_load_model()
492
  if current_model[0] is None:
493
- raise RuntimeError(f"ChatterboxTTS 模型未加载:{current_model[1]}")
494
 
495
  params_list = pre_analyze_lines(texts[start_index-1:])
496
  saved_paths = []
@@ -499,7 +480,7 @@ def generate_all_lines_audio(role, texts, lines, start_index, role_map, spk_emb_
499
  role_pattern = re.compile(r'^#\s*(?:' + re.escape(role) + r'\s*)?\"(.*?)\"')
500
  translate_pattern = re.compile(r'^translate\s+\w+\s+([\w_-]+)\s*:')
501
 
502
- print(f"\033[94m🚀 开始生成角色 {display_name} 的语音,从第 {start_index} 条开始,共 {len(texts[start_index-1:])} \033[0m")
503
 
504
  def generate_single_line(idx, line, params):
505
  if cancel_generation.is_set():
@@ -527,16 +508,16 @@ def generate_all_lines_audio(role, texts, lines, start_index, role_map, spk_emb_
527
  identifier = translate_match.group(1)
528
  break
529
  if identifier is None:
530
- print(f"\033[93m跳过: 未找到 translate id for {line} [角色: {display_name}, 索引: {idx+1}]\033[0m")
531
  return None
532
 
533
  processed_line = line
534
- # print(f"\033[92m📢 生成 {display_name} {idx+1}/{len(texts)} 条语音:{processed_line[:30]}...\033[0m")
535
 
536
  filename = f"{identifier}.wav"
537
  output_path = os.path.join("Outputs", filename)
538
 
539
- # 默认参数
540
  kwargs = {
541
  "do_sample": True,
542
  "top_p": 0.8,
@@ -548,39 +529,39 @@ def generate_all_lines_audio(role, texts, lines, start_index, role_map, spk_emb_
548
  "max_mel_tokens": 600
549
  }
550
 
551
- # 如果启用情感分析,覆盖参数
552
  if use_sentiment:
553
  kwargs.update(params)
554
 
555
- # 调用 TTS 推理
556
  output = tts.infer(ref_wav, processed_line, output_path, verbose=cmd_args.verbose,
557
  max_text_tokens_per_sentence=120, **kwargs)
558
 
559
- print(f"\033[91m-------\n温度: {kwargs['temperature']:.2f}, Top_p: {kwargs['top_p']:.2f}, Top_k: {kwargs['top_k']}, "
560
  f"RP: {kwargs['repetition_penalty']:.2f}, Length_penalty: {kwargs['length_penalty']:.2f}, "
561
- f"Num_beams: {kwargs['num_beams']} {line}\n[角色: {display_name}: {idx+1}/{len(texts)}]{line}\n----------\033[0m")
562
  return output_path
563
 
564
  for idx, (line, params) in enumerate(zip(texts[start_index-1:], params_list), start=start_index-1):
565
  if cancel_generation.is_set():
566
- print(f"\033[93m⚠️ {display_name}:生成被取消,停止于第 {idx+1}\033[0m")
567
- return None, f"{display_name}:生成已取消", button
568
  result = generate_single_line(idx, line, params)
569
  if result:
570
  saved_paths.append(result)
571
 
572
- print(f"\033[94m✅ {display_name}:生成完成,共生成 {len(saved_paths)} 条语音\033[0m")
573
- return None, f"{display_name}:从第 {start_index} 条开始,共生成 {len(saved_paths)} 条语音", gr.Button(interactive=True)
574
  except Exception as e:
575
- print(f"\033[91m❌ {display_name}:批量生成失败: {str(e)}\033[0m")
576
- return None, f"{display_name}:批量生成失败: {str(e)}", gr.Button(interactive=True)
577
  finally:
578
  queued_roles.discard(role_index)
579
  is_generating.clear()
580
  process_queue()
581
 
582
  def process_queue():
583
- """处理队列中的下一个角色"""
584
  global generation_queue, is_generating, queued_roles
585
  while generation_queue and not is_generating.is_set() and not cancel_generation.is_set():
586
  role_config = generation_queue.pop(0)
@@ -588,36 +569,36 @@ def process_queue():
588
  role_data = role_texts[role_index]
589
  if not role_data:
590
  continue
591
- role = role_data.get("role", f"角色{role_index+1}")
592
  display_name = role_data.get("display_name", role)
593
  txt = role_config['txt']
594
  start_idx = role_config['start_idx']
595
  spk_emb_seed_id = role_config['spk_emb_seed_id']
596
  ref_wav = role_config['ref_wav']
597
 
598
- # 检查是否提供 spk_emb_seed_id ref_wav
599
  if not spk_emb_seed_id and not ref_wav:
600
- print(f"\033[93m⚠️ 角色 {display_name} 未提供 spk_emb_seed_id 或参考音频,跳过生成\033[0m")
601
- role_components[role_index][4].value = f"{display_name}:未提供 spk_emb_seed_id 或参考音频,跳过生成"
602
  role_components[role_index][-1].value = gr.Button(interactive=True)
603
  queued_roles.discard(role_index)
604
  continue
605
 
606
- print(f"\033[94m🚀 从队列中取出角色 {display_name} 开始生成\033[0m")
607
- # 设置 UI 状态
608
- role_components[role_index][4].value = f"{display_name}:开始生成"
609
  role_components[role_index][0].value = txt
610
  role_components[role_index][1].value = start_idx
611
  role_components[role_index][2].value = spk_emb_seed_id
612
  role_components[role_index][3].value = ref_wav
613
  is_generating.set()
614
- # 使用缓存的配置调用 generate_all_lines_audio
615
  file_lines = []
616
  if last_uploaded_file:
617
  try:
618
  _, file_lines = parse_roles_from_rpy_file(last_uploaded_file, "")
619
  except Exception as e:
620
- role_components[role_index][4].value = f"文件解析失败: {str(e)}"
621
  role_components[role_index][-1].value = gr.Button(interactive=True)
622
  queued_roles.discard(role_index)
623
  is_generating.clear()
@@ -628,23 +609,23 @@ def process_queue():
628
  spk_emb_seed_id, ref_wav, role_index=role_index,
629
  button=role_components[role_index][-1], status_output=role_components[role_index][4]
630
  )
631
- # 更新 UI 组件
632
  role_components[role_index][4].value = result[1]
633
  role_components[role_index][-1].value = result[2]
634
  break
635
 
636
  role_texts = [{} for _ in range(40)]
637
- role_components = [] # 全局存储角色组件
638
- role_mapping_input = None # 全局存储 role_mapping_input
639
 
640
  def process_file(file, mapping_text, filter_text):
641
  global last_uploaded_file
642
  last_uploaded_file = file
643
  if file is None:
644
- return ["", 1, "", None, "", f"### 角色{i+1}", False, True] * 40 + ["请上传文件"]
645
 
646
  try:
647
- mapping_file_path = os.path.join(os.path.dirname(__file__), "角色映射.txt")
648
  with open(mapping_file_path, "w", encoding="utf-8") as f:
649
  f.write(mapping_text.strip() if mapping_text else "")
650
 
@@ -652,70 +633,70 @@ def process_file(file, mapping_text, filter_text):
652
  role_dict, lines = parse_roles_from_rpy_file(file, filter_text)
653
 
654
  filtered_role_dict = {}
655
- for role, lines in role_dict.items():
656
- if len(lines) >= 100:
657
- filtered_role_dict[role] = lines
658
  else:
659
- print(f"角色 {role} 的对话条数 ({len(lines)}) 小于100,已被过滤")
660
 
661
- # display_name 字母顺序排序
662
- role_items = [(role, lines) for role, lines in filtered_role_dict.items()]
663
  role_items.sort(key=lambda x: role_map.get(x[0], {'display_name': x[0]})['display_name'].lower())
664
- # print(f"\033[94m📋 角色按字母顺序排序:{[role_map.get(role, {'display_name': role})['display_name'] for role, _ in role_items]}\033[0m")
665
 
666
  except Exception as e:
667
- return ["", 1, "", None, "", f"### 角色{i+1}", False, True] * 40 + [f"文件解析失败: {str(e)}"]
668
 
669
  return_values = []
670
  for i in range(40):
671
  if i < len(role_items):
672
- role, lines = role_items[i]
673
  display_name = role_map.get(role, {'display_name': role, 'voice_file': None, 'seed_id': None})['display_name']
674
  seed_id = role_map.get(role, {'display_name': role, 'voice_file': None, 'seed_id': None})['seed_id']
675
  voice_file = role_map.get(role, {'display_name': role, 'voice_file': None, 'seed_id': None})['voice_file']
676
  ref_wav_original = None
677
  if voice_file:
678
  ref_wav_original = os.path.join('voice', voice_file)
679
- # 如果 voice_file 未指定或不存在,尝试使用 display_name.wav
680
  if not ref_wav_original or not os.path.exists(ref_wav_original):
681
  ref_wav_original = os.path.join('voice', display_name + '.wav')
682
  if ref_wav_original and os.path.exists(ref_wav_original):
683
- # 缓存到临时目录
684
  cached_path = os.path.join(audio_cache_dir, f"role_{i}_{display_name}.wav")
685
  shutil.copy(ref_wav_original, cached_path)
686
  ref_wav_path = cached_path
687
- print(f"\033[92m✅ 加载并缓存参考音频 for {display_name}: {ref_wav_original} -> {cached_path}\033[0m")
688
  else:
689
  ref_wav_path = None
690
- # print(f"\033[93m⚠️ 未找到参考音频 for {display_name}: {ref_wav_original}\033[0m")
691
- role_texts[i] = {"role": role, "display_name": display_name, "lines": lines, "ref_wav": ref_wav_path, "seed_id": seed_id}
692
- joined = "\n".join(lines)
693
- return_values.extend([joined, 1, seed_id, ref_wav_path, f"{display_name}:共 {len(lines)} ", f"### {display_name}", True, True])
694
  else:
695
  role_texts[i] = {}
696
- return_values.extend(["", 1, "", None, "", f"### 角色{i+1}", False, True])
697
 
698
- return_values.append("文件处理成功,请为每个角色点击批量生成")
699
  return return_values
700
 
701
  def stop_all_generation():
702
- """停止所有正在进行的生成任务,并释放所有按钮"""
703
  global generation_queue, is_generating, queued_roles
704
  cancel_generation.set()
705
  generation_queue.clear()
706
  queued_roles.clear()
707
  is_generating.clear()
708
- print(f"\033[93m🛑 停止所有生成任务\033[0m")
709
  button_states = [gr.Button(interactive=True) for _ in range(40)]
710
- status_outputs = [f"{role_texts[i].get('display_name', f'角色{i+1}')}:生成已取消" if role_texts[i] else "" for i in range(40)]
711
- # 确保所有按钮恢复为可交互状态
712
  for i in range(40):
713
  if role_components[i][-1]: # bulk_btn
714
  role_components[i][-1].value = gr.Button(interactive=True)
715
- return "正在停止所有生成任务...", status_outputs, button_states
716
 
717
  def get_pt_file_for_download(seed_id):
718
- """为下载按钮生成 .pt 文件并返回路径和状态"""
719
  spk_emb, message = get_pt_file(seed_id)
720
  if spk_emb is not None:
721
  os.makedirs("tmp", exist_ok=True)
@@ -725,66 +706,70 @@ def get_pt_file_for_download(seed_id):
725
  return gr.DownloadButton(value=None, label="Download .pt File", visible=False), message
726
 
727
  def get_download_zip_state():
728
- """检查 zip 文件状态并返回下载按钮状态"""
729
  if os.path.exists(os.path.join("tmp", "outputs.zip")):
730
- return gr.DownloadButton(value=os.path.join("tmp", "outputs.zip"), label="Download Outputs.zip", visible=True), "已准备好下载 outputs.zip"
731
- return gr.DownloadButton(value=None, label="Download Outputs.zip", visible=False), "未找到 outputs.zip"
732
 
733
  def main():
734
- # 在脚本启动时清空缓存文件夹
735
  # clear_audio_cache_dir()
736
 
737
  global role_components, make_bulk_functions, audio_cache_dir, role_mapping_input
738
  MAX_ROLES = 40
739
- mapping_file_path = os.path.join(os.path.dirname(__file__), "角色映射.txt")
740
  try:
741
  with open(mapping_file_path, "r", encoding="utf-8") as f:
742
  role_mapping_placeholder = f.read().strip()
743
  except FileNotFoundError:
744
- role_mapping_placeholder = "角色代号: jud → 显示名: Judge voice:jud seed_id:1403\n角色代号: jury → 显示名: Members of The Jury voice:jury seed_id:1404\n角色代号: noname → 显示名: 无名角色"
 
 
 
 
745
 
746
  with gr.Blocks() as demo:
747
  gr.Markdown("https://huggingface.co/spaces/ResembleAI/Chatterbox")
748
 
749
- model_status = gr.Textbox(label="模型加载状态", interactive=False, value="正在加载模型...")
750
 
751
  with gr.Row():
752
  default_rpy_path = os.path.join(os.path.dirname(__file__), "dialogue.rpy")
753
  file_input = gr.File(
754
- label="上传 .rpy 文件",
755
  type="binary",
756
  value=default_rpy_path if os.path.exists(default_rpy_path) else None
757
  )
758
  with gr.Column():
759
  role_mapping_input = gr.Textbox(
760
- label="角色映射(格式:角色代号: xxx → 显示名: xxx voice:xxx seed_id:xxx",
761
  lines=5,
762
  placeholder=role_mapping_placeholder,
763
  value=role_mapping_placeholder
764
  )
765
  filter_input = gr.Textbox(
766
- label="对话筛选(格式:min_len=X;keyword=Y,留空显示全部)",
767
  lines=2,
768
- placeholder="示例:min_len=20;keyword=hello\n(最小长度20字符,包含'hello'的对话)",
769
  value="min_len=3"
770
  )
771
  with gr.Row():
772
- process_button = gr.Button("处理文件")
773
- stop_button = gr.Button("停止所有生成")
774
- zip_button = gr.Button("打包 Outputs 文件夹")
775
- compress_button = gr.Button("压缩 WAV MP3")
776
- all_status = gr.Textbox(label="总状态", interactive=False, visible=False)
777
  download_zip_button = gr.DownloadButton(label="Download Outputs.zip", visible=False)
778
- zip_status = gr.Textbox(label="打包状态", interactive=False)
779
 
780
  with gr.Row():
781
  seed_id_input = gr.Textbox(
782
- label="输入 seed_id 获取 .pt 文件",
783
- placeholder="例如: 1403 seed_1403",
784
  visible=False
785
  )
786
  pt_download_button = gr.DownloadButton(label="Download .pt File", visible=False)
787
- pt_status = gr.Textbox(label="生成 .pt 文件状态", interactive=False)
788
 
789
  def load_model_on_start():
790
  model, status = get_or_load_model()
@@ -829,9 +814,9 @@ def main():
829
  with gr.Row():
830
  for i in range(row_idx * ROLES_PER_ROW, min((row_idx + 1) * ROLES_PER_ROW, MAX_ROLES)):
831
  with gr.Group(visible=False, elem_classes="compact-group") as group:
832
- role_display = gr.Markdown(f"### 角色{i+1}", elem_classes="compact-header")
833
  text_input = gr.Textbox(
834
- label="文本",
835
  lines=2,
836
  max_lines=4,
837
  elem_classes="compact-textbox",
@@ -839,36 +824,36 @@ def main():
839
  )
840
  start_index = gr.Number(
841
  value=1,
842
- label="从第几条开始",
843
  minimum=1,
844
  step=1,
845
  elem_classes="compact-number"
846
  )
847
  spk_emb_seed_id = gr.Textbox(
848
- label="输入 spk_emb seed_id",
849
- placeholder="例如: 1403 seed_1403",
850
  elem_classes="compact-textbox",
851
  visible=False
852
  )
853
  ref_wav = gr.Audio(
854
  type="filepath",
855
- label="参考音频文件(可选,建议 6 秒以上)",
856
  sources=["upload", "microphone"],
857
  elem_classes="compact-audio"
858
  )
859
  audio_output = gr.Audio(
860
- label="输出音频",
861
  elem_classes="compact-audio",
862
  visible=False
863
  )
864
  status = gr.Textbox(
865
- label="状态",
866
  interactive=False,
867
  elem_classes="compact-textbox",
868
  container=False
869
  )
870
  bulk_btn = gr.Button(
871
- f"生成角色{i+1}",
872
  size="sm",
873
  elem_classes="compact-button",
874
  interactive=True
@@ -881,32 +866,32 @@ def main():
881
  global generation_queue, is_generating, queued_roles
882
  role_data = role_texts[i]
883
  if not role_data:
884
- return None, f"角色{i+1} 无数据", gr.Button(interactive=True)
885
- role = role_data.get("role", f"角色{i+1}")
886
  display_name = role_data.get("display_name", role)
887
 
888
- # 优先使用 UI 提供的 ref_wav,但检查是否需要缓存
889
  ref_wav_path = ref_wav
890
  if ref_wav and os.path.exists(ref_wav):
891
- # 检查 ref_wav 是否已经是缓存目录中的文件
892
  if not ref_wav.startswith(audio_cache_dir):
893
  cached_path = os.path.join(audio_cache_dir, f"role_{i}_{display_name}_ui.wav")
894
  shutil.copy(ref_wav, cached_path)
895
  role_texts[i]["ref_wav"] = cached_path
896
- print(f"\033[92m✅ 缓存 UI 提供的参考音频 for {display_name}: {ref_wav} -> {cached_path}\033[0m")
897
  else:
898
- # 已经是缓存文件,直接使用
899
  role_texts[i]["ref_wav"] = ref_wav
900
- print(f"\033[92m✅ 使用已缓存的参考音频 for {display_name}: {ref_wav}\033[0m")
901
  else:
902
  ref_wav_path = role_texts[i].get("ref_wav")
903
 
904
- # 检查是否提供 spk_emb_seed_id ref_wav
905
  if not spk_emb_seed_id and not ref_wav_path:
906
- print(f"\033[93m⚠️ 角色 {display_name} 未提供 spk_emb_seed_id 或参考音频,跳过生成\033[0m")
907
- return None, f"{display_name}:未提供 spk_emb_seed_id 或参考音频,跳过生成", gr.Button(interactive=True)
908
 
909
- # 缓存角色配置到队列
910
  role_config = {
911
  'role_index': i,
912
  'txt': txt,
@@ -920,19 +905,19 @@ def main():
920
  generation_queue.append(role_config)
921
  queued_roles.add(i)
922
  pos = len(generation_queue)
923
- print(f"\033[94m⏳ 角色 {display_name} 已加入队列,位置 {pos}/{pos}\033[0m")
924
- return None, f"{display_name}:等待队列中,位置 {pos}/{pos}", gr.Button(interactive=False)
925
  else:
926
  is_generating.set()
927
  queued_roles.add(i)
928
  cancel_generation.clear()
929
- print(f"\033[94m🚀 单角色生成:{display_name},从第 {start_idx} 条开始\033[0m")
930
  file_lines = []
931
  if last_uploaded_file:
932
  try:
933
  _, file_lines = parse_roles_from_rpy_file(last_uploaded_file, "")
934
  except Exception as e:
935
- return None, f"文件解析失败: {str(e)}", gr.Button(interactive=True)
936
  role_map = parse_role_mappings(role_mapping_input.value)
937
  result = generate_all_lines_audio(
938
  role, role_data.get("lines", []), file_lines, int(start_idx), role_map,
@@ -941,7 +926,7 @@ def main():
941
  return result
942
  return inner
943
 
944
- # 存储 make_bulk 函数,传递 bulk_btn
945
  make_bulk_functions[i] = make_bulk(i, bulk_btn)
946
 
947
  bulk_btn.click(
 
1
  # ```python
2
  #!/usr/bin/env python3
3
  """
4
+ ChatTTS Gradio Application - Adapted to use ChatterboxTTS
5
+ Adapted for Hugging Face Spaces runtime environment
6
+ Based on Chatterbox-TTS implementation: https://huggingface.co/spaces/ResembleAI/Chatterbox
7
  """
8
 
9
  import os
 
42
  warnings.filterwarnings("ignore", category=UserWarning)
43
 
44
  import argparse
45
+ # ----------------- Command Line Argument Parsing -----------------
46
  parser = argparse.ArgumentParser(description="IndexTTS WebUI")
47
+ parser.add_argument("--verbose", action="store_true", default=False, help="Enable verbose mode")
48
+ parser.add_argument("--port", type=int, default=7860, help="Port to run the web UI on")
49
+ parser.add_argument("--host", type=str, default="127.0.0.1", help="Host to run the web UI on")
50
+ parser.add_argument("--model_dir", type=str, default="checkpoints", help="Model checkpoints directory")
51
  cmd_args = parser.parse_args()
52
  model_dir="checkpoints"
53
 
54
+ # ----------------- Set Module Search Path -----------------
55
  current_dir = os.path.dirname(os.path.abspath(__file__))
56
  sys.path.append(current_dir)
57
  sys.path.append(os.path.join(current_dir, "indextts"))
58
 
59
+ # ----------------- Download Model -----------------
60
  MODE = 'local'
61
+ snapshot_download("IndexTeam/IndexTTS-1.5", local_dir="checkpoints") # Download model from Hugging Face to local
62
 
63
+ # ----------------- Check Model File Integrity -----------------
64
  if not os.path.exists(cmd_args.model_dir):
65
  print(f"Model directory {cmd_args.model_dir} does not exist. Please download the model first.")
66
  sys.exit(1)
 
76
  print(f"Required file {file_path} does not exist. Please download it.")
77
  sys.exit(1)
78
  def analyze_sentiment(text):
79
+ """Use TextBlob to analyze sentiment and return optimized IndexTTS parameter dictionary"""
80
  try:
81
  blob = TextBlob(text)
82
+ polarity = blob.sentiment.polarity # -1 (negative) ~ 1 (positive)
83
+ subjectivity = blob.sentiment.subjectivity # 0 (objective) ~ 1 (subjective)
84
 
85
+ # temperature: higher for positive (lively), lower for negative (plain); range 0.5~1.5
86
  temperature = 1.0 + math.tanh(polarity) * 0.9 # 0.5 ~ 1.5
87
+ # top_p: higher for subjective (diverse), lower for objective (conservative); range 0.7~0.95
88
  top_p = 0.7 + subjectivity * 0.25
89
+ # top_k: higher for subjective (creative), but capped; range 20~50
90
  top_k = int(20 + subjectivity * 30)
91
+ # repetition_penalty: higher for objective (avoid repetition), lower for subjective (allow emphasis); range 5.0~15.0
92
  repetition_penalty = 10.0 + (1 - subjectivity) * 5.0 - polarity * 2.0
93
+ # length_penalty: positive for positive (longer), negative for negative (shorter); range -1.0~1.0
94
  length_penalty = math.tanh(polarity) * 1.0
95
+ # num_beams: higher for positive (natural), but capped for speed; range 2~5
96
  num_beams = int(2 + (polarity + 1) * 1.5)
97
 
98
  return {
 
104
  "num_beams": max(2, min(5, num_beams))
105
  }
106
  except Exception as e:
107
+ print(f"Sentiment analysis failed: {str(e)}")
108
+ return { # Default values
109
  "temperature": 1.0,
110
  "top_p": 0.8,
111
  "top_k": 30,
 
115
  }
116
 
117
 
118
+ # ----------------- Import Gradio and Other Modules -----------------
119
  import gradio as gr
120
  import pandas as pd
121
 
122
+ from indextts.infer import IndexTTS # Core TTS inference class
123
+ from tools.i18n.i18n import I18nAuto # Internationalization tool
124
 
125
+ # ----------------- Initialize TTS Model -----------------
126
+ i18n = I18nAuto(language="en") # Set default language to English
127
+ tts = IndexTTS(model_dir=cmd_args.model_dir, cfg_path=os.path.join(cmd_args.model_dir, "config.yaml")) # Load model
128
 
129
+ # # ----------------- Create Output Directory -----------------
130
+ # # os.makedirs("outputs/tasks", exist_ok=True)
131
+ # # os.makedirs("prompts", exist_ok=True)
132
 
133
+ # ----------------- Core Functions -----------------
134
 
135
  def ensure_wav(file_path):
136
  """
137
+ Ensure input audio is in WAV format.
138
+ If not WAV, convert using ffmpeg.
139
+ Return WAV file path.
140
  """
141
  if not file_path.lower().endswith(".wav"):
142
  wav_path = file_path.rsplit(".", 1)[0] + ".wav"
 
146
 
147
  # def progress_print(step, total, info=""):
148
  # """
149
+ # Print audio generation progress to terminal
150
+ # step: current step
151
+ # total: total number of steps
152
+ # info: extra info
153
  # """
154
  # percent = int(step / total * 100)
155
  # print(f"\r[{percent}%] {info}", end="", flush=True)
156
 
157
 
158
+ # Set up logging
159
  logging.basicConfig(level=logging.INFO)
160
  logger = logging.getLogger(__name__)
161
 
162
+ # Global variable to store last uploaded file
163
  last_uploaded_file = None
164
 
165
+ # Global cancel flag
166
  cancel_generation = threading.Event()
167
 
168
+ # Global generation state and queue
169
  is_generating = threading.Event()
170
+ generation_queue = [] # Stores {'role_index': i, 'txt': str, 'start_idx': int, 'spk_emb_seed_id': str, 'ref_wav': str}
171
  queued_roles = set()
172
 
173
+ # Store make_bulk function for each role
174
  make_bulk_functions = [None] * 40
175
 
176
+ # Temporary directory for audio cache
177
  audio_cache_dir = tempfile.mkdtemp(prefix='audio_cache_')
178
 
179
+ # Clear cache folder at script start
180
  def clear_audio_cache_dir():
181
+ """Clear or recreate audio_cache_dir"""
182
  global audio_cache_dir
183
  try:
184
  if os.path.exists(audio_cache_dir):
185
  shutil.rmtree(audio_cache_dir)
186
+ print(f"\033[92m✅ Cleared cache folder: {audio_cache_dir}\033[0m")
187
  os.makedirs(audio_cache_dir)
188
+ print(f"\033[92m✅ Created new cache folder: {audio_cache_dir}\033[0m")
189
  except Exception as e:
190
+ print(f"\033[91m❌ Failed to clear cache folder: {str(e)}\033[0m")
191
 
192
+ # Monkey patch torch.load to handle device mapping
193
  original_torch_load = torch.load
194
 
195
  def patched_torch_load(f, map_location=None, **kwargs):
196
  """
197
+ Patched torch.load, automatically maps CUDA tensors to CPU/CUDA
198
  """
199
  if map_location is None:
200
  map_location = 'cpu'
201
+ logger.info(f"🔧 Loading with map_location={map_location}")
202
  return original_torch_load(f, map_location=map_location, **kwargs)
203
 
204
+ # Apply patch immediately after torch import
205
  torch.load = patched_torch_load
206
  if 'torch' in sys.modules:
207
  sys.modules['torch'].load = patched_torch_load
208
 
209
+ logger.info("✅ Successfully applied torch.load device mapping patch")
210
 
211
+ # Device detection, adapted for Hugging Face Spaces
212
  if torch.cuda.is_available():
213
  DEVICE = "cuda"
214
+ logger.info("🚀 Running with CUDA GPU")
215
  else:
216
  DEVICE = "cpu"
217
+ logger.info("🚀 Running with CPU")
218
 
219
+ print(f"🚀 Running on device: {DEVICE}")
220
 
221
+ # Global model variable
222
  MODEL = True
223
 
224
  def get_or_load_model():
225
+ """Load ChatterboxTTS model (if not already loaded), and ensure it runs on the correct device"""
226
  global MODEL, DEVICE
227
  if MODEL is None:
228
+ print("Model not loaded, initializing...")
229
  try:
230
+ # Try official import path first
231
  try:
232
  from chatterbox.src.chatterbox.tts import ChatterboxTTS
233
+ logger.info("✅ Using official chatterbox.src import path")
234
  except ImportError:
235
+ # Fallback to alternative import path
236
  from chatterbox import ChatterboxTTS
237
+ logger.info("✅ Using chatterbox direct import path")
238
 
239
+ # Load model to CPU first to avoid device issues
240
  MODEL = ChatterboxTTS.from_pretrained("cpu")
241
 
242
+ # If CUDA, move to target device
243
  if DEVICE == "cuda":
244
+ logger.info(f"Moving model components to {DEVICE}...")
245
  try:
246
  if hasattr(MODEL, 't3'):
247
  MODEL.t3 = MODEL.t3.to(DEVICE)
 
251
  MODEL.ve = MODEL.ve.to(DEVICE)
252
 
253
  MODEL.device = DEVICE
254
+ logger.info(f"✅ All model components moved to {DEVICE}")
255
  except Exception as e:
256
+ logger.warning(f"⚠️ Unable to move some components to {DEVICE}: {e}")
257
+ logger.info("🔄 Falling back to CPU mode for stability")
258
  DEVICE = "cpu"
259
  MODEL.device = "cpu"
260
 
261
+ logger.info(f"✅ Model loaded on {DEVICE}")
262
+ return MODEL, "Model loaded successfully"
263
 
264
  except Exception as e:
265
+ logger.error(f"❌ Failed to load model: {e}")
266
+ return None, f"Model loading failed: {str(e)}"
267
+ return MODEL, "Model already loaded"
268
 
269
  def parse_role_mappings(mapping_text):
270
+ """Parse role mapping text and return {role_code: {'display_name': display_name, 'voice_file': 'xxx.wav', 'seed_id': seed_id}} dict"""
271
  role_map = {}
272
  if not mapping_text:
273
  return role_map
274
  lines = mapping_text.strip().splitlines()
275
+ pattern = re.compile(r'Role Code:\s*(\w+)\s*→\s*Display Name:\s*([^→]+?)(?:\s+voice:\s*([\w_-]+))?(?:\s+seed_id:\s*(\w+))?\s*$')
276
 
277
  for line in lines:
278
  match = pattern.match(line.strip())
 
291
  return role_map
292
 
293
  def filter_lines(lines, filter_text):
294
+ """Filter dialogue lines according to filter conditions"""
295
  if not filter_text:
296
  return lines
297
  filtered = []
 
308
  if len(re.sub(r'[^A-Za-z0-9]', '', line)) < min_len:
309
  include = False
310
  except ValueError:
311
+ print(f"Warning: invalid min_len condition: {condition}")
312
  elif condition.startswith('keyword='):
313
  keyword = condition.split('=')[1].lower()
314
  if keyword not in line.lower():
315
  include = False
316
  else:
317
+ print(f"Warning: unknown filter condition: {condition}")
318
  if include:
319
  filtered.append(line)
320
  return filtered
 
348
 
349
  return role_dict, lines
350
  except Exception as e:
351
+ raise ValueError(f"Failed to parse .rpy file: {str(e)}")
352
 
353
  def save_wav(audio_data, sample_rate, output_path):
354
  try:
 
358
  audio_int16 = audio_data
359
  write(output_path, sample_rate, audio_int16)
360
  except Exception as e:
361
+ raise RuntimeError(f"Failed to save audio file: {str(e)}")
362
 
363
  # def analyze_sentiment(text):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
364
  def pre_analyze_lines(texts):
365
+ """Analyze sentiment for each text line and return a list of parameter dicts"""
366
  context_window = 2
367
  params_list = []
368
  for idx, line in enumerate(texts):
 
384
  return params_list
385
 
386
  def get_pt_file(seed_id, csv_path=os.path.join(os.path.dirname(__file__), "evaluation_results.csv")):
387
+ """Get .pt data from CSV file by seed_id and load as PyTorch tensor"""
388
  try:
389
  if seed_id and not seed_id.startswith("seed_"):
390
  seed_id = f"seed_{seed_id}"
 
392
  df = pd.read_csv(csv_path, encoding="utf-8")
393
  row = df[df["seed_id"] == seed_id]
394
  if row.empty:
395
+ return None, f"seed_id not found: {seed_id}"
396
 
397
  emb_data = row.iloc[0]["emb_data"]
398
  emb_bytes = base64.b64decode(emb_data)
399
  emb_buffer = io.BytesIO(emb_bytes)
400
  spk_emb = torch.load(emb_buffer)
401
 
402
+ return spk_emb, f"Successfully loaded spk_emb data for seed_id: {seed_id}"
403
  except Exception as e:
404
+ return None, f"Failed to load spk_emb data: {str(e)}"
405
 
406
  def zip_outputs_folder():
407
  try:
 
416
  file_path = os.path.join(root, file)
417
  arcname = os.path.relpath(file_path, outputs_dir)
418
  zipf.write(file_path, os.path.join("Outputs", arcname))
419
+ return zip_path, "Outputs folder successfully zipped as outputs.zip"
420
  else:
421
+ return None, "Outputs folder does not exist or is empty"
422
  except Exception as e:
423
+ return None, f"Failed to zip Outputs folder: {str(e)}"
424
 
425
  def compress_wav_to_mp3():
426
+ """Convert .wav files in Outputs folder and subfolders to 80kbps .mp3 files, replacing the originals"""
427
  try:
428
  outputs_dir = "Outputs"
429
  if not os.path.exists(outputs_dir) or not os.path.isdir(outputs_dir):
430
+ return "Outputs folder does not exist or is empty"
431
 
432
  converted_files = 0
433
  for root, _, files in os.walk(outputs_dir):
 
436
  wav_path = os.path.join(root, file)
437
  mp3_path = os.path.join(root, file[:-4] + ".mp3")
438
  try:
439
+ # Read WAV file
440
  audio = AudioSegment.from_wav(wav_path)
441
+ # Convert to MP3, set bitrate to 80kbps
442
  audio.export(mp3_path, format="mp3", bitrate="80k")
443
+ # Delete original WAV file
444
  os.remove(wav_path)
445
+ print(f"\033[92m✅ Converted and replaced: {wav_path} -> {mp3_path}\033[0m")
446
  converted_files += 1
447
  except Exception as e:
448
+ print(f"\033[91m❌ Failed to convert {wav_path}: {str(e)}\033[0m")
449
  continue
450
 
451
  if converted_files == 0:
452
+ return "No .wav files found to convert"
453
+ return f"Successfully converted and replaced {converted_files} .wav files to .mp3"
454
  except Exception as e:
455
+ print(f"\033[91m❌ Failed to compress WAV to MP3: {str(e)}\033[0m")
456
+ return f"Failed to compress WAV to MP3: {str(e)}"
457
 
458
 
459
  def generate_all_lines_audio(role, texts, lines, start_index, role_map, spk_emb_seed_id=None, ref_wav=None, role_index=0, button=None, status_output=None, use_sentiment=True):
460
+ """Batch generate role audio, dynamically adjust IndexTTS parameters based on sentiment analysis"""
461
  try:
462
  display_name = role_map.get(role, {'display_name': role})['display_name']
463
 
464
  if cancel_generation.is_set():
465
+ return None, f"{display_name}: Generation cancelled", button
466
 
467
+ # Check if spk_emb_seed_id or ref_wav is provided
468
  if not spk_emb_seed_id and not ref_wav:
469
+ print(f"\033[93m⚠️ Role {display_name} did not provide spk_emb_seed_id or reference audio, skipping generation\033[0m")
470
+ return None, f"{display_name}: No spk_emb_seed_id or reference audio provided, skipping generation", gr.Button(interactive=True)
471
 
472
  current_model = get_or_load_model()
473
  if current_model[0] is None:
474
+ raise RuntimeError(f"ChatterboxTTS model not loaded: {current_model[1]}")
475
 
476
  params_list = pre_analyze_lines(texts[start_index-1:])
477
  saved_paths = []
 
480
  role_pattern = re.compile(r'^#\s*(?:' + re.escape(role) + r'\s*)?\"(.*?)\"')
481
  translate_pattern = re.compile(r'^translate\s+\w+\s+([\w_-]+)\s*:')
482
 
483
+ print(f"\033[94m🚀 Start generating audio for role {display_name}, from line {start_index}, total {len(texts[start_index-1:])} lines\033[0m")
484
 
485
  def generate_single_line(idx, line, params):
486
  if cancel_generation.is_set():
 
508
  identifier = translate_match.group(1)
509
  break
510
  if identifier is None:
511
+ print(f"\033[93mSkip: No translate id found for {line} [Role: {display_name}, Index: {idx+1}]\033[0m")
512
  return None
513
 
514
  processed_line = line
515
+ # print(f"\033[92m📢 Generate {display_name} line {idx+1}/{len(texts)}: {processed_line[:30]}...\033[0m")
516
 
517
  filename = f"{identifier}.wav"
518
  output_path = os.path.join("Outputs", filename)
519
 
520
+ # Default parameters
521
  kwargs = {
522
  "do_sample": True,
523
  "top_p": 0.8,
 
529
  "max_mel_tokens": 600
530
  }
531
 
532
+ # If sentiment analysis enabled, override parameters
533
  if use_sentiment:
534
  kwargs.update(params)
535
 
536
+ # Call TTS inference
537
  output = tts.infer(ref_wav, processed_line, output_path, verbose=cmd_args.verbose,
538
  max_text_tokens_per_sentence=120, **kwargs)
539
 
540
+ print(f"\033[91m-------\nTemperature: {kwargs['temperature']:.2f}, Top_p: {kwargs['top_p']:.2f}, Top_k: {kwargs['top_k']}, "
541
  f"RP: {kwargs['repetition_penalty']:.2f}, Length_penalty: {kwargs['length_penalty']:.2f}, "
542
+ f"Num_beams: {kwargs['num_beams']} {line}\n[Role: {display_name}: {idx+1}/{len(texts)}]{line}\n----------\033[0m")
543
  return output_path
544
 
545
  for idx, (line, params) in enumerate(zip(texts[start_index-1:], params_list), start=start_index-1):
546
  if cancel_generation.is_set():
547
+ print(f"\033[93m⚠️ {display_name}: Generation cancelled, stopped at line {idx+1}\033[0m")
548
+ return None, f"{display_name}: Generation cancelled", button
549
  result = generate_single_line(idx, line, params)
550
  if result:
551
  saved_paths.append(result)
552
 
553
+ print(f"\033[94m✅ {display_name}: Generation finished, {len(saved_paths)} audio lines generated\033[0m")
554
+ return None, f"{display_name}: From line {start_index}, {len(saved_paths)} audio lines generated", gr.Button(interactive=True)
555
  except Exception as e:
556
+ print(f"\033[91m❌ {display_name}: Batch generation failed: {str(e)}\033[0m")
557
+ return None, f"{display_name}: Batch generation failed: {str(e)}", gr.Button(interactive=True)
558
  finally:
559
  queued_roles.discard(role_index)
560
  is_generating.clear()
561
  process_queue()
562
 
563
  def process_queue():
564
+ """Process the next role in the queue"""
565
  global generation_queue, is_generating, queued_roles
566
  while generation_queue and not is_generating.is_set() and not cancel_generation.is_set():
567
  role_config = generation_queue.pop(0)
 
569
  role_data = role_texts[role_index]
570
  if not role_data:
571
  continue
572
+ role = role_data.get("role", f"Role{role_index+1}")
573
  display_name = role_data.get("display_name", role)
574
  txt = role_config['txt']
575
  start_idx = role_config['start_idx']
576
  spk_emb_seed_id = role_config['spk_emb_seed_id']
577
  ref_wav = role_config['ref_wav']
578
 
579
+ # Check if spk_emb_seed_id or ref_wav is provided
580
  if not spk_emb_seed_id and not ref_wav:
581
+ print(f"\033[93m⚠️ Role {display_name} did not provide spk_emb_seed_id or reference audio, skipping generation\033[0m")
582
+ role_components[role_index][4].value = f"{display_name}: No spk_emb_seed_id or reference audio provided, skipping generation"
583
  role_components[role_index][-1].value = gr.Button(interactive=True)
584
  queued_roles.discard(role_index)
585
  continue
586
 
587
+ print(f"\033[94m🚀 Starting generation for role {display_name} from queue\033[0m")
588
+ # Set UI state
589
+ role_components[role_index][4].value = f"{display_name}: Generation started"
590
  role_components[role_index][0].value = txt
591
  role_components[role_index][1].value = start_idx
592
  role_components[role_index][2].value = spk_emb_seed_id
593
  role_components[role_index][3].value = ref_wav
594
  is_generating.set()
595
+ # Use cached config to call generate_all_lines_audio
596
  file_lines = []
597
  if last_uploaded_file:
598
  try:
599
  _, file_lines = parse_roles_from_rpy_file(last_uploaded_file, "")
600
  except Exception as e:
601
+ role_components[role_index][4].value = f"File parsing failed: {str(e)}"
602
  role_components[role_index][-1].value = gr.Button(interactive=True)
603
  queued_roles.discard(role_index)
604
  is_generating.clear()
 
609
  spk_emb_seed_id, ref_wav, role_index=role_index,
610
  button=role_components[role_index][-1], status_output=role_components[role_index][4]
611
  )
612
+ # Update UI components
613
  role_components[role_index][4].value = result[1]
614
  role_components[role_index][-1].value = result[2]
615
  break
616
 
617
  role_texts = [{} for _ in range(40)]
618
+ role_components = [] # Global storage for role components
619
+ role_mapping_input = None # Global storage for role_mapping_input
620
 
621
  def process_file(file, mapping_text, filter_text):
622
  global last_uploaded_file
623
  last_uploaded_file = file
624
  if file is None:
625
+ return ["", 1, "", None, "", f"### Role {i+1}", False, True] * 40 + ["Please upload a file"]
626
 
627
  try:
628
+ mapping_file_path = os.path.join(os.path.dirname(__file__), "role_mapping.txt")
629
  with open(mapping_file_path, "w", encoding="utf-8") as f:
630
  f.write(mapping_text.strip() if mapping_text else "")
631
 
 
633
  role_dict, lines = parse_roles_from_rpy_file(file, filter_text)
634
 
635
  filtered_role_dict = {}
636
+ for role, lines_ in role_dict.items():
637
+ if len(lines_) >= 100:
638
+ filtered_role_dict[role] = lines_
639
  else:
640
+ print(f"Role {role} dialogue count ({len(lines_)}) is less than 100, filtered out")
641
 
642
+ # Sort by display_name alphabetically
643
+ role_items = [(role, lines_) for role, lines_ in filtered_role_dict.items()]
644
  role_items.sort(key=lambda x: role_map.get(x[0], {'display_name': x[0]})['display_name'].lower())
645
+ # print(f"\033[94m📋 Roles sorted alphabetically: {[role_map.get(role, {'display_name': role})['display_name'] for role, _ in role_items]}\033[0m")
646
 
647
  except Exception as e:
648
+ return ["", 1, "", None, "", f"### Role {i+1}", False, True] * 40 + [f"File parsing failed: {str(e)}"]
649
 
650
  return_values = []
651
  for i in range(40):
652
  if i < len(role_items):
653
+ role, lines_ = role_items[i]
654
  display_name = role_map.get(role, {'display_name': role, 'voice_file': None, 'seed_id': None})['display_name']
655
  seed_id = role_map.get(role, {'display_name': role, 'voice_file': None, 'seed_id': None})['seed_id']
656
  voice_file = role_map.get(role, {'display_name': role, 'voice_file': None, 'seed_id': None})['voice_file']
657
  ref_wav_original = None
658
  if voice_file:
659
  ref_wav_original = os.path.join('voice', voice_file)
660
+ # If voice_file not specified or does not exist, try display_name.wav
661
  if not ref_wav_original or not os.path.exists(ref_wav_original):
662
  ref_wav_original = os.path.join('voice', display_name + '.wav')
663
  if ref_wav_original and os.path.exists(ref_wav_original):
664
+ # Cache to temp directory
665
  cached_path = os.path.join(audio_cache_dir, f"role_{i}_{display_name}.wav")
666
  shutil.copy(ref_wav_original, cached_path)
667
  ref_wav_path = cached_path
668
+ print(f"\033[92m✅ Loaded and cached reference audio for {display_name}: {ref_wav_original} -> {cached_path}\033[0m")
669
  else:
670
  ref_wav_path = None
671
+ # print(f"\033[93m⚠️ Reference audio not found for {display_name}: {ref_wav_original}\033[0m")
672
+ role_texts[i] = {"role": role, "display_name": display_name, "lines": lines_, "ref_wav": ref_wav_path, "seed_id": seed_id}
673
+ joined = "\n".join(lines_)
674
+ return_values.extend([joined, 1, seed_id, ref_wav_path, f"{display_name}: {len(lines_)} lines in total", f"### {display_name}", True, True])
675
  else:
676
  role_texts[i] = {}
677
+ return_values.extend(["", 1, "", None, "", f"### Role {i+1}", False, True])
678
 
679
+ return_values.append("File processed successfully, please click batch generate for each role")
680
  return return_values
681
 
682
  def stop_all_generation():
683
+ """Stop all ongoing generation tasks and release all buttons"""
684
  global generation_queue, is_generating, queued_roles
685
  cancel_generation.set()
686
  generation_queue.clear()
687
  queued_roles.clear()
688
  is_generating.clear()
689
+ print(f"\033[93m🛑 Stopped all generation tasks\033[0m")
690
  button_states = [gr.Button(interactive=True) for _ in range(40)]
691
+ status_outputs = [f"{role_texts[i].get('display_name', f'Role {i+1}')} : Generation cancelled" if role_texts[i] else "" for i in range(40)]
692
+ # Ensure all buttons are restored to interactive state
693
  for i in range(40):
694
  if role_components[i][-1]: # bulk_btn
695
  role_components[i][-1].value = gr.Button(interactive=True)
696
+ return "Stopping all generation tasks...", status_outputs, button_states
697
 
698
  def get_pt_file_for_download(seed_id):
699
+ """Generate .pt file for download button and return path and status"""
700
  spk_emb, message = get_pt_file(seed_id)
701
  if spk_emb is not None:
702
  os.makedirs("tmp", exist_ok=True)
 
706
  return gr.DownloadButton(value=None, label="Download .pt File", visible=False), message
707
 
708
  def get_download_zip_state():
709
+ """Check zip file state and return download button state"""
710
  if os.path.exists(os.path.join("tmp", "outputs.zip")):
711
+ return gr.DownloadButton(value=os.path.join("tmp", "outputs.zip"), label="Download Outputs.zip", visible=True), "Ready to download outputs.zip"
712
+ return gr.DownloadButton(value=None, label="Download Outputs.zip", visible=False), "outputs.zip not found"
713
 
714
  def main():
715
+ # Clear cache folder at script start
716
  # clear_audio_cache_dir()
717
 
718
  global role_components, make_bulk_functions, audio_cache_dir, role_mapping_input
719
  MAX_ROLES = 40
720
+ mapping_file_path = os.path.join(os.path.dirname(__file__), "role_mapping.txt")
721
  try:
722
  with open(mapping_file_path, "r", encoding="utf-8") as f:
723
  role_mapping_placeholder = f.read().strip()
724
  except FileNotFoundError:
725
+ role_mapping_placeholder = (
726
+ "Role Code: jud → Display Name: Judge voice:jud seed_id:1403\n"
727
+ "Role Code: jury → Display Name: Members of The Jury voice:jury seed_id:1404\n"
728
+ "Role Code: noname → Display Name: Anonymous Role"
729
+ )
730
 
731
  with gr.Blocks() as demo:
732
  gr.Markdown("https://huggingface.co/spaces/ResembleAI/Chatterbox")
733
 
734
+ model_status = gr.Textbox(label="Model loading status", interactive=False, value="Loading model...")
735
 
736
  with gr.Row():
737
  default_rpy_path = os.path.join(os.path.dirname(__file__), "dialogue.rpy")
738
  file_input = gr.File(
739
+ label="Upload .rpy file",
740
  type="binary",
741
  value=default_rpy_path if os.path.exists(default_rpy_path) else None
742
  )
743
  with gr.Column():
744
  role_mapping_input = gr.Textbox(
745
+ label="Role Mapping (Format: Role Code: xxx → Display Name: xxx voice:xxx seed_id:xxx)",
746
  lines=5,
747
  placeholder=role_mapping_placeholder,
748
  value=role_mapping_placeholder
749
  )
750
  filter_input = gr.Textbox(
751
+ label="Dialogue Filter (Format: min_len=X;keyword=Y, leave empty to show all)",
752
  lines=2,
753
+ placeholder="Example: min_len=20;keyword=hello\n(Minimum length 20 characters, contains 'hello' dialogues)",
754
  value="min_len=3"
755
  )
756
  with gr.Row():
757
+ process_button = gr.Button("Process File")
758
+ stop_button = gr.Button("Stop All Generation")
759
+ zip_button = gr.Button("Zip Outputs Folder")
760
+ compress_button = gr.Button("Compress WAV to MP3")
761
+ all_status = gr.Textbox(label="Overall Status", interactive=False, visible=False)
762
  download_zip_button = gr.DownloadButton(label="Download Outputs.zip", visible=False)
763
+ zip_status = gr.Textbox(label="Zip Status", interactive=False)
764
 
765
  with gr.Row():
766
  seed_id_input = gr.Textbox(
767
+ label="Enter seed_id to get .pt file",
768
+ placeholder="e.g.: 1403 or seed_1403",
769
  visible=False
770
  )
771
  pt_download_button = gr.DownloadButton(label="Download .pt File", visible=False)
772
+ pt_status = gr.Textbox(label="Generate .pt file status", interactive=False)
773
 
774
  def load_model_on_start():
775
  model, status = get_or_load_model()
 
814
  with gr.Row():
815
  for i in range(row_idx * ROLES_PER_ROW, min((row_idx + 1) * ROLES_PER_ROW, MAX_ROLES)):
816
  with gr.Group(visible=False, elem_classes="compact-group") as group:
817
+ role_display = gr.Markdown(f"### Role {i+1}", elem_classes="compact-header")
818
  text_input = gr.Textbox(
819
+ label="Text",
820
  lines=2,
821
  max_lines=4,
822
  elem_classes="compact-textbox",
 
824
  )
825
  start_index = gr.Number(
826
  value=1,
827
+ label="Start from line",
828
  minimum=1,
829
  step=1,
830
  elem_classes="compact-number"
831
  )
832
  spk_emb_seed_id = gr.Textbox(
833
+ label="Enter spk_emb seed_id",
834
+ placeholder="e.g.: 1403 or seed_1403",
835
  elem_classes="compact-textbox",
836
  visible=False
837
  )
838
  ref_wav = gr.Audio(
839
  type="filepath",
840
+ label="Reference audio file (optional, recommended >6s)",
841
  sources=["upload", "microphone"],
842
  elem_classes="compact-audio"
843
  )
844
  audio_output = gr.Audio(
845
+ label="Output audio",
846
  elem_classes="compact-audio",
847
  visible=False
848
  )
849
  status = gr.Textbox(
850
+ label="Status",
851
  interactive=False,
852
  elem_classes="compact-textbox",
853
  container=False
854
  )
855
  bulk_btn = gr.Button(
856
+ f"Generate Role {i+1}",
857
  size="sm",
858
  elem_classes="compact-button",
859
  interactive=True
 
866
  global generation_queue, is_generating, queued_roles
867
  role_data = role_texts[i]
868
  if not role_data:
869
+ return None, f"Role {i+1} has no data", gr.Button(interactive=True)
870
+ role = role_data.get("role", f"Role{i+1}")
871
  display_name = role_data.get("display_name", role)
872
 
873
+ # Prefer UI provided ref_wav, but check if needs caching
874
  ref_wav_path = ref_wav
875
  if ref_wav and os.path.exists(ref_wav):
876
+ # Check if ref_wav is already in cache dir
877
  if not ref_wav.startswith(audio_cache_dir):
878
  cached_path = os.path.join(audio_cache_dir, f"role_{i}_{display_name}_ui.wav")
879
  shutil.copy(ref_wav, cached_path)
880
  role_texts[i]["ref_wav"] = cached_path
881
+ print(f"\033[92m✅ Cached UI provided reference audio for {display_name}: {ref_wav} -> {cached_path}\033[0m")
882
  else:
883
+ # Already cached, use directly
884
  role_texts[i]["ref_wav"] = ref_wav
885
+ print(f"\033[92m✅ Using already cached reference audio for {display_name}: {ref_wav}\033[0m")
886
  else:
887
  ref_wav_path = role_texts[i].get("ref_wav")
888
 
889
+ # Check if spk_emb_seed_id or ref_wav provided
890
  if not spk_emb_seed_id and not ref_wav_path:
891
+ print(f"\033[93m⚠️ Role {display_name} did not provide spk_emb_seed_id or reference audio, skipping generation\033[0m")
892
+ return None, f"{display_name}: No spk_emb_seed_id or reference audio provided, skipping generation", gr.Button(interactive=True)
893
 
894
+ # Cache role config to queue
895
  role_config = {
896
  'role_index': i,
897
  'txt': txt,
 
905
  generation_queue.append(role_config)
906
  queued_roles.add(i)
907
  pos = len(generation_queue)
908
+ print(f"\033[94m⏳ Role {display_name} added to queue, position {pos}/{pos}\033[0m")
909
+ return None, f"{display_name}: Waiting in queue, position {pos}/{pos}", gr.Button(interactive=False)
910
  else:
911
  is_generating.set()
912
  queued_roles.add(i)
913
  cancel_generation.clear()
914
+ print(f"\033[94m🚀 Single role generation: {display_name}, from line {start_idx}\033[0m")
915
  file_lines = []
916
  if last_uploaded_file:
917
  try:
918
  _, file_lines = parse_roles_from_rpy_file(last_uploaded_file, "")
919
  except Exception as e:
920
+ return None, f"File parsing failed: {str(e)}", gr.Button(interactive=True)
921
  role_map = parse_role_mappings(role_mapping_input.value)
922
  result = generate_all_lines_audio(
923
  role, role_data.get("lines", []), file_lines, int(start_idx), role_map,
 
926
  return result
927
  return inner
928
 
929
+ # Store make_bulk function, pass bulk_btn
930
  make_bulk_functions[i] = make_bulk(i, bulk_btn)
931
 
932
  bulk_btn.click(