Add safe web file management

app.py

@@ -347,6 +347,198 @@ def artifacts_summary(version=DEFAULT_VERSION):
     )
 
 
+def path_is_under(path: Path, root: Path):
+    try:
+        path.resolve().relative_to(root.resolve())
+        return True
+    except ValueError:
+        return False
+
+
+def managed_roots(ctx: VersionContext):
+    return [
+        ("版本日志目录", ctx.exp_dir),
+        ("SoVITS 输出目录", ctx.sovits_output_dir),
+        ("GPT 输出目录", ctx.gpt_output_dir),
+        ("数据音频目录", AUDIO_DIR),
+        ("临时目录", TMP_ROOT),
+    ]
+
+
+def managed_single_paths(ctx: VersionContext):
+    return [
+        ("metadata", DATASET_DIR / "metadata.csv"),
+        ("训练清单", ctx.input_list),
+        ("文本特征总表", ctx.text_path),
+        ("语义特征总表", ctx.semantic_path),
+        ("预处理实时日志", ctx.prep_live_log),
+        ("SoVITS 实时日志", ctx.sovits_live_log),
+        ("GPT 实时日志", ctx.gpt_live_log),
+    ]
+
+
+def managed_path_choices(ctx: VersionContext):
+    items = set()
+    for _, root in managed_roots(ctx):
+        if not root.exists():
+            continue
+        for child in sorted(root.iterdir(), key=lambda item: item.name):
+            rel = child.relative_to(WORK_DIR).as_posix()
+            if child.is_dir():
+                rel = f"{rel}/"
+            items.add(rel)
+    for _, path in managed_single_paths(ctx):
+        if path.exists():
+            items.add(path.relative_to(WORK_DIR).as_posix())
+    return sorted(items)
+
+
+def describe_path(path: Path):
+    if not path.exists():
+        return f"不存在: {path}"
+    lines = [
+        f"路径: {path}",
+        f"类型: {'目录' if path.is_dir() else '文件'}",
+        f"修改时间: {format_mtime(path)}",
+    ]
+    if path.is_file():
+        lines.append(f"大小: {format_size(path.stat().st_size)}")
+    else:
+        children = list(path.iterdir())
+        lines.append(f"直接子项: {len(children)}")
+    return "\n".join(lines)
+
+
+def file_manager_summary(version=DEFAULT_VERSION):
+    ctx = get_version_context(version)
+    lines = [
+        "文件管理",
+        f"版本: {ctx.spec.version}",
+        "说明: 这里只暴露安全清理范围，不包含 GPT-SoVITS 仓库和预训练缓存。",
+    ]
+    for label, root in managed_roots(ctx):
+        if not root.exists():
+            lines.append(f"{label}: 不存在")
+            continue
+        children = sorted(root.iterdir(), key=lambda item: item.name)
+        lines.append(f"{label}: {root} | 直接子项 {len(children)}")
+        for child in children[:12]:
+            rel = child.relative_to(WORK_DIR).as_posix()
+            kind = "D" if child.is_dir() else "F"
+            suffix = "/" if child.is_dir() else ""
+            if child.is_dir():
+                extra = f"{len(list(child.iterdir()))} 项"
+            else:
+                extra = format_size(child.stat().st_size)
+            lines.append(f"[{kind}] {rel}{suffix} | {extra} | {format_mtime(child)}")
+        if len(children) > 12:
+            lines.append(f"... 其余 {len(children) - 12} 项省略")
+    for label, path in managed_single_paths(ctx):
+        if path.exists():
+            lines.append(f"{label}: {path.relative_to(WORK_DIR).as_posix()} | {format_size(path.stat().st_size)}")
+    return "\n".join(lines)
+
+
+def file_manager_state(version=DEFAULT_VERSION, selected_path=None):
+    ctx = get_version_context(version)
+    choices = managed_path_choices(ctx)
+    if selected_path not in choices:
+        selected_path = choices[0] if choices else None
+    detail = describe_managed_path(version, selected_path)
+    return (
+        file_manager_summary(version),
+        gr.update(choices=choices, value=selected_path),
+        detail,
+    )
+
+
+def describe_managed_path(version=DEFAULT_VERSION, selected_path=None):
+    if not selected_path:
+        return "未选择路径。"
+    target = (WORK_DIR / selected_path.rstrip("/")).resolve()
+    return describe_path(target)
+
+
+def allowed_delete_roots(ctx: VersionContext):
+    return [
+        ctx.exp_dir,
+        ctx.sovits_output_dir,
+        ctx.gpt_output_dir,
+        AUDIO_DIR,
+        TMP_ROOT,
+        DATASET_DIR / "metadata.csv",
+        ctx.input_list,
+        ctx.text_path,
+        ctx.semantic_path,
+        ctx.prep_live_log,
+        ctx.sovits_live_log,
+        ctx.gpt_live_log,
+    ]
+
+
+def is_allowed_delete_target(ctx: VersionContext, target: Path):
+    return any(path_is_under(target, root) for root in allowed_delete_roots(ctx))
+
+
+def manager_action_result(version, status, selected_path=None):
+    return (
+        status,
+        *file_manager_state(version, selected_path),
+        *sync_live_state(version),
+    )
+
+
+def delete_managed_path(version=DEFAULT_VERSION, selected_path=None):
+    ctx = get_version_context(version)
+    if not selected_path:
+        return manager_action_result(version, "未选择要删除的路径。", selected_path)
+    target = (WORK_DIR / selected_path.rstrip("/")).resolve()
+    if not target.exists():
+        return manager_action_result(version, f"路径不存在: {selected_path}", None)
+    if not is_allowed_delete_target(ctx, target):
+        return manager_action_result(version, f"拒绝删除: {selected_path} 不在安全范围内。", selected_path)
+    if target.is_dir():
+        shutil.rmtree(target)
+        status = f"已删除目录: {selected_path}"
+    else:
+        target.unlink()
+        status = f"已删除文件: {selected_path}"
+    return manager_action_result(version, status, None)
+
+
+def cleanup_preprocess(version=DEFAULT_VERSION, selected_path=None):
+    ctx = get_version_context(version)
+    reset_preprocess_outputs(ctx)
+    clear_live_log(ctx.prep_live_log)
+    status = f"已清理 {ctx.spec.version} 的预处理产物。"
+    return manager_action_result(version, status, selected_path)
+
+
+def cleanup_model_outputs(version=DEFAULT_VERSION, selected_path=None):
+    ctx = get_version_context(version)
+    for path in ctx.sovits_output_dir.glob("*.pth"):
+        path.unlink()
+    for path in ctx.gpt_output_dir.glob("*.ckpt"):
+        path.unlink()
+    if ctx.sovits_ckpt_dir.exists():
+        shutil.rmtree(ctx.sovits_ckpt_dir)
+        ctx.sovits_ckpt_dir.mkdir(parents=True, exist_ok=True)
+    if ctx.gpt_log_dir.exists():
+        shutil.rmtree(ctx.gpt_log_dir)
+        ctx.gpt_log_dir.mkdir(parents=True, exist_ok=True)
+    status = f"已清理 {ctx.spec.version} 的模型输出和训练断点。"
+    return manager_action_result(version, status, selected_path)
+
+
+def cleanup_live_logs(version=DEFAULT_VERSION, selected_path=None):
+    ctx = get_version_context(version)
+    clear_live_log(ctx.prep_live_log)
+    clear_live_log(ctx.sovits_live_log)
+    clear_live_log(ctx.gpt_live_log)
+    status = f"已清理 {ctx.spec.version} 的实时日志。"
+    return manager_action_result(version, status, selected_path)
+
+
 def dataset_prepared(ctx: VersionContext):
     prepared = (
         ctx.text_path.exists()
@@ -900,6 +1092,7 @@ def load_dashboard():
     return (
         version_markdown(version),
         *sync_live_state(version),
+        *file_manager_state(version),
     )
 
 
@@ -975,7 +1168,8 @@ def create_ui():
             "`/check_environment(version)`、`/download_dataset(version)`、`/prepare_data(version)`、"
             "`/start_training(version, epochs, batch_size, save_every_epoch, lr)`、"
             "`/start_gpt_training(version, epochs, batch_size, save_every_epoch)`、"
-            "`/refresh_outputs(version)`、`/agent_status(version)`。"
+            "`/refresh_outputs(version)`、`/agent_status(version)`、"
+            "`/live_logs(version)`、`/sync_live_state(version)`。"
         )
         agent_status_out = gr.JSON(label="Agent 状态", value=agent_status(DEFAULT_VERSION))
         agent_status_btn = gr.Button("刷新 Agent 状态", variant="secondary")
@@ -984,8 +1178,23 @@ def create_ui():
         live_logs_api_btn = gr.Button(visible=False)
         sync_state_api_btn = gr.Button(visible=False)
 
+        gr.Markdown("### 10. 文件管理")
+        gr.Markdown("支持安全清理当前版本的预处理、模型输出和实时日志，也可以删除选中的关键路径。")
+        manager_status = gr.Textbox(label="文件管理操作结果", lines=3, interactive=False)
+        manager_text = gr.Textbox(label="文件管理概览", lines=18, interactive=False, autoscroll=True)
+        manager_select = gr.Dropdown(choices=[], value=None, label="可管理路径")
+        manager_detail = gr.Textbox(label="所选路径详情", lines=5, interactive=False)
+        with gr.Row():
+            manager_refresh_btn = gr.Button("刷新文件列表", variant="secondary")
+            manager_delete_btn = gr.Button("删除所选路径", variant="stop")
+            cleanup_prep_btn = gr.Button("清理本版本预处理", variant="secondary")
+            cleanup_models_btn = gr.Button("清理本版本模型输出", variant="secondary")
+            cleanup_live_logs_btn = gr.Button("清理本版本实时日志", variant="secondary")
+
         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]
+        manager_targets = [manager_text, manager_select, manager_detail]
+        manager_action_targets = [manager_status, *manager_targets, *sync_targets]
         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")
@@ -1006,6 +1215,32 @@ def create_ui():
         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")
+        manager_refresh_btn.click(file_manager_state, inputs=[version_select, manager_select], outputs=manager_targets, api_name="file_manager_state")
+        manager_select.change(describe_managed_path, inputs=[version_select, manager_select], outputs=manager_detail, api_name=False)
+        manager_delete_btn.click(
+            delete_managed_path,
+            inputs=[version_select, manager_select],
+            outputs=manager_action_targets,
+            api_name="delete_managed_path",
+        )
+        cleanup_prep_btn.click(
+            cleanup_preprocess,
+            inputs=[version_select, manager_select],
+            outputs=manager_action_targets,
+            api_name="cleanup_preprocess",
+        )
+        cleanup_models_btn.click(
+            cleanup_model_outputs,
+            inputs=[version_select, manager_select],
+            outputs=manager_action_targets,
+            api_name="cleanup_model_outputs",
+        )
+        cleanup_live_logs_btn.click(
+            cleanup_live_logs,
+            inputs=[version_select, manager_select],
+            outputs=manager_action_targets,
+            api_name="cleanup_live_logs",
+        )
         version_select.change(version_markdown, inputs=[version_select], outputs=version_note, api_name=False)
         version_select.change(
             sync_live_state,
@@ -1013,6 +1248,12 @@ def create_ui():
             outputs=sync_targets,
             api_name=False,
         )
+        version_select.change(
+            file_manager_state,
+            inputs=[version_select, manager_select],
+            outputs=manager_targets,
+            api_name=False,
+        )
         sovits_event.then(
             sync_live_state,
             inputs=[version_select],
@@ -1035,7 +1276,7 @@ def create_ui():
         )
         demo.load(
             load_dashboard,
-            outputs=[version_note, *sync_targets],
+            outputs=[version_note, *sync_targets, *manager_targets],
             api_name=False,
         )