feat(export): 添加逻辑路径构建功能支持文件管理

- 在导出服务中实现_build_logical_path方法用于构建相对路径
- 更新数据集文件记录以包含logical_path字段
- 在比率任务服务中实现build_logical_path静态方法
- 将逻辑路径信息添加到数据集文件记录中
- 规范化路径处理并替换反斜杠为正斜杠
- 添加无效路径验证防止目录遍历安全问题
This commit is contained in:
2026-02-06 18:46:44 +08:00
parent 05752678cc
commit e862925a06
2 changed files with 22 additions and 0 deletions

View File

@@ -187,11 +187,13 @@ class RatioTaskService:
dst_dir = os.path.dirname(new_path)
await asyncio.to_thread(os.makedirs, dst_dir, exist_ok=True)
await asyncio.to_thread(shutil.copy2, src_path, new_path)
logical_path = RatioTaskService.build_logical_path(dst_prefix, new_path)
file_data = {
"dataset_id": target_ds.id, # type: ignore
"file_name": file_name,
"file_path": new_path,
"logical_path": logical_path,
"file_type": f.file_type,
"file_size": f.file_size,
"check_sum": f.check_sum,
@@ -204,6 +206,15 @@ class RatioTaskService:
session.add(DatasetFiles(**file_record))
existing_paths.add(new_path)
@staticmethod
def build_logical_path(dataset_prefix: str, file_path: str) -> str:
normalized_dataset_prefix = os.path.abspath(dataset_prefix)
normalized_file_path = os.path.abspath(file_path)
relative_path = os.path.relpath(normalized_file_path, normalized_dataset_prefix).replace("\\", "/").strip()
if relative_path in ("", ".") or relative_path.startswith("../"):
raise ValueError(f"Invalid logical path generated for file: {file_path}")
return relative_path
@staticmethod
def get_new_file_name(dst_prefix: str, existing_paths: set[Any], f) -> str:
file_name = f.file_name