From 8de5d1d038dcee3126eb666c8d0bc34a4a2ec839 Mon Sep 17 00:00:00 2001 From: uname <2986773479@qq.com> Date: Mon, 26 Jan 2026 14:37:49 +0800 Subject: [PATCH 1/3] feature: the backend realizes the basic capabilities of data lineage, and improves data lineage registration through collection, data management, and cleaning. --- .../application/CleaningTaskService.java | 33 ++++ .../DatasetApplicationService.java | 50 +++++- .../interfaces/dto/DatasetLineage.java | 30 ++++ .../interfaces/rest/DatasetController.java | 5 + .../common/domain/enums/EdgeType.java | 14 ++ .../common/domain/enums/NodeType.java | 13 ++ .../common/domain/model/LineageEdge.java | 65 +++++++ .../common/domain/model/LineageNode.java | 50 ++++++ .../common/domain/service/LineageService.java | 168 ++++++++++++++++++ .../mapper/LineageEdgeMapper.java | 16 ++ .../mapper/LineageNodeMapper.java | 16 ++ .../app/db/models/base_entity.py | 31 +++- .../module/collection/service/collection.py | 56 +++++- .../app/module/shared/common/lineage.py | 159 +++++++++++++++++ .../app/module/shared/schema/__init__.py | 8 +- .../app/module/shared/schema/lineage.py | 14 ++ scripts/db/data-common-init.sql | 58 ++++++ 17 files changed, 781 insertions(+), 5 deletions(-) create mode 100644 backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/dto/DatasetLineage.java create mode 100644 backend/shared/domain-common/src/main/java/com/datamate/common/domain/enums/EdgeType.java create mode 100644 backend/shared/domain-common/src/main/java/com/datamate/common/domain/enums/NodeType.java create mode 100644 backend/shared/domain-common/src/main/java/com/datamate/common/domain/model/LineageEdge.java create mode 100644 backend/shared/domain-common/src/main/java/com/datamate/common/domain/model/LineageNode.java create mode 100644 backend/shared/domain-common/src/main/java/com/datamate/common/domain/service/LineageService.java create mode 100644 backend/shared/domain-common/src/main/java/com/datamate/common/infrastructure/mapper/LineageEdgeMapper.java create mode 100644 backend/shared/domain-common/src/main/java/com/datamate/common/infrastructure/mapper/LineageNodeMapper.java create mode 100644 runtime/datamate-python/app/module/shared/common/lineage.py create mode 100644 runtime/datamate-python/app/module/shared/schema/lineage.py diff --git a/backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/application/CleaningTaskService.java b/backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/application/CleaningTaskService.java index 8cc2a42c..461e8809 100644 --- a/backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/application/CleaningTaskService.java +++ b/backend/services/data-cleaning-service/src/main/java/com/datamate/cleaning/application/CleaningTaskService.java @@ -10,6 +10,11 @@ import com.datamate.cleaning.domain.repository.OperatorInstanceRepository; import com.datamate.cleaning.infrastructure.validator.CleanTaskValidator; import com.datamate.cleaning.interfaces.dto.*; +import com.datamate.common.domain.enums.EdgeType; +import com.datamate.common.domain.enums.NodeType; +import com.datamate.common.domain.model.LineageEdge; +import com.datamate.common.domain.model.LineageNode; +import com.datamate.common.domain.service.LineageService; import com.datamate.common.infrastructure.exception.BusinessException; import com.datamate.common.infrastructure.exception.SystemErrorCode; import com.datamate.common.interfaces.PagedResponse; @@ -73,6 +78,8 @@ public class CleaningTaskService { private final CleanTaskValidator cleanTaskValidator; + private final LineageService lineageService; + private final String DATASET_PATH = "/dataset"; private final String FLOW_PATH = "/flow"; @@ -134,6 +141,8 @@ public CleaningTaskDto createTask(CreateCleaningTaskRequest request) { task.setBeforeSize(srcDataset.getSizeBytes()); task.setFileCount(srcDataset.getFileCount().intValue()); cleaningTaskRepo.insertTask(task); + // 记录血缘关系 + addCleaningToGraph(srcDataset, task, destDataset); operatorInstanceRepo.insertInstance(taskId, request.getInstance()); operatorRepo.incrementUsageCount(request.getInstance().stream() @@ -146,6 +155,30 @@ public CleaningTaskDto createTask(CreateCleaningTaskRequest request) { return task; } + private void addCleaningToGraph(Dataset srcDataset, CleaningTaskDto task, Dataset destDataset) { + LineageNode fromNode = new LineageNode(); + fromNode.setId(srcDataset.getId()); + fromNode.setName(srcDataset.getName()); + fromNode.setDescription(srcDataset.getDescription()); + fromNode.setNodeType(NodeType.DATASET); + + LineageNode toNode = new LineageNode(); + toNode.setId(destDataset.getId()); + toNode.setName(destDataset.getName()); + toNode.setDescription(destDataset.getDescription()); + toNode.setNodeType(NodeType.DATASET); + + LineageEdge edge = new LineageEdge(); + edge.setProcessId(task.getId()); + edge.setName(task.getName()); + edge.setDescription(task.getDescription()); + edge.setEdgeType(EdgeType.DATA_CLEANING); + edge.setFromNodeId(fromNode.getId()); + edge.setToNodeId(toNode.getId()); + + lineageService.generateGraph(fromNode, edge, toNode); + } + public CleaningTaskDto getTask(String taskId) { CleaningTaskDto task = cleaningTaskRepo.findTaskById(taskId); setProcess(task); diff --git a/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/application/DatasetApplicationService.java b/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/application/DatasetApplicationService.java index 5d4e6ffe..ae5c9b74 100644 --- a/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/application/DatasetApplicationService.java +++ b/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/application/DatasetApplicationService.java @@ -2,6 +2,11 @@ import com.baomidou.mybatisplus.core.metadata.IPage; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; +import com.datamate.common.domain.enums.EdgeType; +import com.datamate.common.domain.enums.NodeType; +import com.datamate.common.domain.model.LineageEdge; +import com.datamate.common.domain.model.LineageNode; +import com.datamate.common.domain.service.LineageService; import com.datamate.common.domain.utils.ChunksSaver; import com.datamate.common.setting.application.SysParamApplicationService; import com.datamate.datamanagement.interfaces.dto.*; @@ -17,7 +22,6 @@ import com.datamate.datamanagement.infrastructure.persistence.repository.DatasetFileRepository; import com.datamate.datamanagement.infrastructure.persistence.repository.DatasetRepository; import com.datamate.datamanagement.interfaces.converter.DatasetConverter; -import com.datamate.datamanagement.interfaces.dto.*; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; @@ -54,6 +58,7 @@ public class DatasetApplicationService { private final CollectionTaskClient collectionTaskClient; private final DatasetFileApplicationService datasetFileApplicationService; private final SysParamApplicationService sysParamService; + private final LineageService lineageService; @Value("${datamate.data-management.base-path:/dataset}") private String datasetBasePath; @@ -72,6 +77,8 @@ public Dataset createDataset(CreateDatasetRequest createDatasetRequest) { dataset.setTags(processTagNames(createDatasetRequest.getTags())); } datasetRepository.save(dataset); + // 记录血缘关系 + addDatasetToGraph(dataset, null); //todo 需要解耦这块逻辑 if (StringUtils.hasText(createDatasetRequest.getDataSource())) { @@ -81,6 +88,43 @@ public Dataset createDataset(CreateDatasetRequest createDatasetRequest) { return dataset; } + private void addDatasetToGraph(Dataset dataset, CollectionTaskDetailResponse collection) { + LineageNode datasetNode = new LineageNode(); + datasetNode.setId(dataset.getId()); + datasetNode.setNodeType(NodeType.DATASET); + datasetNode.setName(dataset.getName()); + datasetNode.setDescription(dataset.getDescription()); + + LineageNode collectionNode = null; + LineageEdge collectionEdge = null; + if(Objects.nonNull(collection)) { + collectionNode = new LineageNode(); + collectionNode.setId(collection.getId()); + collectionNode.setName(collection.getName()); + collectionNode.setDescription(collection.getDescription()); + collectionNode.setNodeType(NodeType.DATASOURCE); + + collectionEdge = new LineageEdge(); + collectionEdge.setProcessId(collection.getId()); + collectionEdge.setName(collection.getName()); + collectionEdge.setEdgeType(EdgeType.DATA_COLLECTION); + collectionEdge.setDescription(dataset.getDescription()); + collectionEdge.setFromNodeId(collectionNode.getId()); + collectionEdge.setToNodeId(datasetNode.getId()); + } + lineageService.generateGraph(collectionNode, collectionEdge, datasetNode); + } + + public DatasetLineage getDatasetLineage(String datasetId) { + Dataset dataset = datasetRepository.getById(datasetId); + if (Objects.isNull(dataset)) { + return new DatasetLineage(); + } + LineageNode datasetNode = lineageService.getNodeById(datasetId); + String graphId = datasetNode.getGraphId(); + return new DatasetLineage(lineageService.getNodesByGraphId(graphId), lineageService.getEdgesByGraphId(graphId)); + } + public String getDatasetPvcName() { return sysParamService.getParamByKey(DATASET_PVC_NAME); } @@ -100,11 +144,11 @@ public Dataset updateDataset(String datasetId, UpdateDatasetRequest updateDatase if (Objects.nonNull(updateDatasetRequest.getStatus())) { dataset.setStatus(updateDatasetRequest.getStatus()); } + datasetRepository.updateById(dataset); if (StringUtils.hasText(updateDatasetRequest.getDataSource())) { // 数据源id不为空，使用异步线程进行文件扫盘落库 processDataSourceAsync(dataset.getId(), updateDatasetRequest.getDataSource()); } - datasetRepository.updateById(dataset); return dataset; } @@ -261,6 +305,8 @@ private List getFilePaths(String dataSourceId, Dataset dataset) { log.warn("Fail to get collection task detail, task ID: {}", dataSourceId); return Collections.emptyList(); } + // 记录血缘关系 + addDatasetToGraph(dataset, taskDetail); Path targetPath = Paths.get(taskDetail.getTargetPath()); if (!Files.exists(targetPath) || !Files.isDirectory(targetPath)) { log.warn("Target path not exists or is not a directory: {}", taskDetail.getTargetPath()); diff --git a/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/dto/DatasetLineage.java b/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/dto/DatasetLineage.java new file mode 100644 index 00000000..b2a6e600 --- /dev/null +++ b/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/dto/DatasetLineage.java @@ -0,0 +1,30 @@ +package com.datamate.datamanagement.interfaces.dto; + +import com.datamate.common.domain.model.LineageEdge; +import com.datamate.common.domain.model.LineageNode; +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +import java.util.List; + +/** + * 数据集血缘 + * + * @since 2026/1/23 + */ +@Getter +@Setter +@NoArgsConstructor +@AllArgsConstructor +public class DatasetLineage { + /** + * 节点列表 + */ + private List lineageNodes; + /** + * 边列表 + */ + private List lineageEdges; +} diff --git a/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/rest/DatasetController.java b/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/rest/DatasetController.java index 05b68a3c..36492143 100644 --- a/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/rest/DatasetController.java +++ b/backend/services/data-management-service/src/main/java/com/datamate/datamanagement/interfaces/rest/DatasetController.java @@ -80,6 +80,11 @@ public DatasetResponse updateDataset(@PathVariable("datasetId") String datasetId return DatasetConverter.INSTANCE.convertToResponse(dataset); } + @GetMapping("/{datasetId}/lineage") + public DatasetLineage getDatasetLineage(@PathVariable("datasetId") String datasetId) { + return datasetApplicationService.getDatasetLineage(datasetId); + } + /** * 根据ID删除数据集 * diff --git a/backend/shared/domain-common/src/main/java/com/datamate/common/domain/enums/EdgeType.java b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/enums/EdgeType.java new file mode 100644 index 00000000..928415b7 --- /dev/null +++ b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/enums/EdgeType.java @@ -0,0 +1,14 @@ +package com.datamate.common.domain.enums; + +/** + * 边类型：DATA_COLLECTION/DATA_CLEANING/DATA_LABELING/DATA_SYNTHESIS/DATA_RATIO + * + * @since 2026/1/23 + */ +public enum EdgeType { + DATA_COLLECTION, + DATA_CLEANING, + DATA_LABELING, + DATA_SYNTHESIS, + DATA_RATIO +} diff --git a/backend/shared/domain-common/src/main/java/com/datamate/common/domain/enums/NodeType.java b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/enums/NodeType.java new file mode 100644 index 00000000..9053f84c --- /dev/null +++ b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/enums/NodeType.java @@ -0,0 +1,13 @@ +package com.datamate.common.domain.enums; + +/** + * 节点类型：DATASOURCE/DATASET/KNOWLEDGE_BASE/MODEL等 + * + * @since 2026/1/23 + */ +public enum NodeType { + DATASOURCE, + DATASET, + KNOWLEDGE_BASE, + MODEL +} diff --git a/backend/shared/domain-common/src/main/java/com/datamate/common/domain/model/LineageEdge.java b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/model/LineageEdge.java new file mode 100644 index 00000000..fe0d6563 --- /dev/null +++ b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/model/LineageEdge.java @@ -0,0 +1,65 @@ +package com.datamate.common.domain.model; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import com.datamate.common.domain.enums.EdgeType; +import lombok.Getter; +import lombok.Setter; + +/** + * 数据血缘：边表 + * 边表示处理流程（归集任务、数据清洗、数据标注、数据合成、数据配比等） + * + * @since 2026/1/23 + */ + +@Getter +@Setter +@TableName("t_lineage_edge") +public class LineageEdge { + /** + * 边ID + */ + @TableId(type = IdType.ASSIGN_ID) + private String id; + /** + * 图ID + */ + private String graphId; + + /** + * 处理流程ID + */ + private String processId; + + /** + * 边类型：DATA_COLLECTION/DATA_CLEANING/DATA_LABELING/DATA_SYNTHESIS/DATA_RATIO等 + */ + private EdgeType edgeType; + + /** + * 边名称 + */ + private String name; + + /** + * 边描述 + */ + private String description; + + /** + * 边扩展信息（JSON） + */ + private String edgeMetadata; + + /** + * 源节点ID + */ + private String fromNodeId; + + /** + * 目标节点ID + */ + private String toNodeId; +} diff --git a/backend/shared/domain-common/src/main/java/com/datamate/common/domain/model/LineageNode.java b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/model/LineageNode.java new file mode 100644 index 00000000..0a45ec60 --- /dev/null +++ b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/model/LineageNode.java @@ -0,0 +1,50 @@ +package com.datamate.common.domain.model; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import com.datamate.common.domain.enums.NodeType; +import lombok.Getter; +import lombok.Setter; + +/** + * 数据血缘：节点表 + * 节点表示实体对象（归集来源、数据集、知识库、模型等） + * + * @since 2026/1/23 + */ +@Getter +@Setter +@TableName("t_lineage_node") +public class LineageNode { + /** + * 节点ID + */ + @TableId(type = IdType.ASSIGN_ID) + private String id; + + /** + * 图ID + */ + private String graphId; + + /** + * 节点类型：DATASOURCE/DATASET/KNOWLEDGE_BASE/MODEL等 + */ + private NodeType nodeType; + + /** + * 节点名称 + */ + private String name; + + /** + * 节点描述 + */ + private String description; + + /** + * 节点扩展信息（JSON） + */ + private String nodeMetadata; +} diff --git a/backend/shared/domain-common/src/main/java/com/datamate/common/domain/service/LineageService.java b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/service/LineageService.java new file mode 100644 index 00000000..10026f57 --- /dev/null +++ b/backend/shared/domain-common/src/main/java/com/datamate/common/domain/service/LineageService.java @@ -0,0 +1,168 @@ +package com.datamate.common.domain.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.datamate.common.domain.model.LineageEdge; +import com.datamate.common.domain.model.LineageNode; +import com.datamate.common.infrastructure.mapper.LineageEdgeMapper; +import com.datamate.common.infrastructure.mapper.LineageNodeMapper; +import lombok.RequiredArgsConstructor; +import org.apache.commons.collections4.CollectionUtils; +import org.springframework.stereotype.Component; + +import java.util.Collections; +import java.util.List; +import java.util.UUID; + +/** + * 血缘服务层 + * + * @since 2026/1/23 + */ +@Component +@RequiredArgsConstructor +public class LineageService { + private final LineageEdgeMapper lineageEdgeMapper; + + private final LineageNodeMapper lineageNodeMapper; + + /** + * 生成血缘图 + * + * @param fromNode 源节点 + * @param edge 边 + * @param toNode 目的节点 + */ + public void generateGraph(LineageNode fromNode, LineageEdge edge, LineageNode toNode) { + // 如果 fromNode 为空，直接返回不做任何处理 + if (fromNode == null) { + return; + } + + // 如果 fromNode 有值，检查 fromNode 是否存在 + LineageNode existingFrom = lineageNodeMapper.selectById(fromNode.getId()); + + String fromGraphId; + if (existingFrom == null) { + // fromNode 不存在，创建 fromNode，记录 fromGraphId + fromGraphId = UUID.randomUUID().toString(); + fromNode.setGraphId(fromGraphId); + lineageNodeMapper.insert(fromNode); + // 处理 edge 和 toNode + } else { + // fromNode 存在，记录 fromGraphId + fromGraphId = existingFrom.getGraphId(); + } + // 处理 edge 和 toNode + generateGraph(edge, toNode, fromGraphId); + } + + private void generateGraph(LineageEdge edge, LineageNode toNode, String graphId) { + // 无 edge，直接返回 + if (edge == null) { + return; + } + + // 有 edge，处理 edge，检查是否有 toNode + handleLineageEdge(graphId, edge); + if (toNode == null) { + // 无 toNode，返回 + return; + } + + // 有 toNode，检查 toNode 是否存在 + LineageNode existingTo = lineageNodeMapper.selectById(toNode.getId()); + + if (existingTo == null) { + // toNode 不存在，创建 toNode 后结束 + toNode.setGraphId(graphId); + lineageNodeMapper.insert(toNode); + } else { + // toNode 存在，将 fromNode 所在的图 并入 toNode 所在的图后结束 + mergeGraph(graphId, existingTo.getGraphId()); + } + } + + private void handleLineageEdge(String graphId, LineageEdge edge) { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper() + .eq(LineageEdge::getGraphId, graphId) + .eq(LineageEdge::getFromNodeId, edge.getFromNodeId()) + .eq(LineageEdge::getToNodeId, edge.getToNodeId()); + List lineageEdges = lineageEdgeMapper.selectList(wrapper); + if (CollectionUtils.isEmpty(lineageEdges)) { + edge.setId(UUID.randomUUID().toString()); + edge.setGraphId(graphId); + lineageEdgeMapper.insert(edge); + } else { + edge.setId(lineageEdges.getFirst().getId()); + lineageEdgeMapper.updateById(edge); + } + } + + + + private void mergeGraph(String fromGraphId, String toGraphId) { + if (fromGraphId == null || toGraphId == null || fromGraphId.equals(toGraphId)) { + return; + } + + List fromNodes = getNodesByGraphId(fromGraphId); + List toNodes = getNodesByGraphId(toGraphId); + + // choose smaller graph as source, larger as target + String sourceGraphId = fromNodes.size() <= toNodes.size() ? fromGraphId : toGraphId; + String targetGraphId = sourceGraphId.equals(fromGraphId) ? toGraphId : fromGraphId; + List sourceNodes = sourceGraphId.equals(fromGraphId) ? fromNodes : toNodes; + + // update nodes' graphId + for (LineageNode node : sourceNodes) { + if (node == null) continue; + node.setGraphId(targetGraphId); + lineageNodeMapper.updateById(node); + } + + // update edges' graphId (edges belonging to the source graph) + List edges = getEdgesByGraphId(sourceGraphId); + if (edges == null) { + edges = Collections.emptyList(); + } + for (LineageEdge edge : edges) { + if (edge == null) continue; + edge.setGraphId(targetGraphId); + lineageEdgeMapper.updateById(edge); + } + } + + /** + * 从图ID获取图的节点列表 + * + * @param graphId 图ID + * @return 图的节点列表 + */ + public List getNodesByGraphId(String graphId) { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper() + .eq(LineageNode::getGraphId, graphId); + return lineageNodeMapper.selectList(wrapper); + } + + /** + * 从图ID获取图的边列表 + * + * @param graphId 图ID + * @return 图的边列表 + */ + public List getEdgesByGraphId(String graphId) { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper() + .eq(LineageEdge::getGraphId, graphId); + return lineageEdgeMapper.selectList(wrapper); + } + + /** + * 从节点ID获取节点 + * + * @param nodeId 节点ID + * @return 对应节点 + */ + public LineageNode getNodeById(String nodeId) { + return lineageNodeMapper.selectById(nodeId); + } +} \ No newline at end of file diff --git a/backend/shared/domain-common/src/main/java/com/datamate/common/infrastructure/mapper/LineageEdgeMapper.java b/backend/shared/domain-common/src/main/java/com/datamate/common/infrastructure/mapper/LineageEdgeMapper.java new file mode 100644 index 00000000..8d8f6cd0 --- /dev/null +++ b/backend/shared/domain-common/src/main/java/com/datamate/common/infrastructure/mapper/LineageEdgeMapper.java @@ -0,0 +1,16 @@ +package com.datamate.common.infrastructure.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.datamate.common.domain.model.LineageEdge; +import com.datamate.common.infrastructure.config.IgnoreDataScopeAnnotation; +import org.apache.ibatis.annotations.Mapper; + +/** + * 边映射器接口 + * + * @since 2026/1/23 + */ +@Mapper +@IgnoreDataScopeAnnotation +public interface LineageEdgeMapper extends BaseMapper { +} diff --git a/backend/shared/domain-common/src/main/java/com/datamate/common/infrastructure/mapper/LineageNodeMapper.java b/backend/shared/domain-common/src/main/java/com/datamate/common/infrastructure/mapper/LineageNodeMapper.java new file mode 100644 index 00000000..3f8ec99d --- /dev/null +++ b/backend/shared/domain-common/src/main/java/com/datamate/common/infrastructure/mapper/LineageNodeMapper.java @@ -0,0 +1,16 @@ +package com.datamate.common.infrastructure.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.datamate.common.domain.model.LineageNode; +import com.datamate.common.infrastructure.config.IgnoreDataScopeAnnotation; +import org.apache.ibatis.annotations.Mapper; + +/** + * 节点映射器接口 + * + * @since 2026/1/23 + */ +@Mapper +@IgnoreDataScopeAnnotation +public interface LineageNodeMapper extends BaseMapper { +} diff --git a/runtime/datamate-python/app/db/models/base_entity.py b/runtime/datamate-python/app/db/models/base_entity.py index 9d31203a..56a6aaea 100644 --- a/runtime/datamate-python/app/db/models/base_entity.py +++ b/runtime/datamate-python/app/db/models/base_entity.py @@ -1,4 +1,4 @@ -from sqlalchemy import Column, String, TIMESTAMP +from sqlalchemy import Column, String, TIMESTAMP, Text, JSON from sqlalchemy.orm import declarative_base from sqlalchemy.sql import func @@ -20,3 +20,32 @@ class BaseEntity(Base): # default: do enforce data scope unless subclass sets this to True __ignore_data_scope__ = False + + +class LineageNode(Base): + """数据血缘：节点表（实体对象）""" + + __tablename__ = "t_lineage_node" + + id = Column(String(36), primary_key=True, comment="节点ID") + graph_id = Column(String(36), nullable=True, comment="图ID") + node_type = Column(String(64), nullable=False, comment="节点类型") + name = Column(String(256), nullable=False, comment="节点名称") + description = Column(Text, nullable=True, comment="节点描述") + node_metadata = Column(Text, nullable=True, comment="节点扩展信息（JSON）") + + +class LineageEdge(Base): + """数据血缘：边表（处理流程）""" + + __tablename__ = "t_lineage_edge" + + id = Column(String(36), primary_key=True, comment="边ID") + graph_id = Column(String(36), nullable=True, comment="图ID") + process_id = Column(String(36), nullable=True, comment="处理流程ID") + edge_type = Column(String(64), nullable=False, comment="边类型") + name = Column(String(256), nullable=True, comment="边名称") + description = Column(Text, nullable=True, comment="边描述") + edge_metadata = Column(Text, nullable=True, comment="边扩展信息（JSON）") + from_node_id = Column(String(36), nullable=False, comment="源节点ID") + to_node_id = Column(String(36), nullable=False, comment="目标节点ID") diff --git a/runtime/datamate-python/app/module/collection/service/collection.py b/runtime/datamate-python/app/module/collection/service/collection.py index e2e63128..dc0a674d 100644 --- a/runtime/datamate-python/app/module/collection/service/collection.py +++ b/runtime/datamate-python/app/module/collection/service/collection.py @@ -13,7 +13,9 @@ from app.module.collection.client.datax_client import DataxClient from app.module.collection.schema.collection import SyncMode, create_execute_record from app.module.dataset.service.service import Service -from app.module.shared.schema import TaskStatus +from app.module.shared.schema import TaskStatus, NodeType, EdgeType +from app.module.shared.common.lineage import LineageService +from app.db.models.base_entity import LineageNode, LineageEdge logger = get_logger(__name__) @@ -81,3 +83,55 @@ async def run_async(task_id: str, dataset_id: str = None): if file_path.is_file(): source_paths.append(str(file_path.absolute())) await dataset_service.add_files_to_dataset(dataset_id=dataset_id, source_paths=source_paths) + await CollectionTaskService._add_dataset_to_graph( + session=session, + dataset_id=dataset_id, + task=task + ) + + @staticmethod + async def _add_dataset_to_graph( + session: AsyncSession, + dataset_id: str, + task: CollectionTask + ) -> None: + """ + 在归集完成后，将数据集加入血缘图。 + 参考 Java 侧 addDatasetToGraph 逻辑： + collection(DATASOURCE) -> dataset(DATASET) via DATA_COLLECTION edge + """ + try: + dataset = await session.get(Dataset, dataset_id) + if not dataset: + logger.warning(f"dataset {dataset_id} not found when building lineage graph") + return + + dataset_node = LineageNode( + id=dataset.id, + node_type=NodeType.DATASET.value, + name=dataset.name, + description=dataset.description + ) + + collection_node = LineageNode( + id=task.id, + node_type=NodeType.DATASOURCE.value, + name=task.name, + description=task.description + ) + + collection_edge = LineageEdge( + process_id=task.id, + name=task.name, + edge_type=EdgeType.DATA_COLLECTION.value, + description=dataset.description, + from_node_id=task.id, + to_node_id=dataset.id + ) + + lineage_service = LineageService(db=session) + await lineage_service.generate_graph(collection_node, collection_edge, dataset_node) + await session.commit() + except Exception as exc: + logger.error(f"Failed to add dataset lineage graph: {exc}") + await session.rollback() diff --git a/runtime/datamate-python/app/module/shared/common/lineage.py b/runtime/datamate-python/app/module/shared/common/lineage.py new file mode 100644 index 00000000..c69179ea --- /dev/null +++ b/runtime/datamate-python/app/module/shared/common/lineage.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +from typing import List, Optional +from uuid import uuid4 + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.logging import get_logger +from app.db.models.base_entity import LineageNode, LineageEdge + +log = get_logger(__name__) + + +class LineageService: + """血缘服务（Python 版）""" + + def __init__(self, db: AsyncSession): + self.db = db + + async def generate_graph( + self, + from_node: Optional[LineageNode], + edge: Optional[LineageEdge], + to_node: Optional[LineageNode] + ) -> None: + """ + 生成血缘图 + + Args: + from_node: 源节点 + edge: 边 + to_node: 目的节点 + """ + # 1. 如果 from_node 为空，直接返回不做任何处理 + if from_node is None: + return + + # 2. 如果 from_node 有值，检查 from_node 是否存在 + existing_from = await self.get_node_by_id(from_node.id) if from_node.id else None + + if existing_from is None: + # from_node 不存在，创建 from_node，记录 from_graph_id + from_graph_id = str(uuid4()) + if not from_node.id: + from_node.id = str(uuid4()) + from_node.graph_id = from_graph_id + self.db.add(from_node) + await self.db.flush() + else: + # from_node 存在，记录 from_graph_id + from_graph_id = existing_from.graph_id + + # 处理 edge 和 to_node + await self._generate_graph(edge, to_node, from_graph_id) + + async def _generate_graph( + self, + edge: Optional[LineageEdge], + to_node: Optional[LineageNode], + graph_id: Optional[str] + ) -> None: + # 无 edge，直接返回 + if edge is None: + return + + # 有 edge，处理 edge，检查是否有 to_node + await self._handle_lineage_edge(graph_id, edge) + if to_node is None: + return + + # 有 to_node，检查 to_node 是否存在 + existing_to = await self.get_node_by_id(to_node.id) if to_node.id else None + + if existing_to is None: + # to_node 不存在，创建 to_node 后结束 + if not to_node.id: + to_node.id = str(uuid4()) + to_node.graph_id = graph_id + self.db.add(to_node) + await self.db.flush() + else: + # to_node 存在，将 from_node 所在的图并入 to_node 所在的图后结束 + await self._merge_graph(graph_id, existing_to.graph_id) + + async def _handle_lineage_edge(self, graph_id: Optional[str], edge: LineageEdge) -> None: + stmt = select(LineageEdge).where( + LineageEdge.graph_id == graph_id, + LineageEdge.from_node_id == edge.from_node_id, + LineageEdge.to_node_id == edge.to_node_id + ) + result = await self.db.execute(stmt) + existing = result.scalar_one_or_none() + + if existing is None: + if not edge.id: + edge.id = str(uuid4()) + edge.graph_id = graph_id + self.db.add(edge) + await self.db.flush() + else: + edge.id = existing.id + if edge.graph_id is None: + edge.graph_id = existing.graph_id + # 覆盖更新现有记录 + existing.graph_id = edge.graph_id + existing.process_id = edge.process_id + existing.edge_type = edge.edge_type + existing.name = edge.name + existing.description = edge.description + existing.edge_metadata = edge.edge_metadata + existing.from_node_id = edge.from_node_id + existing.to_node_id = edge.to_node_id + await self.db.flush() + + async def _merge_graph(self, from_graph_id: Optional[str], to_graph_id: Optional[str]) -> None: + if not from_graph_id or not to_graph_id or from_graph_id == to_graph_id: + return + + from_nodes = await self.get_nodes_by_graph_id(from_graph_id) + to_nodes = await self.get_nodes_by_graph_id(to_graph_id) + + # choose smaller graph as source, larger as target + source_graph_id = from_graph_id if len(from_nodes) <= len(to_nodes) else to_graph_id + target_graph_id = to_graph_id if source_graph_id == from_graph_id else from_graph_id + source_nodes = from_nodes if source_graph_id == from_graph_id else to_nodes + + # update nodes' graph_id + for node in source_nodes: + if node is None: + continue + node.graph_id = target_graph_id + self.db.add(node) + + # update edges' graph_id (edges belonging to the source graph) + edges = await self.get_edges_by_graph_id(source_graph_id) + for edge in edges: + if edge is None: + continue + edge.graph_id = target_graph_id + self.db.add(edge) + + await self.db.flush() + + async def get_nodes_by_graph_id(self, graph_id: str) -> List[LineageNode]: + """从图ID获取图的节点列表""" + stmt = select(LineageNode).where(LineageNode.graph_id == graph_id) + result = await self.db.execute(stmt) + return list(result.scalars().all()) + + async def get_edges_by_graph_id(self, graph_id: str) -> List[LineageEdge]: + """从图ID获取图的边列表""" + stmt = select(LineageEdge).where(LineageEdge.graph_id == graph_id) + result = await self.db.execute(stmt) + return list(result.scalars().all()) + + async def get_node_by_id(self, node_id: str) -> Optional[LineageNode]: + """从节点ID获取节点""" + return await self.db.get(LineageNode, node_id) diff --git a/runtime/datamate-python/app/module/shared/schema/__init__.py b/runtime/datamate-python/app/module/shared/schema/__init__.py index f0c2765a..b2c77a28 100644 --- a/runtime/datamate-python/app/module/shared/schema/__init__.py +++ b/runtime/datamate-python/app/module/shared/schema/__init__.py @@ -4,10 +4,16 @@ PaginatedData, TaskStatus ) +from .lineage import ( + NodeType, + EdgeType +) __all__ = [ "BaseResponseModel", "StandardResponse", "PaginatedData", - "TaskStatus" + "TaskStatus", + "NodeType", + "EdgeType" ] diff --git a/runtime/datamate-python/app/module/shared/schema/lineage.py b/runtime/datamate-python/app/module/shared/schema/lineage.py new file mode 100644 index 00000000..45091c57 --- /dev/null +++ b/runtime/datamate-python/app/module/shared/schema/lineage.py @@ -0,0 +1,14 @@ +from enum import Enum + +class NodeType(str, Enum): + DATASOURCE = "DATASOURCE" + DATASET = "DATASET" + KNOWLEDGE_BASE = "KNOWLEDGE_BASE" + MODEL = "MODEL" + +class EdgeType(str, Enum): + DATA_COLLECTION = "DATA_COLLECTION" + DATA_CLEANING = "DATA_CLEANING" + DATA_LABELING = "DATA_LABELING" + DATA_SYNTHESIS = "DATA_SYNTHESIS" + DATA_RATIO = "DATA_RATIO" diff --git a/scripts/db/data-common-init.sql b/scripts/db/data-common-init.sql index 8e13ed55..32c8c729 100644 --- a/scripts/db/data-common-init.sql +++ b/scripts/db/data-common-init.sql @@ -25,3 +25,61 @@ COMMENT ON COLUMN t_chunk_upload_request.check_info IS '业务信息'; -- 如果需要索引可以添加 CREATE INDEX IF NOT EXISTS idx_chunk_upload_service ON t_chunk_upload_request(service_id); + +-- ========================= +-- 数据血缘：节点表 +-- 节点表示实体对象（归集来源、数据集、知识库、模型等） +-- ========================= +CREATE TABLE IF NOT EXISTS t_lineage_node +( + id VARCHAR(36) PRIMARY KEY, + graph_id VARCHAR(36) NOT NULL, + node_type VARCHAR(64) NOT NULL, + name VARCHAR(256) NOT NULL, + description TEXT, + node_metadata TEXT +); + +COMMENT ON TABLE t_lineage_node IS '数据血缘节点表（实体对象）'; +COMMENT ON COLUMN t_lineage_node.id IS '主键ID'; +COMMENT ON COLUMN t_lineage_node.graph_id IS '图ID'; +COMMENT ON COLUMN t_lineage_node.node_type IS '节点类型：DATASOURCE/DATASET/KNOWLEDGE_BASE/MODEL等'; +COMMENT ON COLUMN t_lineage_node.name IS '节点名称'; +COMMENT ON COLUMN t_lineage_node.description IS '节点描述'; +COMMENT ON COLUMN t_lineage_node.node_metadata IS '节点扩展信息（JSON）'; + +CREATE INDEX IF NOT EXISTS idx_lineage_node_id ON t_lineage_node(id); +CREATE INDEX IF NOT EXISTS idx_lineage_graph_id ON t_lineage_node(graph_id); + +-- ========================= +-- 数据血缘：边表 +-- 边表示处理流程（归集任务、数据清洗、数据标注、数据合成、数据配比等） +-- ========================= +CREATE TABLE IF NOT EXISTS t_lineage_edge +( + id VARCHAR(36) PRIMARY KEY, + process_id VARCHAR(36) NOT NULL, + graph_id VARCHAR(36) NOT NULL, + edge_type VARCHAR(64) NOT NULL, + name VARCHAR(256), + description TEXT, + edge_metadata TEXT, + from_node_id VARCHAR(36) NOT NULL, + to_node_id VARCHAR(36) NOT NULL +); + +COMMENT ON TABLE t_lineage_edge IS '数据血缘边表（处理流程）'; +COMMENT ON COLUMN t_lineage_edge.id IS '边ID'; +COMMENT ON COLUMN t_lineage_edge.graph_id IS '图ID'; +COMMENT ON COLUMN t_lineage_edge.process_id IS '处理流程ID'; +COMMENT ON COLUMN t_lineage_edge.edge_type IS '边类型：DATA_COLLECTION/DATA_CLEANING/DATA_LABELING/DATA_SYNTHESIS/DATA_RATIO等'; +COMMENT ON COLUMN t_lineage_edge.name IS '边名称'; +COMMENT ON COLUMN t_lineage_edge.description IS '边描述'; +COMMENT ON COLUMN t_lineage_edge.edge_metadata IS '边扩展信息（JSON）'; +COMMENT ON COLUMN t_lineage_edge.from_node_id IS '源节点ID'; +COMMENT ON COLUMN t_lineage_edge.to_node_id IS '目标节点ID'; + +CREATE INDEX IF NOT EXISTS idx_lineage_process_id ON t_lineage_edge(process_id); +CREATE INDEX IF NOT EXISTS idx_lineage_edge_graph_id ON t_lineage_edge(graph_id); +CREATE INDEX IF NOT EXISTS idx_lineage_edge_from ON t_lineage_edge(from_node_id); +CREATE INDEX IF NOT EXISTS idx_lineage_edge_to ON t_lineage_edge(to_node_id); \ No newline at end of file From 2573377ddd3ca006c57cb2c767cf3f25f5e9bcb2 Mon Sep 17 00:00:00 2001 From: uname <2986773479@qq.com> Date: Wed, 7 Jan 2026 14:59:30 +0800 Subject: [PATCH 2/3] feature: add data lineage page and data quality page --- .../DataManagement/Detail/DatasetDetail.tsx | 520 ++++++------ .../Detail/components/DataLineageFlow.tsx | 607 ++++++++++++- .../Detail/components/DataQuality.tsx | 200 ++++- .../DataManagement/Home/DataManagement.tsx | 797 +++++++++--------- .../pages/DataManagement/dataset.const.tsx | 2 + 5 files changed, 1405 insertions(+), 721 deletions(-) diff --git a/frontend/src/pages/DataManagement/Detail/DatasetDetail.tsx b/frontend/src/pages/DataManagement/Detail/DatasetDetail.tsx index d670c986..cac462d3 100644 --- a/frontend/src/pages/DataManagement/Detail/DatasetDetail.tsx +++ b/frontend/src/pages/DataManagement/Detail/DatasetDetail.tsx @@ -1,260 +1,260 @@ -import { useEffect, useMemo, useState } from "react"; -import { Breadcrumb, App, Tabs } from "antd"; -import { - ReloadOutlined, - DownloadOutlined, - UploadOutlined, - EditOutlined, - DeleteOutlined, -} from "@ant-design/icons"; -import DetailHeader from "@/components/DetailHeader"; -import { mapDataset, datasetTypeMap } from "../dataset.const"; -import type { Dataset } from "@/pages/DataManagement/dataset.model"; -import { Link, useNavigate, useParams } from "react-router"; -import { useFilesOperation } from "./useFilesOperation"; -import { - createDatasetTagUsingPost, - deleteDatasetByIdUsingDelete, - downloadDatasetUsingGet, - queryDatasetByIdUsingGet, - queryDatasetTagsUsingGet, - updateDatasetByIdUsingPut, -} from "../dataset.api"; -import DataQuality from "./components/DataQuality"; -import DataLineageFlow from "./components/DataLineageFlow"; -import Overview from "./components/Overview"; -import { Activity, Clock, File, FileType } from "lucide-react"; -import EditDataset from "../Create/EditDataset"; -import ImportConfiguration from "./components/ImportConfiguration"; - -const tabList = [ - { - key: "overview", - label: "概览", - }, - { - key: "lineage", - label: "数据血缘", - }, - { - key: "quality", - label: "数据质量", - }, -]; - -export default function DatasetDetail() { - const { id } = useParams(); // 获取动态路由参数 - const navigate = useNavigate(); - const [activeTab, setActiveTab] = useState("overview"); - const { message } = App.useApp(); - const [showEditDialog, setShowEditDialog] = useState(false); - - const [dataset, setDataset] = useState({} as Dataset); - const filesOperation = useFilesOperation(dataset); - - const [showUploadDialog, setShowUploadDialog] = useState(false); - const navigateItems = useMemo( - () => [ - { - title: 数据管理, - }, - { - title: dataset.name || "数据集详情", - }, - ], - [dataset] - ); - const fetchDataset = async () => { - const { data } = await queryDatasetByIdUsingGet(id as unknown as number); - setDataset(mapDataset(data)); - }; - - useEffect(() => { - fetchDataset(); - filesOperation.fetchFiles('', 1, 10); // 从根目录开始，第一页 - }, []); - - const handleRefresh = async (showMessage = true, prefixOverride?: string) => { - fetchDataset(); - // 刷新当前目录，保持在当前页 - const targetPrefix = - prefixOverride !== undefined - ? prefixOverride - : filesOperation.pagination.prefix; - filesOperation.fetchFiles( - targetPrefix, - filesOperation.pagination.current, - filesOperation.pagination.pageSize - ); - if (showMessage) message.success({ content: "数据刷新成功" }); - }; - - const handleDownload = async () => { - await downloadDatasetUsingGet(dataset.id); - message.success("文件下载成功"); - }; - - const handleDeleteDataset = async () => { - await deleteDatasetByIdUsingDelete(dataset.id); - navigate("/data/management"); - message.success("数据集删除成功"); - }; - - useEffect(() => { - const refreshData = (e: Event) => { - const custom = e as CustomEvent<{ prefix?: string }>; - const prefixOverride = custom.detail?.prefix; - handleRefresh(false, prefixOverride); - }; - window.addEventListener("update:dataset", refreshData as EventListener); - return () => { - window.removeEventListener( - "update:dataset", - refreshData as EventListener - ); - }; - }, []); - - // 基本信息描述项 - const statistics = [ - { - icon: , - key: "file", - value: dataset?.fileCount || 0, - }, - { - icon: , - key: "size", - value: dataset?.size || "0 B", - }, - { - icon: , - key: "type", - value: - datasetTypeMap[dataset?.datasetType as keyof typeof datasetTypeMap] - ?.label || - dataset?.type || - "未知", - }, - { - icon: , - key: "time", - value: dataset?.updatedAt, - }, - ]; - - // 数据集操作列表 - const operations = [ - { - key: "edit", - label: "编辑", - icon: , - onClick: () => { - setShowEditDialog(true); - }, - }, - - { - key: "upload", - label: "导入数据", - icon: , - onClick: () => setShowUploadDialog(true), - }, - { - key: "export", - label: "导出", - icon: , - // isDropdown: true, - // items: [ - // { key: "alpaca", label: "Alpaca 格式", icon: }, - // { key: "jsonl", label: "JSONL 格式", icon: }, - // { key: "csv", label: "CSV 格式", icon: }, - // { key: "coco", label: "COCO 格式", icon: }, - // ], - onClick: () => handleDownload(), - }, - { - key: "refresh", - label: "刷新", - icon: , - onClick: handleRefresh, - }, - { - key: "delete", - label: "删除", - danger: true, - confirm: { - title: "确认删除该数据集？", - description: "删除后该数据集将无法恢复，请谨慎操作。", - okText: "删除", - cancelText: "取消", - okType: "danger", - }, - icon: , - onClick: handleDeleteDataset, - }, - ]; - - return ( -
- - {/* Header */} - { - const res = await queryDatasetTagsUsingGet({ - page: 0, - pageSize: 1000, - }); - return res.data || []; - }, - onCreateAndTag: async (tagName) => { - const res = await createDatasetTagUsingPost({ name: tagName }); - if (res.data) { - await updateDatasetByIdUsingPut(dataset.id, { - tags: [...dataset.tags.map((tag) => tag.name), res.data.name], - }); - handleRefresh(); - } - }, - onAddTag: async (tag) => { - const res = await updateDatasetByIdUsingPut(dataset.id, { - tags: [...dataset.tags.map((tag) => tag.name), tag], - }); - if (res.data) { - handleRefresh(); - } - }, - }} - /> -
- -
- {activeTab === "overview" && ( - - )} - {activeTab === "lineage" && } - {activeTab === "quality" && } -
-
- setShowUploadDialog(false)} - prefix={filesOperation.pagination.prefix} - updateEvent="update:dataset" - /> - setShowEditDialog(false)} - onRefresh={handleRefresh} - /> -
- ); -} +import { useEffect, useMemo, useState } from "react"; +import { Breadcrumb, App, Tabs } from "antd"; +import { + ReloadOutlined, + DownloadOutlined, + UploadOutlined, + EditOutlined, + DeleteOutlined, +} from "@ant-design/icons"; +import DetailHeader from "@/components/DetailHeader"; +import { mapDataset, datasetTypeMap } from "../dataset.const"; +import type { Dataset } from "@/pages/DataManagement/dataset.model"; +import { Link, useNavigate, useParams } from "react-router"; +import { useFilesOperation } from "./useFilesOperation"; +import { + createDatasetTagUsingPost, + deleteDatasetByIdUsingDelete, + downloadDatasetUsingGet, + queryDatasetByIdUsingGet, + queryDatasetTagsUsingGet, + updateDatasetByIdUsingPut, +} from "../dataset.api"; +import DataQuality from "./components/DataQuality"; +import DataLineageFlow from "./components/DataLineageFlow"; +import Overview from "./components/Overview"; +import { Activity, Clock, File, FileType } from "lucide-react"; +import EditDataset from "../Create/EditDataset"; +import ImportConfiguration from "./components/ImportConfiguration"; + +const tabList = [ + { + key: "overview", + label: "概览", + }, + { + key: "lineage", + label: "数据血缘", + }, + { + key: "quality", + label: "数据质量", + }, +]; + +export default function DatasetDetail() { + const { id } = useParams(); // 获取动态路由参数 + const navigate = useNavigate(); + const [activeTab, setActiveTab] = useState("overview"); + const { message } = App.useApp(); + const [showEditDialog, setShowEditDialog] = useState(false); + + const [dataset, setDataset] = useState({} as Dataset); + const filesOperation = useFilesOperation(dataset); + + const [showUploadDialog, setShowUploadDialog] = useState(false); + const navigateItems = useMemo( + () => [ + { + title: 数据管理, + }, + { + title: dataset.name || "数据集详情", + }, + ], + [dataset] + ); + const fetchDataset = async () => { + const { data } = await queryDatasetByIdUsingGet(id as unknown as number); + setDataset(mapDataset(data)); + }; + + useEffect(() => { + fetchDataset(); + filesOperation.fetchFiles('', 1, 10); // 从根目录开始，第一页 + }, []); + + const handleRefresh = async (showMessage = true, prefixOverride?: string) => { + fetchDataset(); + // 刷新当前目录，保持在当前页 + const targetPrefix = + prefixOverride !== undefined + ? prefixOverride + : filesOperation.pagination.prefix; + filesOperation.fetchFiles( + targetPrefix, + filesOperation.pagination.current, + filesOperation.pagination.pageSize + ); + if (showMessage) message.success({ content: "数据刷新成功" }); + }; + + const handleDownload = async () => { + await downloadDatasetUsingGet(dataset.id); + message.success("文件下载成功"); + }; + + const handleDeleteDataset = async () => { + await deleteDatasetByIdUsingDelete(dataset.id); + navigate("/data/management"); + message.success("数据集删除成功"); + }; + + useEffect(() => { + const refreshData = (e: Event) => { + const custom = e as CustomEvent<{ prefix?: string }>; + const prefixOverride = custom.detail?.prefix; + handleRefresh(false, prefixOverride); + }; + window.addEventListener("update:dataset", refreshData as EventListener); + return () => { + window.removeEventListener( + "update:dataset", + refreshData as EventListener + ); + }; + }, []); + + // 基本信息描述项 + const statistics = [ + { + icon: , + key: "file", + value: dataset?.fileCount || 0, + }, + { + icon: , + key: "size", + value: dataset?.size || "0 B", + }, + { + icon: , + key: "type", + value: + datasetTypeMap[dataset?.datasetType as keyof typeof datasetTypeMap] + ?.label || + dataset?.type || + "未知", + }, + { + icon: , + key: "time", + value: dataset?.updatedAt, + }, + ]; + + // 数据集操作列表 + const operations = [ + { + key: "edit", + label: "编辑", + icon: , + onClick: () => { + setShowEditDialog(true); + }, + }, + + { + key: "upload", + label: "导入数据", + icon: , + onClick: () => setShowUploadDialog(true), + }, + { + key: "export", + label: "导出", + icon: , + // isDropdown: true, + // items: [ + // { key: "alpaca", label: "Alpaca 格式", icon: }, + // { key: "jsonl", label: "JSONL 格式", icon: }, + // { key: "csv", label: "CSV 格式", icon: }, + // { key: "coco", label: "COCO 格式", icon: }, + // ], + onClick: () => handleDownload(), + }, + { + key: "refresh", + label: "刷新", + icon: , + onClick: handleRefresh, + }, + { + key: "delete", + label: "删除", + danger: true, + confirm: { + title: "确认删除该数据集？", + description: "删除后该数据集将无法恢复，请谨慎操作。", + okText: "删除", + cancelText: "取消", + okType: "danger", + }, + icon: , + onClick: handleDeleteDataset, + }, + ]; + + return ( +
+ + {/* Header */} + { + const res = await queryDatasetTagsUsingGet({ + page: 0, + pageSize: 1000, + }); + return res.data || []; + }, + onCreateAndTag: async (tagName) => { + const res = await createDatasetTagUsingPost({ name: tagName }); + if (res.data) { + await updateDatasetByIdUsingPut(dataset.id, { + tags: [...dataset.tags.map((tag) => tag.name), res.data.name], + }); + handleRefresh(); + } + }, + onAddTag: async (tag) => { + const res = await updateDatasetByIdUsingPut(dataset.id, { + tags: [...dataset.tags.map((tag) => tag.name), tag], + }); + if (res.data) { + handleRefresh(); + } + }, + }} + /> +
+ +
+ {activeTab === "overview" && ( + + )} + {activeTab === "lineage" && } + {activeTab === "quality" && } +
+
+ setShowUploadDialog(false)} + prefix={filesOperation.pagination.prefix} + updateEvent="update:dataset" + /> + setShowEditDialog(false)} + onRefresh={handleRefresh} + /> +
+ ); +} diff --git a/frontend/src/pages/DataManagement/Detail/components/DataLineageFlow.tsx b/frontend/src/pages/DataManagement/Detail/components/DataLineageFlow.tsx index fc8d8f0b..0d3e11a4 100644 --- a/frontend/src/pages/DataManagement/Detail/components/DataLineageFlow.tsx +++ b/frontend/src/pages/DataManagement/Detail/components/DataLineageFlow.tsx @@ -1,52 +1,579 @@ -import DevelopmentInProgress from "@/components/DevelopmentInProgress"; -import { Dataset } from "../../dataset.model"; +import type React from "react" +import { useState, useRef, useEffect } from "react" +import { Card, Badge } from "antd" +import { Database, Table, Brain, BookOpen, X } from "lucide-react" +import type { Dataset } from "@/pages/DataManagement/dataset.model.ts" + +interface Node { + id: string + type: "datasource" | "dataset" | "model" | "knowledge" + label: string + x: number + y: number + description?: string + status?: string + fileCount?: number + size?: string + updateTime?: string +} + +interface Edge { + from: string + to: string + label: string +} + +const nodes: Node[] = [ + { + id: "source1", + type: "datasource", + label: "MySQL 数据库", + x: 80, + y: 100, + description: "业务数据库", + status: "运转", + updateTime: "2026-01-06 15:22:08", + }, + { + id: "source2", + type: "datasource", + label: "API 接口", + x: 80, + y: 250, + description: "外部数据源", + status: "运转", + updateTime: "2026-01-06 14:30:15", + }, + { + id: "source3", + type: "datasource", + label: "日志文件", + x: 80, + y: 400, + description: "系统日志", + status: "运转", + updateTime: "2026-01-06 16:05:42", + }, + { + id: "dataset1", + type: "dataset", + label: "原始数据集", + x: 380, + y: 100, + description: "未处理的原始数据", + fileCount: 14, + size: "3.449 MB", + updateTime: "2026-01-06 15:22:08", + }, + { + id: "dataset2", + type: "dataset", + label: "清洗数据集", + x: 680, + y: 175, + description: "清洗后的干净数据", + fileCount: 8, + size: "2.156 MB", + updateTime: "2026-01-06 15:45:20", + }, + { + id: "dataset3", + type: "dataset", + label: "合成数据集", + x: 980, + y: 250, + description: "特征工程后的数据", + fileCount: 5, + size: "1.823 MB", + updateTime: "2026-01-06 16:10:35", + }, + { + id: "model1", + type: "model", + label: "预测模型", + x: 1280, + y: 150, + description: "ML 预测模型", + status: "训练中", + updateTime: "2026-01-06 16:30:12", + }, + { + id: "model2", + type: "model", + label: "分类模型", + x: 1280, + y: 350, + description: "分类算法模型", + status: "已完成", + updateTime: "2026-01-06 16:25:48", + }, + { + id: "kb1", + type: "knowledge", + label: "业务知识库", + x: 1600, + y: 250, + description: "结构化知识存储", + fileCount: 32, + size: "8.742 MB", + updateTime: "2026-01-06 16:45:05", + }, +] + +const edges: Edge[] = [ + { from: "source1", to: "dataset1", label: "数据归集" }, + { from: "source2", to: "dataset1", label: "数据归集" }, + { from: "source3", to: "dataset1", label: "数据归集" }, + { from: "dataset1", to: "dataset2", label: "数据清洗" }, + { from: "dataset2", to: "dataset3", label: "数据合成" }, + { from: "dataset2", to: "kb1", label: "知识生成" }, + { from: "dataset3", to: "model1", label: "模型训练" }, + { from: "dataset3", to: "model2", label: "模型训练" }, +] + +const nodeConfig = { + datasource: { + icon: Database, + color: "oklch(0.5 0.2 250)", + bgColor: "oklch(0.92 0.05 250)", + borderColor: "oklch(0.7 0.15 250)", + }, + dataset: { + icon: Table, + color: "oklch(0.5 0.18 200)", + bgColor: "oklch(0.92 0.05 200)", + borderColor: "oklch(0.7 0.15 200)", + }, + model: { + icon: Brain, + color: "oklch(0.5 0.18 320)", + bgColor: "oklch(0.92 0.05 320)", + borderColor: "oklch(0.7 0.15 320)", + }, + knowledge: { + icon: BookOpen, + color: "oklch(0.5 0.18 140)", + bgColor: "oklch(0.92 0.05 140)", + borderColor: "oklch(0.7 0.15 140)", + }, +} export default function DataLineageFlow(dataset: Dataset) { - return - const lineage = dataset.lineage; - if (!lineage) return null; - - const steps = [ - { name: "数据源", value: lineage.source, icon: Database }, - ...lineage.processing.map((step, index) => ({ - name: `处理${index + 1}`, - value: step, - icon: GitBranch, - })), - ]; - - if (lineage.training) { - steps.push({ - name: "模型训练", - value: `${lineage.training.model} (准确率: ${lineage.training.accuracy}%)`, - icon: Target, - }); + const [selectedNode, setSelectedNode] = useState(null) + const [hoveredNode, setHoveredNode] = useState(null) + const [draggedNode, setDraggedNode] = useState(null) + const [dragOffset, setDragOffset] = useState({ x: 0, y: 0 }) + const [renderTrigger, setRenderTrigger] = useState(0) + const canvasRef = useRef(null) + const containerRef = useRef(null) + + useEffect(() => { + const canvas = canvasRef.current + if (!canvas) return + + const ctx = canvas.getContext("2d") + if (!ctx) return + + const rect = canvas.getBoundingClientRect() + canvas.width = rect.width * window.devicePixelRatio + canvas.height = rect.height * window.devicePixelRatio + ctx.scale(window.devicePixelRatio, window.devicePixelRatio) + + ctx.clearRect(0, 0, rect.width, rect.height) + + edges.forEach((edge) => { + const fromNode = nodes.find((n) => n.id === edge.from) + const toNode = nodes.find((n) => n.id === edge.to) + if (!fromNode || !toNode) return + + const isHighlighted = + hoveredNode === edge.from || + hoveredNode === edge.to || + selectedNode?.id === edge.from || + selectedNode?.id === edge.to + + const startX = fromNode.x + 140 + const startY = fromNode.y + 35 + const endX = toNode.x + const endY = toNode.y + 35 + + const controlPointOffset = Math.abs(endX - startX) * 0.4 + const cp1x = startX + controlPointOffset + const cp1y = startY + const cp2x = endX - controlPointOffset + const cp2y = endY + + ctx.beginPath() + ctx.moveTo(startX, startY) + ctx.bezierCurveTo(cp1x, cp1y, cp2x, cp2y, endX, endY) + + const gradient = ctx.createLinearGradient(startX, startY, endX, endY) + const fromConfig = nodeConfig[fromNode.type] + const toConfig = nodeConfig[toNode.type] + + if (isHighlighted) { + gradient.addColorStop(0, fromConfig.color) + gradient.addColorStop(1, toConfig.color) + ctx.strokeStyle = gradient + ctx.lineWidth = 3 + } else { + ctx.strokeStyle = "oklch(0.85 0.03 250)" + ctx.lineWidth = 2 + } + + if (isHighlighted) { + ctx.setLineDash([]) + } else { + ctx.setLineDash([5, 3]) + } + + ctx.stroke() + ctx.setLineDash([]) + + const arrowSize = isHighlighted ? 10 : 8 + const angle = Math.atan2(endY - cp2y, endX - cp2x) + + ctx.beginPath() + ctx.moveTo(endX, endY) + ctx.lineTo(endX - arrowSize * Math.cos(angle - Math.PI / 6), endY - arrowSize * Math.sin(angle - Math.PI / 6)) + ctx.lineTo(endX - arrowSize * Math.cos(angle + Math.PI / 6), endY - arrowSize * Math.sin(angle + Math.PI / 6)) + ctx.closePath() + ctx.fillStyle = isHighlighted ? toConfig.color : "oklch(0.85 0.03 250)" + ctx.fill() + + const t = 0.5 + const midX = + Math.pow(1 - t, 3) * startX + + 3 * Math.pow(1 - t, 2) * t * cp1x + + 3 * (1 - t) * Math.pow(t, 2) * cp2x + + Math.pow(t, 3) * endX + const midY = + Math.pow(1 - t, 3) * startY + + 3 * Math.pow(1 - t, 2) * t * cp1y + + 3 * (1 - t) * Math.pow(t, 2) * cp2y + + Math.pow(t, 3) * endY + + const padding = 6 + const textWidth = ctx.measureText(edge.label).width + ctx.fillStyle = "oklch(1 0 0)" + ctx.shadowColor = "rgba(0, 0, 0, 0.1)" + ctx.shadowBlur = 4 + ctx.shadowOffsetY = 1 + ctx.beginPath() + ctx.roundRect(midX - textWidth / 2 - padding, midY - 8, textWidth + padding * 2, 16, 4) + ctx.fill() + ctx.shadowBlur = 0 + ctx.shadowOffsetY = 0 + + ctx.fillStyle = isHighlighted ? fromConfig.color : "oklch(0.5 0.05 250)" + ctx.font = "600 11px Geist" + ctx.textAlign = "center" + ctx.textBaseline = "middle" + ctx.fillText(edge.label, midX, midY) + }) + }, [hoveredNode, renderTrigger, selectedNode]) + + useEffect(() => { + const handleMouseMove = (e: MouseEvent) => { + if (!draggedNode || !containerRef.current) return + + const container = containerRef.current + const rect = container.getBoundingClientRect() + const x = e.clientX - rect.left - dragOffset.x + const y = e.clientY - rect.top - dragOffset.y + + const nodeIndex = nodes.findIndex((n) => n.id === draggedNode) + if (nodeIndex !== -1) { + nodes[nodeIndex].x = Math.max(0, Math.min(x, rect.width - 120)) + nodes[nodeIndex].y = Math.max(0, Math.min(y, rect.height - 70)) + setRenderTrigger((prev) => prev + 1) + } + } + + const handleMouseUp = () => { + setDraggedNode(null) + } + + if (draggedNode) { + document.addEventListener("mousemove", handleMouseMove) + document.addEventListener("mouseup", handleMouseUp) + } + + return () => { + document.removeEventListener("mousemove", handleMouseMove) + document.removeEventListener("mouseup", handleMouseUp) + } + }, [draggedNode, dragOffset]) + + const handleNodeClick = (node: Node) => { + setSelectedNode(node) + } + + const handleNodeMouseDown = (e: React.MouseEvent, node: Node) => { + e.stopPropagation() + if (!containerRef.current) return + + const container = containerRef.current + const rect = container.getBoundingClientRect() + const offsetX = e.clientX - rect.left - node.x + const offsetY = e.clientY - rect.top - node.y + + setDragOffset({ x: offsetX, y: offsetY }) + setDraggedNode(node.id) + } + + const getRelatedNodes = (nodeId: string): string[] => { + const related = new Set() + edges.forEach((edge) => { + if (edge.from === nodeId) related.add(edge.to) + if (edge.to === nodeId) related.add(edge.from) + }) + return Array.from(related) } return ( -
-
- {steps.map((step, index) => ( -
-
-
- +
+ +
+ + + {nodes.map((node) => { + const config = nodeConfig[node.type] + const Icon = config.icon + const isSelected = selectedNode?.id === node.id + const isHovered = hoveredNode === node.id + const relatedNodes = selectedNode ? getRelatedNodes(selectedNode.id) : [] + const isRelated = selectedNode && relatedNodes.includes(node.id) + const isDimmed = selectedNode && selectedNode.id !== node.id && !isRelated + + return ( +
handleNodeClick(node)} + onMouseDown={(e) => handleNodeMouseDown(e, node)} + onMouseEnter={() => setHoveredNode(node.id)} + onMouseLeave={() => setHoveredNode(null)} + > +
+
+ +
+ +
+ +
+
{node.label}
+ {node.status && ( +
+
+ {node.status} +
+ )} + {node.fileCount !== undefined && ( +
+ {node.fileCount} 个文件 · {node.size} +
+ )} +
+
- {index < steps.length - 1 && ( -
- )} + ) + })} +
+ + + {selectedNode && ( + +
+
+
+
+ {(() => { + const Icon = nodeConfig[selectedNode.type].icon + return + })()} +
+
+

{selectedNode.label}

+ + {selectedNode.type === "datasource" && "数据源"} + {selectedNode.type === "dataset" && "数据集"} + {selectedNode.type === "model" && "模型"} + {selectedNode.type === "knowledge" && "知识库"} + +
+
+
-
-
-
- {step.name} -
-

{step.value}

+ +
+
+

基本信息

+
+
+ ID: + {selectedNode.id} +
+
+ 名称: + {selectedNode.label} +
+ {selectedNode.status && ( +
+ 状态: + +
+ {selectedNode.status} + +
+ )} + {selectedNode.fileCount !== undefined && ( +
+ 文件数: + {selectedNode.fileCount} +
+ )} + {selectedNode.size && ( +
+ 数据大小: + {selectedNode.size} +
+ )} + {selectedNode.updateTime && ( +
+ 更新时间: + {selectedNode.updateTime} +
+ )} +
+
+ +
+

描述

+

{selectedNode.description}

+
+ +
+

上游依赖

+
+ {edges + .filter((e) => e.to === selectedNode.id) + .map((e) => { + const fromNode = nodes.find((n) => n.id === e.from) + return fromNode ? ( +
+
+ {fromNode.label} +
+ ) : null + })} + {edges.filter((e) => e.to === selectedNode.id).length === 0 && ( +

无上游依赖

+ )} +
+
+ +
+

下游影响

+
+ {edges + .filter((e) => e.from === selectedNode.id) + .map((e) => { + const toNode = nodes.find((n) => n.id === e.to) + return toNode ? ( +
+
+ {toNode.label} +
+ ) : null + })} + {edges.filter((e) => e.from === selectedNode.id).length === 0 && ( +

无下游影响

+ )} +
- ))} -
+ + )}
- ); + ) } diff --git a/frontend/src/pages/DataManagement/Detail/components/DataQuality.tsx b/frontend/src/pages/DataManagement/Detail/components/DataQuality.tsx index ab0f1e35..17208d4a 100644 --- a/frontend/src/pages/DataManagement/Detail/components/DataQuality.tsx +++ b/frontend/src/pages/DataManagement/Detail/components/DataQuality.tsx @@ -1,18 +1,178 @@ -import DevelopmentInProgress from "@/components/DevelopmentInProgress"; +// typescript +// File: `frontend/src/pages/DataManagement/Detail/components/DataQuality.tsx` +import React from "react"; +// Run `npm install antd lucide-react` if your editor reports "Module is not installed" import { Card } from "antd"; import { AlertTriangle } from "lucide-react"; +import DevelopmentInProgress from "@/components/DevelopmentInProgress"; +import { Dataset } from "@/pages/DataManagement/dataset.model.ts"; + +type DatasetType = "image" | "text" | "tabular"; + +interface FileStats { + totalFiles: number; + corrupted?: number; + unlabeled?: number; + lowQuality?: number; + missingFields?: number; + duplicateRows?: number; +} + +interface Props { + dataset?: Dataset; + datasetType?: DatasetType; + fileStats?: FileStats; +} + +function clamp(v: number, min = 0, max = 100) { + return Math.max(min, Math.min(max, v)); +} + +function randInt(min: number, max: number) { + return Math.floor(Math.random() * (max - min + 1)) + min; +} + +function getMockMetrics(datasetType: DatasetType, stats: FileStats) { + const total = Math.max(1, stats.totalFiles || 1); + const corrupted = stats.corrupted || 0; + const unlabeled = stats.unlabeled || 0; + const lowQuality = stats.lowQuality || 0; + const missingFields = stats.missingFields || 0; + const duplicateRows = stats.duplicateRows || 0; + + if (datasetType === "image") { + const clarity = clamp(100 - (lowQuality / total) * 120 - (corrupted / total) * 100); + const colorConsistency = clamp(100 - (lowQuality / total) * 80); + const annotationCompleteness = clamp(100 - (unlabeled / total) * 150 - (corrupted / total) * 50); + return [ + { metric: "图像清晰度", value: Math.round(clarity * 10) / 10, color: "bg-green-500" }, + { metric: "色彩一致性", value: Math.round(colorConsistency * 10) / 10, color: "bg-blue-500" }, + { metric: "标注完整性", value: Math.round(annotationCompleteness * 10) / 10, color: "bg-purple-500" }, + ]; + } + + if (datasetType === "text") { + const tokenQuality = clamp(100 - (corrupted / total) * 90 - (missingFields / total) * 60); + const labelConsistency = clamp(100 - (unlabeled / total) * 140 - (corrupted / total) * 40); + const metadataCompleteness = clamp(100 - (missingFields / total) * 150); + return [ + { metric: "分词/Token质量", value: Math.round(tokenQuality * 10) / 10, color: "bg-green-500" }, + { metric: "标签一致性", value: Math.round(labelConsistency * 10) / 10, color: "bg-blue-500" }, + { metric: "元数据完整性", value: Math.round(metadataCompleteness * 10) / 10, color: "bg-purple-500" }, + ]; + } + + // tabular + const missingValueScore = clamp(100 - (missingFields / total) * 200 - (corrupted / total) * 50); + const typeConsistency = clamp(100 - (corrupted / total) * 120 - (duplicateRows / total) * 40); + const uniqueness = clamp(100 - (duplicateRows / total) * 200); + return [ + { metric: "缺失值比例控制", value: Math.round(missingValueScore * 10) / 10, color: "bg-green-500" }, + { metric: "类型一致性", value: Math.round(typeConsistency * 10) / 10, color: "bg-blue-500" }, + { metric: "唯一性/去重", value: Math.round(uniqueness * 10) / 10, color: "bg-purple-500" }, + ]; +} + +export default function DataQuality(props: Props = {}) { + const { dataset, datasetType: propDatasetType, fileStats: propFileStats } = props; + + // Prefer dataset fields when available, then explicit props, then sensible defaults + const inferredTypeFromDataset = (dataset && ((dataset as any).type || (dataset as any).datasetType)) as DatasetType | undefined; + const datasetType: DatasetType = (propDatasetType || inferredTypeFromDataset || "image") as DatasetType; + + // Try to obtain file stats from dataset if provided + let fileStatsFromSource: FileStats | undefined = propFileStats; + let detailedFieldsProvided = false; // track whether detailed fields exist (not defaulted) + + if (!fileStatsFromSource && dataset) { + if ((dataset as any).fileStats) { + fileStatsFromSource = (dataset as any).fileStats as FileStats; + // consider detailed if any field beyond totalFiles present + const fs = fileStatsFromSource as any; + detailedFieldsProvided = fs.corrupted !== undefined || fs.unlabeled !== undefined || fs.lowQuality !== undefined || fs.missingFields !== undefined || fs.duplicateRows !== undefined; + } else { + // attempt to infer total files from common fields + let total = 0; + const dsAny = dataset as any; + if (typeof dsAny.files === "number") total = dsAny.files; + else if (Array.isArray(dsAny.files)) total = dsAny.files.length; + else if (typeof dsAny.fileCount === "number") total = dsAny.fileCount; + + fileStatsFromSource = { + totalFiles: Math.max(1, total || 1), + corrupted: dsAny.corrupted !== undefined ? dsAny.corrupted : undefined, + unlabeled: dsAny.unlabeled !== undefined ? dsAny.unlabeled : undefined, + lowQuality: dsAny.lowQuality !== undefined ? dsAny.lowQuality : undefined, + missingFields: dsAny.missingFields !== undefined ? dsAny.missingFields : undefined, + duplicateRows: dsAny.duplicateRows !== undefined ? dsAny.duplicateRows : undefined, + }; + detailedFieldsProvided = !!(dsAny.corrupted || dsAny.unlabeled || dsAny.lowQuality || dsAny.missingFields || dsAny.duplicateRows); + } + } + + // if props provided, check if they included detailed fields + if (propFileStats) { + fileStatsFromSource = propFileStats; + const p = propFileStats as any; + detailedFieldsProvided = p.corrupted !== undefined || p.unlabeled !== undefined || p.lowQuality !== undefined || p.missingFields !== undefined || p.duplicateRows !== undefined; + } + + // final fallback defaults (note: these are complete defaults) + const finalFileStats: FileStats = fileStatsFromSource ?? { totalFiles: 120, corrupted: 3, unlabeled: 6, lowQuality: 5, missingFields: 0, duplicateRows: 0 }; + // if we landed on fallback defaults, mark detailedFieldsProvided = false so we apply jitter + const completeSource = detailedFieldsProvided || !!fileStatsFromSource; + + // compute metrics once and apply jitter if data incomplete + const { metrics, integrityMetrics } = React.useMemo(() => { + const baseMetrics = getMockMetrics(datasetType, finalFileStats); + + const baseIntegrity = + datasetType === "image" + ? [ + { metric: "文件完整性", value: clamp(100 - ((finalFileStats.corrupted || 0) / Math.max(1, finalFileStats.totalFiles)) * 100), color: "bg-green-500" }, + { metric: "元数据完整性", value: clamp(100 - ((finalFileStats.missingFields || 0) / Math.max(1, finalFileStats.totalFiles)) * 100), color: "bg-blue-500" }, + { metric: "标签一致性", value: clamp(100 - ((finalFileStats.unlabeled || 0) / Math.max(1, finalFileStats.totalFiles)) * 120), color: "bg-purple-500" }, + ] + : datasetType === "text" + ? [ + { metric: "文件完整性", value: clamp(100 - ((finalFileStats.corrupted || 0) / Math.max(1, finalFileStats.totalFiles)) * 100), color: "bg-green-500" }, + { metric: "字段完整性", value: clamp(100 - ((finalFileStats.missingFields || 0) / Math.max(1, finalFileStats.totalFiles)) * 120), color: "bg-blue-500" }, + { metric: "标签一致性", value: clamp(100 - ((finalFileStats.unlabeled || 0) / Math.max(1, finalFileStats.totalFiles)) * 120), color: "bg-purple-500" }, + ] + : [ + { metric: "文件完整性", value: clamp(100 - ((finalFileStats.corrupted || 0) / Math.max(1, finalFileStats.totalFiles)) * 100), color: "bg-green-500" }, + { metric: "列完整性", value: clamp(100 - ((finalFileStats.missingFields || 0) / Math.max(1, finalFileStats.totalFiles)) * 120), color: "bg-blue-500" }, + { metric: "重复率", value: clamp(100 - ((finalFileStats.duplicateRows || 0) / Math.max(1, finalFileStats.totalFiles)) * 200), color: "bg-purple-500" }, + ]; + + // if source data is incomplete or only totalFiles known, apply a small random reduction so values are not all 100% + if (!completeSource) { + // jitter range can be tuned; using 4-12% to make results realistic but not drastic + const jitterMax = 12; + const jitterMin = 4; + + const jittered = baseMetrics.map((m) => { + // don't reduce below 40 for readability + const jitter = randInt(jitterMin, jitterMax); + return { ...m, value: clamp(Math.round((m.value - jitter) * 10) / 10) }; + }); + + const integrityJittered = baseIntegrity.map((m) => { + const jitter = randInt(jitterMin, jitterMax); + return { ...m, value: clamp(Math.round((m.value - jitter) * 10) / 10) }; + }); + + return { metrics: jittered, integrityMetrics: integrityJittered }; + } + + return { metrics: baseMetrics, integrityMetrics: baseIntegrity }; + }, [datasetType, finalFileStats, completeSource]); -export default function DataQuality() { - return return ( -
+
- {[ - { metric: "图像清晰度", value: 96.2, color: "bg-green-500" }, - { metric: "色彩一致性", value: 94.8, color: "bg-blue-500" }, - { metric: "标注完整性", value: 98.1, color: "bg-purple-500" }, - ].map((item, index) => ( + {metrics.map((item, index) => (
{item.metric} @@ -22,18 +182,14 @@ export default function DataQuality() {
+ />
))} - {[ - { metric: "文件完整性", value: 99.7, color: "bg-green-500" }, - { metric: "元数据完整性", value: 97.3, color: "bg-blue-500" }, - { metric: "标签一致性", value: 95.6, color: "bg-purple-500" }, - ].map((item, index) => ( + {integrityMetrics.map((item, index) => (
{item.metric} @@ -43,7 +199,7 @@ export default function DataQuality() {
+ />
))} @@ -57,16 +213,16 @@ export default function DataQuality() {

质量改进建议

  • - - 建议对42张图像进行重新标注以提高准确性 + + 建议对{Math.max(1, Math.round((finalFileStats.lowQuality || 0) * 1))}项低质量样本进行复查或重新采集
  • - - 检查并补充缺失的病理分级信息 + + 检查并补充缺失的元数据字段（现有缺失：{finalFileStats.missingFields || 0}）
  • - - 考虑增加更多低分化样本以平衡数据分布 + + 考虑增加更多低代表性样本以平衡数据分布
diff --git a/frontend/src/pages/DataManagement/Home/DataManagement.tsx b/frontend/src/pages/DataManagement/Home/DataManagement.tsx index 4fb93d65..61c037f5 100644 --- a/frontend/src/pages/DataManagement/Home/DataManagement.tsx +++ b/frontend/src/pages/DataManagement/Home/DataManagement.tsx @@ -1,399 +1,398 @@ -import { Card, Button, Statistic, Table, Tooltip, Tag, App } from "antd"; -import { - DownloadOutlined, - EditOutlined, - DeleteOutlined, - PlusOutlined, - UploadOutlined, -} from "@ant-design/icons"; -import TagManager from "@/components/business/TagManagement"; -import { Link, useNavigate } from "react-router"; -import { useEffect, useMemo, useState } from "react"; -import { SearchControls } from "@/components/SearchControls"; -import CardView from "@/components/CardView"; -import type { Dataset } from "@/pages/DataManagement/dataset.model"; -import { datasetStatusMap, datasetTypeMap, mapDataset } from "../dataset.const"; -import useFetchData from "@/hooks/useFetchData"; -import { - downloadDatasetUsingGet, - getDatasetStatisticsUsingGet, - queryDatasetsUsingGet, - deleteDatasetByIdUsingDelete, - createDatasetTagUsingPost, - queryDatasetTagsUsingGet, - deleteDatasetTagUsingDelete, - updateDatasetTagUsingPut, -} from "../dataset.api"; -import { formatBytes } from "@/utils/unit"; -import EditDataset from "../Create/EditDataset"; -import ImportConfiguration from "../Detail/components/ImportConfiguration"; - -export default function DatasetManagementPage() { - const navigate = useNavigate(); - const { message } = App.useApp(); - const [viewMode, setViewMode] = useState<"card" | "list">("card"); - const [editDatasetOpen, setEditDatasetOpen] = useState(false); - const [currentDataset, setCurrentDataset] = useState(null); - const [showUploadDialog, setShowUploadDialog] = useState(false); - const [statisticsData, setStatisticsData] = useState({ - count: {}, - size: {}, - }); - - async function fetchStatistics() { - const { data } = await getDatasetStatisticsUsingGet(); - - const statistics = { - size: [ - { - title: "数据集总数", - value: data?.totalDatasets || 0, - }, - { - title: "文件总数", - value: data?.totalFiles || 0, - }, - { - title: "总大小", - value: formatBytes(data?.totalSize) || '0 B', - }, - ], - count: [ - { - title: "文本", - value: data?.count?.text || 0, - }, - { - title: "图像", - value: data?.count?.image || 0, - }, - { - title: "音频", - value: data?.count?.audio || 0, - }, - { - title: "视频", - value: data?.count?.video || 0, - }, - ], - }; - setStatisticsData(statistics); - } - - const [tags, setTags] = useState([]); - - useEffect(() => { - const fetchTags = async () => { - const { data } = await queryDatasetTagsUsingGet(); - setTags(data.map((tag) => tag.name)); - }; - fetchTags(); - }, []); - - const filterOptions = useMemo( - () => [ - { - key: "type", - label: "类型", - options: [...Object.values(datasetTypeMap)], - }, - { - key: "status", - label: "状态", - options: [...Object.values(datasetStatusMap)], - }, - { - key: "tags", - label: "标签", - mode: "multiple", - options: tags.map((tag) => ({ label: tag, value: tag })), - }, - ], - [tags] - ); - - const { - loading, - tableData, - searchParams, - pagination, - fetchData, - setSearchParams, - handleFiltersChange, - handleKeywordChange, - } = useFetchData( - queryDatasetsUsingGet, - mapDataset, - 30000, // 30秒轮询间隔 - true, // 自动刷新 - [fetchStatistics], // 额外的轮询函数 - 0 - ); - - const handleDownloadDataset = async (dataset: Dataset) => { - await downloadDatasetUsingGet(dataset.id, dataset.name); - message.success("数据集下载成功"); - }; - - const handleDeleteDataset = async (id: number) => { - if (!id) return; - await deleteDatasetByIdUsingDelete(id); - fetchData({ pageOffset: 0 }); - message.success("数据删除成功"); - }; - - const handleImportData = (dataset: Dataset) => { - setCurrentDataset(dataset); - setShowUploadDialog(true); - }; - - const handleRefresh = async (showMessage = true) => { - await fetchData({ pageOffset: 0 }); - if (showMessage) { - message.success("数据已刷新"); - } - }; - - const operations = [ - { - key: "edit", - label: "编辑", - icon: , - onClick: (item: Dataset) => { - setCurrentDataset(item); - setEditDatasetOpen(true); - }, - }, - { - key: "import", - label: "导入", - icon: , - onClick: (item: Dataset) => { - handleImportData(item); - }, - }, - { - key: "download", - label: "下载", - icon: , - onClick: (item: Dataset) => { - if (!item.id) return; - handleDownloadDataset(item); - }, - }, - { - key: "delete", - label: "删除", - danger: true, - confirm: { - title: "确认删除该数据集？", - description: "删除后该数据集将无法恢复，请谨慎操作。", - okText: "删除", - cancelText: "取消", - okType: "danger", - }, - icon: , - onClick: (item: Dataset) => handleDeleteDataset(item.id), - }, - ]; - - const columns = [ - { - title: "名称", - dataIndex: "name", - key: "name", - fixed: "left", - render: (name, record) => ( - - ), - }, - { - title: "类型", - dataIndex: "type", - key: "type", - width: 100, - }, - { - title: "状态", - dataIndex: "status", - key: "status", - render: (status: any) => { - return ( - - {status?.label} - - ); - }, - width: 120, - }, - { - title: "大小", - dataIndex: "size", - key: "size", - width: 120, - }, - { - title: "文件数", - dataIndex: "fileCount", - key: "fileCount", - width: 100, - }, - // { - // title: "创建者", - // dataIndex: "createdBy", - // key: "createdBy", - // width: 120, - // }, - { - title: "存储路径", - dataIndex: "targetLocation", - key: "targetLocation", - width: 200, - ellipsis: true, - }, - { - title: "创建时间", - dataIndex: "createdAt", - key: "createdAt", - width: 180, - }, - { - title: "更新时间", - dataIndex: "updatedAt", - key: "updatedAt", - width: 180, - }, - { - title: "操作", - key: "actions", - width: 200, - fixed: "right", - render: (_: any, record: Dataset) => ( -
- {operations.map((op) => ( - -
- ), - }, - ]; - - const renderCardView = () => ( - { - navigate("/data/management/detail/" + dataset.id); - }} - /> - ); - - const renderListView = () => ( - - - - ); - - useEffect(() => { - const refresh = () => { - handleRefresh(true); - }; - window.addEventListener("update:datasets", refresh); - return () => { - window.removeEventListener("update:datasets", refresh); - }; - }, []); - - return ( -
- {/* Header */} -
-

数据管理

-
- {/* tasks */} - deleteDatasetTagUsingDelete({ ids })} - onUpdate={updateDatasetTagUsingPut} - onFetch={queryDatasetTagsUsingGet} - /> - - - -
-
- - {/* Statistics */} -
- -
- {statisticsData.size?.map?.((item) => ( - - ))} -
- -
- setSearchParams({ ...searchParams, filter: {} })} - viewMode={viewMode} - onViewModeChange={setViewMode} - showViewToggle - onReload={handleRefresh} - /> - {viewMode === "card" ? renderCardView() : renderListView()} - { - setCurrentDataset(null); - setEditDatasetOpen(false); - }} - onRefresh={handleRefresh} - /> - { - setCurrentDataset(null); - setShowUploadDialog(false); - }} - prefix="" - updateEvent="update:datasets" - /> -
- ); -} +import { Card, Button, Statistic, Table, Tooltip, Tag, App } from "antd"; +import { + DownloadOutlined, + EditOutlined, + DeleteOutlined, + PlusOutlined, + UploadOutlined, +} from "@ant-design/icons"; +import TagManager from "@/components/business/TagManagement"; +import { Link, useNavigate } from "react-router"; +import { useEffect, useMemo, useState } from "react"; +import { SearchControls } from "@/components/SearchControls"; +import CardView from "@/components/CardView"; +import type { Dataset } from "@/pages/DataManagement/dataset.model"; +import { datasetStatusMap, datasetTypeMap, mapDataset } from "../dataset.const"; +import useFetchData from "@/hooks/useFetchData"; +import { + downloadDatasetUsingGet, + getDatasetStatisticsUsingGet, + queryDatasetsUsingGet, + deleteDatasetByIdUsingDelete, + createDatasetTagUsingPost, + queryDatasetTagsUsingGet, + deleteDatasetTagUsingDelete, + updateDatasetTagUsingPut, +} from "../dataset.api"; +import { formatBytes } from "@/utils/unit"; +import EditDataset from "../Create/EditDataset"; +import ImportConfiguration from "../Detail/components/ImportConfiguration"; + +export default function DatasetManagementPage() { + const navigate = useNavigate(); + const { message } = App.useApp(); + const [viewMode, setViewMode] = useState<"card" | "list">("card"); + const [editDatasetOpen, setEditDatasetOpen] = useState(false); + const [currentDataset, setCurrentDataset] = useState(null); + const [showUploadDialog, setShowUploadDialog] = useState(false); + const [statisticsData, setStatisticsData] = useState({ + count: {}, + size: {}, + }); + + async function fetchStatistics() { + const { data } = await getDatasetStatisticsUsingGet(); + + const statistics = { + size: [ + { + title: "数据集总数", + value: data?.totalDatasets || 0, + }, + { + title: "文件总数", + value: data?.totalFiles || 0, + }, + { + title: "总大小", + value: formatBytes(data?.totalSize) || '0 B', + }, + ], + count: [ + { + title: "文本", + value: data?.count?.text || 0, + }, + { + title: "图像", + value: data?.count?.image || 0, + }, + { + title: "音频", + value: data?.count?.audio || 0, + }, + { + title: "视频", + value: data?.count?.video || 0, + }, + ], + }; + setStatisticsData(statistics); + } + + const [tags, setTags] = useState([]); + + useEffect(() => { + const fetchTags = async () => { + const { data } = await queryDatasetTagsUsingGet(); + setTags(data.map((tag) => tag.name)); + }; + fetchTags(); + }, []); + + const filterOptions = useMemo( + () => [ + { + key: "type", + label: "类型", + options: [...Object.values(datasetTypeMap)], + }, + { + key: "status", + label: "状态", + options: [...Object.values(datasetStatusMap)], + }, + { + key: "tags", + label: "标签", + mode: "multiple", + options: tags.map((tag) => ({ label: tag, value: tag })), + }, + ], + [tags] + ); + + const { + loading, + tableData, + searchParams, + pagination, + fetchData, + setSearchParams, + handleFiltersChange, + handleKeywordChange, + } = useFetchData( + queryDatasetsUsingGet, + mapDataset, + 30000, // 30秒轮询间隔 + true, // 自动刷新 + [fetchStatistics], // 额外的轮询函数 + 0 + ); + + const handleDownloadDataset = async (dataset: Dataset) => { + await downloadDatasetUsingGet(dataset.id, dataset.name); + message.success("数据集下载成功"); + }; + + const handleDeleteDataset = async (id: number) => { + if (!id) return; + await deleteDatasetByIdUsingDelete(id); + fetchData({ pageOffset: 0 }); + message.success("数据删除成功"); + }; + + const handleImportData = (dataset: Dataset) => { + setCurrentDataset(dataset); + setShowUploadDialog(true); + }; + + const handleRefresh = async (showMessage = true) => { + await fetchData({ pageOffset: 0 }); + if (showMessage) { + message.success("数据已刷新"); + } + }; + + const operations = [ + { + key: "edit", + label: "编辑", + icon: , + onClick: (item: Dataset) => { + setCurrentDataset(item); + setEditDatasetOpen(true); + }, + }, + { + key: "import", + label: "导入", + icon: , + onClick: (item: Dataset) => { + handleImportData(item); + }, + }, + { + key: "download", + label: "下载", + icon: , + onClick: (item: Dataset) => { + if (!item.id) return; + handleDownloadDataset(item); + }, + }, + { + key: "delete", + label: "删除", + danger: true, + confirm: { + title: "确认删除该数据集？", + description: "删除后该数据集将无法恢复，请谨慎操作。", + okText: "删除", + cancelText: "取消", + okType: "danger", + }, + icon: , + onClick: (item: Dataset) => handleDeleteDataset(item.id), + }, + ]; + + const columns = [ + { + title: "名称", + dataIndex: "name", + key: "name", + fixed: "left", + render: (name, record) => ( + + ), + }, + { + title: "类型", + dataIndex: "type", + key: "type", + width: 100, + }, + { + title: "状态", + dataIndex: "status", + key: "status", + render: (status: any) => { + return ( + + {status?.label} + + ); + }, + width: 120, + }, + { + title: "大小", + dataIndex: "size", + key: "size", + width: 120, + }, + { + title: "文件数", + dataIndex: "fileCount", + key: "fileCount", + width: 100, + }, + // { + // title: "创建者", + // dataIndex: "createdBy", + // key: "createdBy", + // width: 120, + // }, + { + title: "存储路径", + dataIndex: "targetLocation", + key: "targetLocation", + width: 200, + ellipsis: true, + }, + { + title: "创建时间", + dataIndex: "createdAt", + key: "createdAt", + width: 180, + }, + { + title: "更新时间", + dataIndex: "updatedAt", + key: "updatedAt", + width: 180, + }, + { + title: "操作", + key: "actions", + width: 200, + fixed: "right", + render: (_: any, record: Dataset) => ( +
+ {operations.map((op) => ( + +
+ ), + }, + ]; + + const renderCardView = () => ( + { + navigate("/data/management/detail/" + dataset.id); + }} + /> + ); + + const renderListView = () => ( + +
+ + ); + + useEffect(() => { + const refresh = () => { + handleRefresh(true); + }; + window.addEventListener("update:datasets", refresh); + return () => { + window.removeEventListener("update:datasets", refresh); + }; + }, []); + + return ( +
+ {/* Header */} +
+

数据管理

+
+ {/* tasks */} + deleteDatasetTagUsingDelete({ ids })} + onUpdate={updateDatasetTagUsingPut} + onFetch={queryDatasetTagsUsingGet} + /> + + + +
+
+ + {/* Statistics */} +
+ +
+ {statisticsData.size?.map?.((item) => ( + + ))} +
+ +
+ setSearchParams({ ...searchParams, filter: {} })} + viewMode={viewMode} + onViewModeChange={setViewMode} + showViewToggle + onReload={handleRefresh} + /> + {viewMode === "card" ? renderCardView() : renderListView()} + { + setCurrentDataset(null); + setEditDatasetOpen(false); + }} + onRefresh={handleRefresh} + /> + { + setCurrentDataset(null); + setShowUploadDialog(false); + }} + prefix="" + updateEvent="update:datasets" + /> +
+ ); +} diff --git a/frontend/src/pages/DataManagement/dataset.const.tsx b/frontend/src/pages/DataManagement/dataset.const.tsx index d5b70901..219ac353 100644 --- a/frontend/src/pages/DataManagement/dataset.const.tsx +++ b/frontend/src/pages/DataManagement/dataset.const.tsx @@ -211,7 +211,9 @@ export function mapDataset(dataset: AnyObject): Dataset { status: datasetStatusMap[dataset.status], statistics: [ { label: "文件数", value: dataset.fileCount || 0 }, + { label: "已标注", value: Math.floor(dataset.fileCount / 10) * 10}, { label: "大小", value: formatBytes(dataset.totalSize || 0) }, + { label: "关联归集任务", value: Math.floor(dataset.fileCount / 10)}, ], lastModified: dataset.updatedAt, }; From e9b1cc6aa320c531670c711c08b074a8dd476d56 Mon Sep 17 00:00:00 2001 From: uname <2986773479@qq.com> Date: Mon, 26 Jan 2026 17:45:18 +0800 Subject: [PATCH 3/3] fix: add data lineage front-end implementation. --- .../Detail/components/DataLineageFlow.tsx | 495 +++++++++++++----- .../Detail/components/DataQuality.tsx | 1 + .../src/pages/DataManagement/dataset.api.ts | 5 + 3 files changed, 365 insertions(+), 136 deletions(-) diff --git a/frontend/src/pages/DataManagement/Detail/components/DataLineageFlow.tsx b/frontend/src/pages/DataManagement/Detail/components/DataLineageFlow.tsx index 0d3e11a4..f3980d3f 100644 --- a/frontend/src/pages/DataManagement/Detail/components/DataLineageFlow.tsx +++ b/frontend/src/pages/DataManagement/Detail/components/DataLineageFlow.tsx @@ -1,8 +1,9 @@ import type React from "react" -import { useState, useRef, useEffect } from "react" +import { useState, useRef, useEffect, useMemo } from "react" import { Card, Badge } from "antd" import { Database, Table, Brain, BookOpen, X } from "lucide-react" -import type { Dataset } from "@/pages/DataManagement/dataset.model.ts" +import {Dataset} from "@/pages/DataManagement/dataset.model.ts"; +import { queryDatasetLineageByIdUsingGet } from "@/pages/DataManagement/dataset.api.ts"; interface Node { id: string @@ -18,118 +19,37 @@ interface Node { } interface Edge { + id: string from: string to: string label: string + edgeType?: string + processId?: string + description?: string +} +interface LineageNodeDTO { + id: string + name: string + nodeType?: string + type?: string + graphId?: string + description?: string + nodeMetadata?: string + metadata?: string } -const nodes: Node[] = [ - { - id: "source1", - type: "datasource", - label: "MySQL 数据库", - x: 80, - y: 100, - description: "业务数据库", - status: "运转", - updateTime: "2026-01-06 15:22:08", - }, - { - id: "source2", - type: "datasource", - label: "API 接口", - x: 80, - y: 250, - description: "外部数据源", - status: "运转", - updateTime: "2026-01-06 14:30:15", - }, - { - id: "source3", - type: "datasource", - label: "日志文件", - x: 80, - y: 400, - description: "系统日志", - status: "运转", - updateTime: "2026-01-06 16:05:42", - }, - { - id: "dataset1", - type: "dataset", - label: "原始数据集", - x: 380, - y: 100, - description: "未处理的原始数据", - fileCount: 14, - size: "3.449 MB", - updateTime: "2026-01-06 15:22:08", - }, - { - id: "dataset2", - type: "dataset", - label: "清洗数据集", - x: 680, - y: 175, - description: "清洗后的干净数据", - fileCount: 8, - size: "2.156 MB", - updateTime: "2026-01-06 15:45:20", - }, - { - id: "dataset3", - type: "dataset", - label: "合成数据集", - x: 980, - y: 250, - description: "特征工程后的数据", - fileCount: 5, - size: "1.823 MB", - updateTime: "2026-01-06 16:10:35", - }, - { - id: "model1", - type: "model", - label: "预测模型", - x: 1280, - y: 150, - description: "ML 预测模型", - status: "训练中", - updateTime: "2026-01-06 16:30:12", - }, - { - id: "model2", - type: "model", - label: "分类模型", - x: 1280, - y: 350, - description: "分类算法模型", - status: "已完成", - updateTime: "2026-01-06 16:25:48", - }, - { - id: "kb1", - type: "knowledge", - label: "业务知识库", - x: 1600, - y: 250, - description: "结构化知识存储", - fileCount: 32, - size: "8.742 MB", - updateTime: "2026-01-06 16:45:05", - }, -] - -const edges: Edge[] = [ - { from: "source1", to: "dataset1", label: "数据归集" }, - { from: "source2", to: "dataset1", label: "数据归集" }, - { from: "source3", to: "dataset1", label: "数据归集" }, - { from: "dataset1", to: "dataset2", label: "数据清洗" }, - { from: "dataset2", to: "dataset3", label: "数据合成" }, - { from: "dataset2", to: "kb1", label: "知识生成" }, - { from: "dataset3", to: "model1", label: "模型训练" }, - { from: "dataset3", to: "model2", label: "模型训练" }, -] +interface LineageEdgeDTO { + id: string + graphId?: string + processId?: string + edgeType?: string + name?: string + description?: string + edgeMetadata?: string + metadata?: string + fromNodeId: string + toNodeId: string +} const nodeConfig = { datasource: { @@ -158,14 +78,132 @@ const nodeConfig = { }, } -export default function DataLineageFlow(dataset: Dataset) { +const edgeTypeLabels: Record = { + DATA_COLLECTION: "数据归集", + DATA_CLEANING: "数据清洗", + DATA_LABELING: "数据标注", + DATA_SYNTHESIS: "数据合成", + DATA_RATIO: "数据配比", +} + +const nodeTypeToUi: Record = { + DATASOURCE: "datasource", + DATASET: "dataset", + MODEL: "model", + KNOWLEDGE_BASE: "knowledge", + KNOWLEDGE: "knowledge", +} + +const layoutColumns: Record = { + datasource: 0, + dataset: 1, + model: 2, + knowledge: 3, +} + +const layoutConfig = { + startX: 80, + startY: 90, + columnGap: 300, + rowGap: 150, + nodeWidth: 180, + nodeHeight: 74, + canvasWidth: 2000, + canvasHeight: 720, +} + +export default function DataLineageFlow({ dataset }: { dataset: Dataset }) { + const [graphNodes, setGraphNodes] = useState([]) + const [graphEdges, setGraphEdges] = useState([]) const [selectedNode, setSelectedNode] = useState(null) + const [selectedEdge, setSelectedEdge] = useState(null) const [hoveredNode, setHoveredNode] = useState(null) + const [hoveredEdge, setHoveredEdge] = useState(null) const [draggedNode, setDraggedNode] = useState(null) const [dragOffset, setDragOffset] = useState({ x: 0, y: 0 }) const [renderTrigger, setRenderTrigger] = useState(0) const canvasRef = useRef(null) const containerRef = useRef(null) + const edgeHitAreasRef = useRef>([]) + const datasetId = dataset?.id + + const layoutGraph = useMemo(() => { + return (nodes: LineageNodeDTO[], edges: LineageEdgeDTO[]): { nodes: Node[]; edges: Edge[] } => { + const columns: Record = { + datasource: [], + dataset: [], + model: [], + knowledge: [], + } + + const mappedNodes: Node[] = nodes.map((node) => { + const rawType = node.nodeType ?? node.type ?? "DATASET" + const uiType = nodeTypeToUi[rawType] ?? "dataset" + const columnIndex = layoutColumns[uiType] + const baseX = layoutConfig.startX + columnIndex * layoutConfig.columnGap + const currentY = layoutConfig.startY + columns[uiType].length * layoutConfig.rowGap + const mapped: Node = { + id: node.id, + type: uiType, + label: node.name, + description: node.description, + x: baseX, + y: currentY, + } + columns[uiType].push(mapped) + return mapped + }) + + const mappedEdges: Edge[] = edges.map((edge, index) => ({ + id: edge.id || `${edge.fromNodeId}-${edge.toNodeId}-${index}`, + from: edge.fromNodeId, + to: edge.toNodeId, + label: edge.name || (edge.edgeType ? edgeTypeLabels[edge.edgeType] || edge.edgeType : "处理流程"), + edgeType: edge.edgeType, + processId: edge.processId, + description: edge.description, + })) + + return { nodes: mappedNodes, edges: mappedEdges } + } + }, []) + + useEffect(() => { + if (!datasetId) { + setGraphNodes([]) + setGraphEdges([]) + return + } + + const fetchLineage = async () => { + try { + const res = await queryDatasetLineageByIdUsingGet(datasetId) + const payload = res?.data?.data ?? res?.data + const lineageNodes: LineageNodeDTO[] = payload?.lineageNodes ?? [] + const lineageEdges: LineageEdgeDTO[] = payload?.lineageEdges ?? [] + const { nodes, edges } = layoutGraph(lineageNodes, lineageEdges) + setGraphNodes(nodes) + setGraphEdges(edges) + } catch (error) { + setGraphNodes([]) + setGraphEdges([]) + } + } + + fetchLineage() + }, [datasetId, layoutGraph]) + + useEffect(() => { + if (selectedNode && !graphNodes.some((node) => node.id === selectedNode.id)) { + setSelectedNode(null) + } + }, [graphNodes, selectedNode]) + + useEffect(() => { + if (selectedEdge && !graphEdges.some((edge) => edge.id === selectedEdge.id)) { + setSelectedEdge(null) + } + }, [graphEdges, selectedEdge]) useEffect(() => { const canvas = canvasRef.current @@ -180,22 +218,25 @@ export default function DataLineageFlow(dataset: Dataset) { ctx.scale(window.devicePixelRatio, window.devicePixelRatio) ctx.clearRect(0, 0, rect.width, rect.height) + edgeHitAreasRef.current = [] - edges.forEach((edge) => { - const fromNode = nodes.find((n) => n.id === edge.from) - const toNode = nodes.find((n) => n.id === edge.to) + graphEdges.forEach((edge) => { + const fromNode = graphNodes.find((n) => n.id === edge.from) + const toNode = graphNodes.find((n) => n.id === edge.to) if (!fromNode || !toNode) return + const isEdgeActive = hoveredEdge === edge.id || selectedEdge?.id === edge.id const isHighlighted = + isEdgeActive || hoveredNode === edge.from || hoveredNode === edge.to || selectedNode?.id === edge.from || selectedNode?.id === edge.to - const startX = fromNode.x + 140 - const startY = fromNode.y + 35 + const startX = fromNode.x + layoutConfig.nodeWidth + const startY = fromNode.y + layoutConfig.nodeHeight / 2 const endX = toNode.x - const endY = toNode.y + 35 + const endY = toNode.y + layoutConfig.nodeHeight / 2 const controlPointOffset = Math.abs(endX - startX) * 0.4 const cp1x = startX + controlPointOffset @@ -270,8 +311,16 @@ export default function DataLineageFlow(dataset: Dataset) { ctx.textAlign = "center" ctx.textBaseline = "middle" ctx.fillText(edge.label, midX, midY) + + edgeHitAreasRef.current.push({ + id: edge.id, + x: midX - textWidth / 2 - padding, + y: midY - 8, + width: textWidth + padding * 2, + height: 16, + }) }) - }, [hoveredNode, renderTrigger, selectedNode]) + }, [graphEdges, graphNodes, hoveredEdge, hoveredNode, renderTrigger, selectedEdge, selectedNode]) useEffect(() => { const handleMouseMove = (e: MouseEvent) => { @@ -282,12 +331,18 @@ export default function DataLineageFlow(dataset: Dataset) { const x = e.clientX - rect.left - dragOffset.x const y = e.clientY - rect.top - dragOffset.y - const nodeIndex = nodes.findIndex((n) => n.id === draggedNode) - if (nodeIndex !== -1) { - nodes[nodeIndex].x = Math.max(0, Math.min(x, rect.width - 120)) - nodes[nodeIndex].y = Math.max(0, Math.min(y, rect.height - 70)) - setRenderTrigger((prev) => prev + 1) - } + setGraphNodes((prevNodes) => { + const nodeIndex = prevNodes.findIndex((node) => node.id === draggedNode) + if (nodeIndex === -1) return prevNodes + const updated = [...prevNodes] + updated[nodeIndex] = { + ...updated[nodeIndex], + x: Math.max(12, Math.min(x, rect.width - layoutConfig.nodeWidth - 12)), + y: Math.max(12, Math.min(y, rect.height - layoutConfig.nodeHeight - 12)), + } + return updated + }) + setRenderTrigger((prev) => prev + 1) } const handleMouseUp = () => { @@ -307,6 +362,36 @@ export default function DataLineageFlow(dataset: Dataset) { const handleNodeClick = (node: Node) => { setSelectedNode(node) + setSelectedEdge(null) + } + + const findEdgeAt = (x: number, y: number): Edge | null => { + const hit = edgeHitAreasRef.current.find( + (area) => x >= area.x && x <= area.x + area.width && y >= area.y && y <= area.y + area.height + ) + if (!hit) return null + return graphEdges.find((edge) => edge.id === hit.id) || null + } + + const handleCanvasClick = (e: React.MouseEvent) => { + if (!containerRef.current) return + const rect = containerRef.current.getBoundingClientRect() + const x = e.clientX - rect.left + const y = e.clientY - rect.top + const hitEdge = findEdgeAt(x, y) + if (hitEdge) { + setSelectedEdge(hitEdge) + setSelectedNode(null) + } + } + + const handleCanvasMouseMove = (e: React.MouseEvent) => { + if (!containerRef.current) return + const rect = containerRef.current.getBoundingClientRect() + const x = e.clientX - rect.left + const y = e.clientY - rect.top + const hitEdge = findEdgeAt(x, y) + setHoveredEdge(hitEdge?.id ?? null) } const handleNodeMouseDown = (e: React.MouseEvent, node: Node) => { @@ -324,24 +409,89 @@ export default function DataLineageFlow(dataset: Dataset) { const getRelatedNodes = (nodeId: string): string[] => { const related = new Set() - edges.forEach((edge) => { + graphEdges.forEach((edge) => { if (edge.from === nodeId) related.add(edge.to) if (edge.to === nodeId) related.add(edge.from) }) return Array.from(related) } + const stats = useMemo( + () => ({ + nodes: graphNodes.length, + edges: graphEdges.length, + }), + [graphNodes.length, graphEdges.length] + ) + return (
+
+
+
数据血缘图
+
+ {stats.nodes} 个节点 · {stats.edges} 条边 +
+
+
+ + + 数据源 + + + + 数据集 + + + + 模型 + + + + 知识库 + +
+
{ + setSelectedNode(null) + setSelectedEdge(null) + }} > - +
+ { + e.stopPropagation() + handleCanvasClick(e) + }} + onMouseMove={handleCanvasMouseMove} + onMouseLeave={() => setHoveredEdge(null)} + /> + + {graphNodes.length === 0 && ( +
+
暂无血缘数据
+
完成归集或关联流程后将自动生成血缘图
+
+ )} - {nodes.map((node) => { + {graphNodes.map((node) => { const config = nodeConfig[node.type] const Icon = config.icon const isSelected = selectedNode?.id === node.id @@ -361,14 +511,18 @@ export default function DataLineageFlow(dataset: Dataset) { transform: isHovered || isSelected ? "scale(1.05)" : "scale(1)", filter: isHovered || isSelected ? "drop-shadow(0 4px 12px rgba(0,0,0,0.15))" : "none", }} - onClick={() => handleNodeClick(node)} + onClick={(e) => { + e.stopPropagation() + handleNodeClick(node) + }} onMouseDown={(e) => handleNodeMouseDown(e, node)} onMouseEnter={() => setHoveredNode(node.id)} onMouseLeave={() => setHoveredNode(null)} >

上游依赖

- {edges + {graphEdges .filter((e) => e.to === selectedNode.id) .map((e) => { - const fromNode = nodes.find((n) => n.id === e.from) + const fromNode = graphNodes.find((n) => n.id === e.from) return fromNode ? (
) : null })} - {edges.filter((e) => e.to === selectedNode.id).length === 0 && ( + {graphEdges.filter((e) => e.to === selectedNode.id).length === 0 && (

无上游依赖

)}
@@ -548,10 +702,10 @@ export default function DataLineageFlow(dataset: Dataset) {

下游影响

- {edges + {graphEdges .filter((e) => e.from === selectedNode.id) .map((e) => { - const toNode = nodes.find((n) => n.id === e.to) + const toNode = graphNodes.find((n) => n.id === e.to) return toNode ? (
) : null })} - {edges.filter((e) => e.from === selectedNode.id).length === 0 && ( + {graphEdges.filter((e) => e.from === selectedNode.id).length === 0 && (

无下游影响

)}
@@ -574,6 +728,75 @@ export default function DataLineageFlow(dataset: Dataset) {
)} + + {!selectedNode && selectedEdge && ( + +
+
+
+

流程详情

+ + {selectedEdge.edgeType + ? edgeTypeLabels[selectedEdge.edgeType] || selectedEdge.edgeType + : "处理流程"} + +
+ +
+
+
+

基本信息

+
+
+ ID: + {selectedEdge.id} +
+
+ 名称: + {selectedEdge.label} +
+ {selectedEdge.processId && ( +
+ 流程ID: + {selectedEdge.processId} +
+ )} +
+
+
+

上下游关系

+
+
+ 上游: + {graphNodes.find((n) => n.id === selectedEdge.from)?.label || selectedEdge.from} +
+
+ 下游: + {graphNodes.find((n) => n.id === selectedEdge.to)?.label || selectedEdge.to} +
+
+
+ {selectedEdge.description && ( +
+

描述

+

{selectedEdge.description}

+
+ )} +
+
+ + )}
) } diff --git a/frontend/src/pages/DataManagement/Detail/components/DataQuality.tsx b/frontend/src/pages/DataManagement/Detail/components/DataQuality.tsx index 17208d4a..db8d1b66 100644 --- a/frontend/src/pages/DataManagement/Detail/components/DataQuality.tsx +++ b/frontend/src/pages/DataManagement/Detail/components/DataQuality.tsx @@ -74,6 +74,7 @@ function getMockMetrics(datasetType: DatasetType, stats: FileStats) { } export default function DataQuality(props: Props = {}) { + return const { dataset, datasetType: propDatasetType, fileStats: propFileStats } = props; // Prefer dataset fields when available, then explicit props, then sensible defaults diff --git a/frontend/src/pages/DataManagement/dataset.api.ts b/frontend/src/pages/DataManagement/dataset.api.ts index a7f36fd0..c00ba0e7 100644 --- a/frontend/src/pages/DataManagement/dataset.api.ts +++ b/frontend/src/pages/DataManagement/dataset.api.ts @@ -24,6 +24,11 @@ export function queryDatasetByIdUsingGet(id: string | number) { return get(`/api/data-management/datasets/${id}`); } +// 根据ID获取数据集详情 +export function queryDatasetLineageByIdUsingGet(id: string | number) { + return get(`/api/data-management/datasets/${id}/lineage`); +} + // 更新数据集 export function updateDatasetByIdUsingPut(id: string | number, data: any) { return put(`/api/data-management/datasets/${id}`, data);