Create app.py

app.py

@@ -0,0 +1,130 @@
+import gradio as gr
+import os
+import shutil
+import zipfile
+import time # 引入 time 模块，用于在快速迭代时确保进度条有时间刷新
+
+# 1. 持久存储的挂载路径
+PERSISTENT_STORAGE_PATH = "/data"
+
+# 确保持久存储的根目录存在
+if not os.path.exists(PERSISTENT_STORAGE_PATH):
+    os.makedirs(PERSISTENT_STORAGE_PATH, exist_ok=True)
+    print(f"已创建持久存储目录: {PERSISTENT_STORAGE_PATH}")
+else:
+    print(f"持久存储目录已找到: {PERSISTENT_STORAGE_PATH}")
+
+def upload_and_extract_with_progress(zip_file_object, progress: gr.Progress):
+    """
+    处理上传的 .zip 文件，将其解压缩到持久存储，并更新进度条。
+    Gradio 会自动将 gr.Progress() 对象注入到 'progress' 参数中。
+    """
+    if zip_file_object is None:
+        return "错误：没有文件被上传。请选择一个 .zip 文件。"
+
+    temp_zip_file_path = zip_file_object.name  # 服务器上临时 .zip 文件的路径
+
+    # 2. 定义解压缩的目标文件夹名称
+    # 您可以基于上传的 .zip 文件名来命名，或者使用固定名称
+    # target_folder_name = os.path.splitext(os.path.basename(zip_file_object.name))[0] # 使用zip文件名(无扩展名)
+    target_folder_name = "my_extracted_content"  # 或者一个固定的文件夹名
+    
+    full_extraction_path = os.path.join(PERSISTENT_STORAGE_PATH, target_folder_name)
+
+    try:
+        # 3. 确保存放解压缩内容的目标文件夹存在
+        if not os.path.exists(full_extraction_path):
+            os.makedirs(full_extraction_path, exist_ok=True)
+            print(f"在持久存储中创建了目标文件夹: {full_extraction_path}")
+        else:
+            print(f"目标文件夹 '{full_extraction_path}' 已存在。文件将被解压缩到此位置（可能会覆盖）。")
+            # 可选：如果文件夹已存在，您可以在这里添加逻辑来清空它或创建新的带时间戳的文件夹
+            # shutil.rmtree(full_extraction_path)
+            # os.makedirs(full_extraction_path, exist_ok=True)
+
+
+        # 4. 解压缩过程并更新进度
+        # Gradio 的文件上传组件本身会处理上传进度。
+        # 此处的 progress 主要用于服务器端的解压过程。
+        progress(0, desc="准备解压缩...") # 初始化进度条
+
+        with zipfile.ZipFile(temp_zip_file_path, 'r') as zip_ref:
+            file_list = zip_ref.infolist() # 获取压缩包内所有文件信息
+            total_files = len(file_list)
+
+            if total_files == 0:
+                # os.remove(temp_zip_file_path) # 清理空的zip文件（可选）
+                return f"成功上传 '{os.path.basename(zip_file_object.name)}'，但压缩包为空。已将其放置在（或尝试放置在）'{temp_zip_file_path}'（临时）。目标解压路径：'{full_extraction_path}'。"
+
+
+            # 迭代解压每个文件
+            for i, member in enumerate(file_list):
+                zip_ref.extract(member, path=full_extraction_path)
+                # 更新进度：(当前步骤, 总步骤, 描述)
+                # 为了防止UI更新过于频繁导致卡顿（特别是有成千上万个小文件时），
+                # 可以选择不是每个文件都更新，或者在更新后短暂 sleep。
+                # 但对于一般情况，每个文件更新一次是可接受的。
+                progress( (i + 1) / total_files, desc=f"正在解压: {member.filename} ({i+1}/{total_files})")
+                # 如果文件非常多，且单个文件解压飞快，可能需要 time.sleep(0.001) 来让UI有机会刷新，
+                # 但通常Gradio的progress会自动处理好。
+
+        # （可选）清理上传的临时 .zip 文件
+        # try:
+        #     if os.path.exists(temp_zip_file_path):
+        #         os.remove(temp_zip_file_path)
+        #         print(f"已清理临时文件: {temp_zip_file_path}")
+        # except Exception as e_remove:
+        #     print(f"清理临时文件 {temp_zip_file_path} 时出错: {e_remove}")
+
+
+        return f"文件已成功上传并解压缩到 '{full_extraction_path}' 目录。共处理 {total_files} 个压缩包成员。"
+
+    except zipfile.BadZipFile:
+        # progress(1, desc="错误") # 更新进度条状态
+        return "错误：上传的文件不是一个有效的 ZIP 文件。"
+    except Exception as e:
+        # progress(1, desc="错误") # 更新进度条状态
+        error_message = f"处理上传和解压文件时发生错误: {str(e)}"
+        print(error_message) # 打印到容器日志，方便调试
+        return error_message
+
+# --- 创建 Gradio 应用界面 ---
+with gr.Blocks() as demo:
+    gr.Markdown(f"## 带进度条：上传 .zip 并解压到 Space 的 `/data` 目录")
+    gr.Markdown(
+        f"请选择您本地的 `.zip` 文件进行上传。"
+        f"文件上传本身会有浏览器/Gradio提供的进度指示。"
+        f"上传完成后，服务器端的解压缩过程将通��下方的进度条显示。"
+        f"最终文件将被解压缩到 Space 持久存储 `{PERSISTENT_STORAGE_PATH}` 下的一个子目录中。"
+    )
+
+    zip_file_uploader = gr.File(label="选择 .zip 文件", file_types=[".zip"])
+    # 注意：Gradio 的 gr.File 组件在文件上传时，其UI本身会显示上传百分比。
+    # 我们添加的 gr.Progress 主要用于指示服务器端解压的进度。
+
+    process_button = gr.Button("上传并解压缩")
+    
+    status_output = gr.Textbox(label="处理状态和结果", lines=5, interactive=False)
+    # gr.Progress() 组件不需要显式添加到布局中，当它作为参数传递给事件处理函数时，
+    # Gradio 会自动在合适的位置（通常是触发按钮附近或全局）显示进度条。
+
+    process_button.click(
+        fn=upload_and_extract_with_progress, # 函数签名包含 progress: gr.Progress
+        inputs=[zip_file_uploader],          # zip_file_object 作为第一个参数
+        outputs=[status_output]              # 函数的返回值将更新这个文本框
+        # Gradio 会自动将 gr.Progress() 实例传递给 fn 中的 progress 参数
+    )
+
+if __name__ == "__main__":
+    # 本地测试的注意事项
+    if PERSISTENT_STORAGE_PATH == "/data" and not os.path.exists("/data"):
+        print("警告：正在本地运行，且目标路径 /data 不存在。")
+        print("如果您想在本地完整测试持久存储逻辑，请修改 PERSISTENT_STORAGE_PATH")
+        print("为一个有效的本地目录，例如 './tmp_data/'，并确保它存在或可以被创建。")
+        # 例如，为本地测试创建一个模拟目录：
+        # test_path = "./tmp_data_for_testing"
+        # if not os.path.exists(test_path):
+        #    os.makedirs(test_path)
+        # PERSISTENT_STORAGE_PATH = test_path
+    
+    demo.launch()