Persist and poll live training logs

app.py

@@ -98,6 +98,9 @@ class VersionContext:
     exp_dir: Path
     text_path: Path
     semantic_path: Path
+    prep_live_log: Path
+    sovits_live_log: Path
+    gpt_live_log: Path
     sovits_ckpt_dir: Path
     gpt_log_dir: Path
     sovits_output_dir: Path
@@ -165,6 +168,9 @@ def get_version_context(version: str):
         exp_dir=exp_dir,
         text_path=exp_dir / "2-name2text.txt",
         semantic_path=exp_dir / "6-name2semantic.tsv",
+        prep_live_log=exp_dir / "_live_prepare.log",
+        sovits_live_log=exp_dir / "_live_sovits.log",
+        gpt_live_log=exp_dir / "_live_gpt.log",
         sovits_ckpt_dir=exp_dir / f"logs_s2_{version}",
         gpt_log_dir=exp_dir / f"logs_s1_{version}",
         sovits_output_dir=OUTPUT_ROOT / f"SoVITS_weights_{version}",
@@ -204,8 +210,28 @@ def hf_kwargs():
     return {"token": HF_TOKEN} if HF_TOKEN else {}
 
 
-def push(logs, message):
+def append_live_log(path: Path, message: str):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(message + "\n")
+
+
+def clear_live_log(path: Path):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("", encoding="utf-8")
+
+
+def read_live_log(path: Path):
+    if not path.exists():
+        return ""
+    lines = path.read_text(encoding="utf-8", errors="ignore").splitlines()
+    return "\n".join(lines[-200:])
+
+
+def push(logs, message, live_path: Path | None = None):
     logs.append(message)
+    if live_path is not None:
+        append_live_log(live_path, message)
     return "\n".join(logs[-200:])
 
 
@@ -569,37 +595,38 @@ def create_gpt_config(ctx: VersionContext, epochs, batch_size, save_every_epoch)
     return tmp_config
 
 
-def setup_environment_steps(logs, version):
+def setup_environment_steps(logs, version, live_path: Path | None = None):
     ctx = get_version_context(version)
     ensure_dirs()
-    yield push(logs, f"目标版本：{ctx.spec.version}（{ctx.spec.stability}）")
+    yield push(logs, f"目标版本：{ctx.spec.version}（{ctx.spec.stability}）", live_path)
     if (GPT_SOVITS_DIR / "webui.py").exists():
-        yield push(logs, "GPT-SoVITS 仓库已存在，跳过克隆。")
+        yield push(logs, "GPT-SoVITS 仓库已存在，跳过克隆。", live_path)
     else:
-        yield push(logs, "克隆 GPT-SoVITS 仓库...")
+        yield push(logs, "克隆 GPT-SoVITS 仓库...", live_path)
         ensure_upstream_repo()
-        yield push(logs, "✅ GPT-SoVITS 仓库已就绪。")
+        yield push(logs, "✅ GPT-SoVITS 仓库已就绪。", live_path)
     patch_upstream_repo()
-    yield push(logs, "✅ 已应用 Space 兼容补丁。")
+    yield push(logs, "✅ 已应用 Space 兼容补丁。", live_path)
     if not has_transformers_model(BERT_DIR):
-        yield push(logs, "下载中文 BERT 特征模型...")
+        yield push(logs, "下载中文 BERT 特征模型...", live_path)
     if not has_transformers_model(CNHUBERT_DIR):
-        yield push(logs, "下载 CN-HuBERT 特征模型...")
+        yield push(logs, "下载 CN-HuBERT 特征模型...", live_path)
     if not ctx.pretrained_s1.exists() or not ctx.pretrained_s2g.exists():
-        yield push(logs, f"下载 GPT-SoVITS {ctx.spec.version} 底模...")
+        yield push(logs, f"下载 GPT-SoVITS {ctx.spec.version} 底模...", live_path)
     if ctx.sv_pretrained and not ctx.sv_pretrained.exists():
-        yield push(logs, f"下载 {ctx.spec.version} 的 speaker embedding 底模...")
+        yield push(logs, f"下载 {ctx.spec.version} 的 speaker embedding 底模...", live_path)
     ensure_base_assets(version)
     yield push(
         logs,
         f"✅ 环境就绪：GPT-SoVITS 仓库、中文特征模型和 {ctx.spec.version} 底模均已准备完成。",
+        live_path,
     )
 
 
-def download_dataset_steps(logs, version):
+def download_dataset_steps(logs, version, live_path: Path | None = None):
     ensure_dirs()
-    yield from setup_environment_steps(logs, version)
-    yield push(logs, "下载 Daniya 数据集...")
+    yield from setup_environment_steps(logs, version, live_path)
+    yield push(logs, "下载 Daniya 数据集...", live_path)
     snapshot_download(
         repo_id=DATASET_REPO,
         repo_type="dataset",
@@ -611,34 +638,37 @@ def download_dataset_steps(logs, version):
     yield push(
         logs,
         f"✅ 数据集已下载：音频 {audio_count} 个，metadata {len(rows)} 条。",
+        live_path,
     )
 
 
-def prepare_data_steps(logs, version):
+def prepare_data_steps(logs, version, live_path: Path | None = None):
     ctx = get_version_context(version)
     ensure_dirs()
     lock_path = ctx.exp_dir / ".preprocess.lock"
-    wait_logged = False
+    wait_notice_at = 0.0
     with lock_path.open("a+", encoding="utf-8") as lock_handle:
         while True:
             try:
                 fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                 break
             except BlockingIOError:
-                if not wait_logged:
-                    yield push(logs, f"{ctx.spec.version} 预处理中，等待已有任务完成...")
-                    wait_logged = True
+                now = time.time()
+                if now >= wait_notice_at:
+                    yield push(logs, f"{ctx.spec.version} 预处理中，等待已有任务完成...", live_path)
+                    wait_notice_at = now + 10
                 time.sleep(2)
         try:
             if dataset_prepared(ctx):
                 return f"✅ 预处理已就绪（{ctx.spec.version}），无需重复执行。"
-            yield from download_dataset_steps(logs, version)
+            yield from download_dataset_steps(logs, version, live_path)
             reset_preprocess_outputs(ctx)
             sample_count, audio_count, unlisted = build_manifest(ctx)
             ensure_preprocess_dirs(ctx)
             yield push(
                 logs,
                 f"训练清单已生成：metadata 可用样本 {sample_count} 条，音频总数 {audio_count} 个，未标注音频 {len(unlisted)} 个。",
+                live_path,
             )
             env = build_process_env(ctx)
             for line in run_cmd(
@@ -646,33 +676,33 @@ def prepare_data_steps(logs, version):
                 cwd=GPT_SOVITS_DIR,
                 env=env,
             ):
-                yield push(logs, line)
+                yield push(logs, line, live_path)
             part_text = ctx.exp_dir / "2-name2text-0.txt"
             if not part_text.exists():
                 raise RuntimeError("文本特征提取完成后未生成 2-name2text-0.txt")
             part_text.replace(ctx.text_path)
-            yield push(logs, "✅ 文本分词与 BERT 特征提取完成。")
+            yield push(logs, "✅ 文本分词与 BERT 特征提取完成。", live_path)
             for line in run_cmd(
                 [sys.executable, "-s", "GPT_SoVITS/prepare_datasets/2-get-hubert-wav32k.py"],
                 cwd=GPT_SOVITS_DIR,
                 env=env,
             ):
-                yield push(logs, line)
-            yield push(logs, "✅ CN-HuBERT 特征与 32k wav 已生成。")
+                yield push(logs, line, live_path)
+            yield push(logs, "✅ CN-HuBERT 特征与 32k wav 已生成。", live_path)
             if ctx.spec.uses_sv:
                 for line in run_cmd(
                     [sys.executable, "-s", "GPT_SoVITS/prepare_datasets/2-get-sv.py"],
                     cwd=GPT_SOVITS_DIR,
                     env=env,
                 ):
-                    yield push(logs, line)
-                yield push(logs, "✅ speaker embedding 特征已生成。")
+                    yield push(logs, line, live_path)
+                yield push(logs, "✅ speaker embedding 特征已生成。", live_path)
             for line in run_cmd(
                 [sys.executable, "-s", "GPT_SoVITS/prepare_datasets/3-get-semantic.py"],
                 cwd=GPT_SOVITS_DIR,
                 env=env,
             ):
-                yield push(logs, line)
+                yield push(logs, line, live_path)
             part_semantic = ctx.exp_dir / "6-name2semantic-0.tsv"
             if not part_semantic.exists():
                 raise RuntimeError("语义 token 提取完成后未生成 6-name2semantic-0.tsv")
@@ -682,7 +712,7 @@ def prepare_data_steps(logs, version):
                 encoding="utf-8",
             )
             part_semantic.unlink()
-            yield push(logs, "✅ 语义 token 提取完成。")
+            yield push(logs, "✅ 语义 token 提取完成。", live_path)
             return f"✅ 预处理完成（{ctx.spec.version}），可用于训练的样本 {sample_count} 条。"
         finally:
             fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
@@ -690,98 +720,123 @@ def prepare_data_steps(logs, version):
 
 def check_environment(version=DEFAULT_VERSION):
     logs = []
+    ctx = get_version_context(version)
+    clear_live_log(ctx.prep_live_log)
     try:
         final = None
-        final = yield from setup_environment_steps(logs, version)
+        final = yield from setup_environment_steps(logs, version, ctx.prep_live_log)
         if final:
             yield final
     except Exception as exc:
         log.exception("check_environment")
-        yield push(logs, f"❌ 环境准备失败: {exc}")
+        yield push(logs, f"❌ 环境准备失败: {exc}", ctx.prep_live_log)
 
 
 def download_dataset(version=DEFAULT_VERSION):
     logs = []
+    ctx = get_version_context(version)
+    clear_live_log(ctx.prep_live_log)
     try:
         final = None
-        final = yield from download_dataset_steps(logs, version)
+        final = yield from download_dataset_steps(logs, version, ctx.prep_live_log)
         if final:
             yield final
     except Exception as exc:
         log.exception("download_dataset")
-        yield push(logs, f"❌ 数据集下载失败: {exc}")
+        yield push(logs, f"❌ 数据集下载失败: {exc}", ctx.prep_live_log)
 
 
 def prepare_data(version=DEFAULT_VERSION):
     logs = []
+    ctx = get_version_context(version)
+    clear_live_log(ctx.prep_live_log)
     try:
-        final = yield from prepare_data_steps(logs, version)
-        yield push(logs, final)
+        final = yield from prepare_data_steps(logs, version, ctx.prep_live_log)
+        yield push(logs, final, ctx.prep_live_log)
     except Exception as exc:
         log.exception("prepare_data")
-        yield push(logs, f"❌ 预处理失败: {exc}")
+        yield push(logs, f"❌ 预处理失败: {exc}", ctx.prep_live_log)
 
 
 def start_training(version=DEFAULT_VERSION, epochs=2, batch_size=1, save_every_epoch=1, lr=0.0001):
     logs = []
     ctx = get_version_context(version)
+    clear_live_log(ctx.sovits_live_log)
     try:
-        yield push(logs, f"当前版本：{ctx.spec.version}（{ctx.spec.stability}）"), None
+        yield push(logs, f"当前版本：{ctx.spec.version}（{ctx.spec.stability}）", ctx.sovits_live_log), None
         if not dataset_prepared(ctx):
-            yield push(logs, f"{ctx.spec.version} 缺少预处理产物，开始自动补齐..."), None
-            for update in prepare_data_steps(logs, version):
+            yield push(logs, f"{ctx.spec.version} 缺少预处理产物，开始自动补齐...", ctx.sovits_live_log), None
+            for update in prepare_data_steps(logs, version, ctx.sovits_live_log):
                 yield update, None
         config_path = create_sovits_config(ctx, epochs, batch_size, save_every_epoch, lr)
         env = build_process_env(ctx)
-        yield push(logs, f"开始 SoVITS 训练（{ctx.spec.version}）..."), None
+        yield push(logs, f"开始 SoVITS 训练（{ctx.spec.version}）...", ctx.sovits_live_log), None
         for line in run_cmd(
             [sys.executable, "-s", "GPT_SoVITS/s2_train.py", "--config", str(config_path)],
             cwd=GPT_SOVITS_DIR,
             env=env,
         ):
-            yield push(logs, line), None
+            yield push(logs, line, ctx.sovits_live_log), None
         latest = latest_file(ctx.sovits_output_dir, ".pth")
         if not latest:
             raise RuntimeError("训练结束后没有找到导出的 SoVITS 权重文件")
-        yield push(logs, f"✅ SoVITS 训练完成，最新权重：{latest}"), latest
+        yield push(logs, f"✅ SoVITS 训练完成，最新权重：{latest}", ctx.sovits_live_log), latest
     except Exception as exc:
         log.exception("start_training")
-        yield push(logs, f"❌ SoVITS 训练失败: {exc}"), None
+        yield push(logs, f"❌ SoVITS 训练失败: {exc}", ctx.sovits_live_log), None
 
 
 def start_gpt_training(version=DEFAULT_VERSION, epochs=1, batch_size=1, save_every_epoch=1):
     logs = []
     ctx = get_version_context(version)
+    clear_live_log(ctx.gpt_live_log)
     try:
-        yield push(logs, f"当前版本：{ctx.spec.version}（{ctx.spec.stability}）"), None
+        yield push(logs, f"当前版本：{ctx.spec.version}（{ctx.spec.stability}）", ctx.gpt_live_log), None
         if not dataset_prepared(ctx):
-            yield push(logs, f"{ctx.spec.version} 缺少预处理产物，开始自动补齐..."), None
-            for update in prepare_data_steps(logs, version):
+            yield push(logs, f"{ctx.spec.version} 缺少预处理产物，开始自动补齐...", ctx.gpt_live_log), None
+            for update in prepare_data_steps(logs, version, ctx.gpt_live_log):
                 yield update, None
         reset_gpt_training_outputs(ctx)
-        yield push(logs, "已清理旧 GPT 断点与导出文件，避免恢复到不兼容 checkpoint。"), None
+        yield push(logs, "已清理旧 GPT 断点与导出文件，避免恢复到不兼容 checkpoint。", ctx.gpt_live_log), None
         config_path = create_gpt_config(ctx, epochs, batch_size, save_every_epoch)
         env = build_process_env(ctx)
-        yield push(logs, f"开始 GPT 训练（{ctx.spec.version}）..."), None
+        yield push(logs, f"开始 GPT 训练（{ctx.spec.version}）...", ctx.gpt_live_log), None
         for line in run_cmd(
             [sys.executable, "-s", "GPT_SoVITS/s1_train.py", "--config_file", str(config_path)],
             cwd=GPT_SOVITS_DIR,
             env=env,
         ):
-            yield push(logs, line), None
+            yield push(logs, line, ctx.gpt_live_log), None
         latest = latest_file(ctx.gpt_output_dir, ".ckpt")
         if not latest:
             raise RuntimeError("训练结束后没有找到导出的 GPT 权重文件")
-        yield push(logs, f"✅ GPT 训练完成，最新权重：{latest}"), latest
+        yield push(logs, f"✅ GPT 训练完成，最新权重：{latest}", ctx.gpt_live_log), latest
     except Exception as exc:
         log.exception("start_gpt_training")
-        yield push(logs, f"❌ GPT 训练失败: {exc}"), None
+        yield push(logs, f"❌ GPT 训练失败: {exc}", ctx.gpt_live_log), None
 
 
 def refresh_outputs(version=DEFAULT_VERSION):
     return artifacts_summary(version)
 
 
+def live_logs(version=DEFAULT_VERSION):
+    ctx = get_version_context(version)
+    return (
+        read_live_log(ctx.prep_live_log),
+        read_live_log(ctx.sovits_live_log),
+        read_live_log(ctx.gpt_live_log),
+    )
+
+
+def sync_live_state(version=DEFAULT_VERSION):
+    return (
+        *live_logs(version),
+        *refresh_outputs(version),
+        agent_status(version),
+    )
+
+
 def version_markdown(version=DEFAULT_VERSION):
     spec = get_version_spec(version)
     return (
@@ -844,8 +899,7 @@ def load_dashboard():
     version = DEFAULT_VERSION
     return (
         version_markdown(version),
-        *refresh_outputs(version),
-        agent_status(version),
+        *sync_live_state(version),
     )
 
 
@@ -859,6 +913,7 @@ def create_ui():
             "打开页面会自动加载当前输出。训练完成后，最新模型会直接出现在下载框里，"
             "下面的“输出与目录”也会自动刷新，避免找不到文件。"
         )
+        gr.Markdown("日志会同步写入 `/data`，页面每 2 秒自动轮询；即使刷新页面，也会把当前训练日志重新拉回来。")
         version_select = gr.Dropdown(
             choices=list(SUPPORTED_VERSIONS),
             value=DEFAULT_VERSION,
@@ -924,8 +979,13 @@ def create_ui():
         )
         agent_status_out = gr.JSON(label="Agent 状态", value=agent_status(DEFAULT_VERSION))
         agent_status_btn = gr.Button("刷新 Agent 状态", variant="secondary")
+        sync_timer = gr.Timer(value=2, active=True)
+        refresh_api_btn = gr.Button(visible=False)
+        live_logs_api_btn = gr.Button(visible=False)
+        sync_state_api_btn = gr.Button(visible=False)
 
         refresh_outputs_targets = [refresh_text, output_dirs, refresh_sovits, refresh_gpt, all_sovits, all_gpt]
+        sync_targets = [prep_out, sovits_log, gpt_log, *refresh_outputs_targets, agent_status_out]
         env_btn.click(check_environment, inputs=[version_select], outputs=env_out, api_name="check_environment")
         dataset_btn.click(download_dataset, inputs=[version_select], outputs=dataset_out, api_name="download_dataset")
         prep_btn.click(prepare_data, inputs=[version_select], outputs=prep_out, api_name="prepare_data")
@@ -941,30 +1001,41 @@ def create_ui():
             outputs=[gpt_log, gpt_file],
             api_name="start_gpt_training",
         )
-        refresh_btn.click(refresh_outputs, inputs=[version_select], outputs=refresh_outputs_targets, api_name="refresh_outputs")
+        refresh_btn.click(sync_live_state, inputs=[version_select], outputs=sync_targets, api_name=False)
+        refresh_api_btn.click(refresh_outputs, inputs=[version_select], outputs=refresh_outputs_targets, api_name="refresh_outputs")
+        live_logs_api_btn.click(live_logs, inputs=[version_select], outputs=[prep_out, sovits_log, gpt_log], api_name="live_logs")
+        sync_state_api_btn.click(sync_live_state, inputs=[version_select], outputs=sync_targets, api_name="sync_live_state")
         agent_status_btn.click(agent_status, inputs=[version_select], outputs=agent_status_out, api_name="agent_status")
         version_select.change(version_markdown, inputs=[version_select], outputs=version_note, api_name=False)
-        version_select.change(refresh_outputs, inputs=[version_select], outputs=refresh_outputs_targets, api_name=False).then(
-            agent_status,
+        version_select.change(
+            sync_live_state,
+            inputs=[version_select],
+            outputs=sync_targets,
+            api_name=False,
+        )
+        sovits_event.then(
+            sync_live_state,
             inputs=[version_select],
-            outputs=agent_status_out,
+            outputs=sync_targets,
             api_name=False,
         )
-        sovits_event.then(refresh_outputs, inputs=[version_select], outputs=refresh_outputs_targets, api_name=False).then(
-            agent_status,
+        gpt_event.then(
+            sync_live_state,
             inputs=[version_select],
-            outputs=agent_status_out,
+            outputs=sync_targets,
             api_name=False,
         )
-        gpt_event.then(refresh_outputs, inputs=[version_select], outputs=refresh_outputs_targets, api_name=False).then(
-            agent_status,
+        sync_timer.tick(
+            sync_live_state,
             inputs=[version_select],
-            outputs=agent_status_out,
+            outputs=sync_targets,
             api_name=False,
+            queue=False,
+            show_progress="hidden",
         )
         demo.load(
             load_dashboard,
-            outputs=[version_note, *refresh_outputs_targets, agent_status_out],
+            outputs=[version_note, *sync_targets],
             api_name=False,
         )