[v0.0.3] fix download issue of processed data (#67)

* [fix] Fixed serving list sync issue.

* [task] update `/execution/{task_id}/download` to fix url bug and task registry init bug (#66)

---------

Co-authored-by: alever_sn <759636538@qq.com>

Author: SunnyHaze

backend/app/api/v1/endpoints/tasks.py

@@ -115,7 +115,7 @@ def get_execution_log(task_id: str, operator_name: str = Query(None, description
         raise HTTPException(500, f"Failed to get task logs: {str(e)}")
 
 
-@router.get("/execution/{task_id}/download", operation_id="download_task_result", summary="下载任务执行结果文件")
+@router.get("/execution/{task_id}/download", operation_id="download_task_result", summary="下载任务执行结果文件,step从0开始计数，想请求第一个算子传step=0")
 def download_task_result(task_id: str, step: int = None):
     """
     下载任务执行结果文件
@@ -160,15 +160,20 @@ def download_task_result(task_id: str, step: int = None):
         # 构建缓存文件路径（使用绝对路径）
         from app.core.config import settings
         cache_path = settings.CACHE_DIR
+        cache_task_dir = f"{task_id}_output"
         cache_file_prefix = "dataflow_cache_step"
-        cache_file = os.path.join(cache_path, f"{cache_file_prefix}_step{step}.jsonl")
+        actual_step_for_json = step + 1
+
+        cache_file_name = f"{cache_file_prefix}_step{actual_step_for_json}.jsonl"
+
+        cache_file = os.path.join(cache_path, cache_task_dir, cache_file_name)
 
         # 检查文件是否存在
         if not os.path.exists(cache_file):
             raise HTTPException(404, f"Result file not found for step {step}: {cache_file}")
 
         # 返回文件下载
-        filename = f"{task_id}_{operator_name}_step{step}.jsonl"
+        filename = f"{task_id}_{operator_name}_step{actual_step_for_json}.jsonl"
         logger.info(f"Downloading file: {cache_file} as {filename}")
 
         return FileResponse(

backend/app/services/task_registry.py

@@ -19,11 +19,28 @@ def __init__(self, path: str | None = None):
         self._ensure()
 
     def _ensure(self):
-        """确保注册表文件存在"""
-        os.makedirs(os.path.dirname(self.path), exist_ok=True)
-        initial_data = {"tasks": {}}
-        self._write(initial_data)   
-    
+        # 只在文件不存在时初始化
+        if not os.path.exists(self.path):
+            initial_data = {"tasks": {}}
+            self._write(initial_data)
+            return
+
+        # 文件存在但为空/损坏时兜底（可选）
+        try:
+            with open(self.path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+            if not isinstance(data, dict) or "tasks" not in data:
+                self._write({"tasks": {}})
+        except Exception:
+            # 读取失败：备份旧文件再重建（推荐）
+            broken = self.path + f".broken.{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}"
+            try:
+                os.replace(self.path, broken)
+            except Exception:
+                pass
+            self._write({"tasks": {}})
+
+
     def _read(self) -> Dict:
         """读取注册表"""
         with open(self.path, "r", encoding="utf-8") as f:

frontend/src/stores/dataflow.js

@@ -150,7 +150,7 @@ export const useDataflow = defineStore('useDataflow', () => {
     const servingList = ref([])
     const currentServing = ref(null)
     const getServingList = async () => {
-        await proxy.$api.serving.list_serving_instances_api_v1_serving__get().then((res) => {
+        return await proxy.$api.serving.list_serving_instances_api_v1_serving__get().then((res) => {
             if (res.code === 200) {
                 res.data.forEach((item) => {
                     item.key = item.id
@@ -162,6 +162,7 @@ export const useDataflow = defineStore('useDataflow', () => {
                     status: 'warning'
                 })
             }
+            return res
         })
     }
     const chooseServing = (item) => {

frontend/src/views/manage/dataflow/index.vue

@@ -610,12 +610,22 @@ export default {
                     }
                 })
         },
+        pathJoin(...parts) {
+            return parts
+                .map((part, index) => {
+                    if (index === 0) {
+                        return part.replace(/\/+$/, '')   // 去掉结尾 /
+                    }
+                    return part.replace(/^\/+|\/+$/g, '') // 去掉两边 /
+                })
+                .join('/')
+        },
         downloadData(pipeline_idx) {
             if (!this.executionInfo.task_id) return
             let baseURL = axios.defaults.baseURL
             let url =
-                baseURL +
-                `/api/v1/tasks/execution/${this.executionInfo.task_id}/download?step=${pipeline_idx - 1}`
+                this.pathJoin(baseURL,
+                    `/api/v1/tasks/execution/${this.executionInfo.task_id}/download?step=${pipeline_idx - 1}`)
             window.open(url, '_blank')
         },
         selectPipelineCallback() {

frontend/src/views/manage/serving/index.vue

@@ -115,9 +115,10 @@
 </template>
 
 <script>
-import { mapState } from 'pinia'
+import { mapActions, mapState } from 'pinia'
 import { useAppConfig } from '@/stores/appConfig'
 import { useTheme } from '@/stores/theme'
+import { useDataflow } from '@/stores/dataflow';
 
 export default {
     data() {
@@ -167,6 +168,9 @@ export default {
         this.getServingList()
     },
     methods: {
+        ...mapActions(useDataflow, {
+            getGlobalServingList: 'getServingList',
+        }),
         getCreateProps() {
             this.$api.serving.list_serving_classes().then((res) => {
                 if (res.data) {
@@ -184,16 +188,15 @@ export default {
                 }
             })
         },
-        getServingList() {
-            this.$api.serving.list_serving_instances_api_v1_serving__get().then((res) => {
-                if (res.data) {
-                    let servingList = res.data
-                    servingList.forEach((item) => {
-                        this.resetEditParams(item, true)
-                    })
-                    this.servingList = servingList
-                }
-            })
+        async getServingList() {
+            let res = await this.getGlobalServingList()
+            if (res.data) {
+                let servingList = res.data
+                servingList.forEach((item) => {
+                    this.resetEditParams(item, true)
+                })
+                this.servingList = servingList
+            }
         },
         formatPropsValue(val) {
             if (val === null) return null