修改data_flow表结构

修改元数据字段。
修改数据模型编辑功能。
修改数据资源编辑功能。
修改数据流编辑功能。

app/api/data_flow/routes.py

@@ -148,15 +148,8 @@ def create_script():
             res = failed("请求数据不能为空", code=400)
             return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
         
-        # 提取文本描述
-        request_data = json_data.get('request_data')
-        
-        if not request_data or not isinstance(request_data, str):
-            res = failed("请求数据必须是文本描述", code=400)
-            return json.dumps(res, ensure_ascii=False, cls=MyEncoder)
-        
-        # 调用DataFlowService的create_script方法
-        script_content = DataFlowService.create_script(request_data)
+        # 直接使用前端提交的json_data作为request_data参数
+        script_content = DataFlowService.create_script(json_data)
         
         result_data = {
             'script_content': script_content,

app/core/data_model/model.py

@@ -235,7 +235,7 @@ def resource_handle_meta_data_model(id_lists, data_model_node_id):
             query = """
             MATCH (source:DataModel), (target:DataMeta)
             WHERE id(source)=$source_id AND id(target) IN $target_ids
-            MERGE (source)-[:INCLUDE]->(target)
+            MERGE (source)-[:INCLUDES]->(target)
             RETURN count(*) as count
             """
             with connect_graph().session() as session:
@@ -440,7 +440,7 @@ def handle_id_model(id):
     node_id = id
     cql = """
         MATCH (n:DataModel) WHERE id(n) = $nodeId
-        OPTIONAL MATCH (n)-[:INCLUDE]->(meta:DataMeta)  
+        OPTIONAL MATCH (n)-[:INCLUDES]->(meta:DataMeta)  
         OPTIONAL MATCH (n)-[:DERIVES_FROM]->(resource:DataResource)
         OPTIONAL MATCH (n)-[:label]->(tag:DataLabel)
         OPTIONAL MATCH (uses:model_use)-[:use]->(n)
@@ -1063,17 +1063,40 @@ def data_model_edit(receiver):
     category = receiver.get('category')
     describe = receiver.get('describe')
     tag = receiver.get('tag')
+    frequency = receiver.get('frequency')
+    leader = receiver.get('leader')
+    organization = receiver.get('organization')
+    status = receiver.get('status')
+    meta_data = receiver.get('metaData', [])
     
-    # 更新数据模型节点
+    # 更新数据模型节点 - 添加新的字段
     query = """
     MATCH (n:DataModel) WHERE id(n) = $id
-    SET n.name = $name, n.en_name = $en_name, n.category = $category, n.describe = $describe
+    SET n.name = $name, 
+        n.en_name = $en_name, 
+        n.category = $category, 
+        n.describe = $describe,
+        n.frequency = $frequency,
+        n.leader = $leader,
+        n.organization = $organization,
+        n.status = $status,
+        n.updateTime = $update_time
     RETURN n
     """
     
+    update_time = get_formatted_time()
     with connect_graph().session() as session:
-        result = session.run(query, id=id, name=name, en_name=en_name, 
-                             category=category, describe=describe).data()
+        result = session.run(query, 
+                             id=id, 
+                             name=name, 
+                             en_name=en_name, 
+                             category=category, 
+                             describe=describe,
+                             frequency=frequency,
+                             leader=leader,
+                             organization=organization,
+                             status=status,
+                             update_time=update_time).data()
     
     # 处理标签关系
     if tag:
@@ -1114,4 +1137,48 @@ def data_model_edit(receiver):
                             )
                         )
     
+    # 处理DataMeta节点关系更新
+    with connect_graph().session() as session:
+        # 先删除DataModel关联的所有DataMeta关系
+        delete_meta_query = """
+        MATCH (n:DataModel)-[r:INCLUDES]->(m:DataMeta)
+        WHERE id(n) = $id
+        DELETE r
+        """
+        session.run(delete_meta_query, id=id)
+        logger.info(f"已删除DataModel({id})的所有DataMeta关系")
+        
+        # 根据上传的metaData数据是否有值来决定是否重新构建INCLUDES关系
+        if meta_data:
+            # 根据上传的metaData数据重新构建INCLUDES关系
+            for meta_item in meta_data:
+                meta_id = meta_item.get('id')
+                if meta_id:
+                    try:
+                        meta_id = int(meta_id)
+                        # 验证DataMeta节点是否存在
+                        check_meta_query = """
+                        MATCH (m:DataMeta)
+                        WHERE id(m) = $meta_id
+                        RETURN m
+                        """
+                        meta_result = session.run(check_meta_query, meta_id=meta_id)
+                        
+                        if meta_result.single():
+                            # 创建INCLUDES关系
+                            create_includes_query = """
+                            MATCH (n:DataModel), (m:DataMeta)
+                            WHERE id(n) = $model_id AND id(m) = $meta_id
+                            CREATE (n)-[:INCLUDES]->(m)
+                            RETURN n, m
+                            """
+                            session.run(create_includes_query, model_id=id, meta_id=meta_id)
+                            logger.info(f"成功创建INCLUDES关系: DataModel({id}) -> DataMeta({meta_id})")
+                        else:
+                            logger.warning(f"DataMeta节点不存在，ID: {meta_id}")
+                    except (ValueError, TypeError) as e:
+                        logger.error(f"无效的meta_id: {meta_id}, 错误: {str(e)}")
+        else:
+            logger.info(f"meta_data为空，不需要重新创建INCLUDES关系，DataModel({id})将不关联任何DataMeta节点")
+    
     return {"message": "数据模型更新成功"} 

app/core/data_resource/resource.py

@@ -1254,16 +1254,25 @@ def data_resource_edit(data):
             # 添加更新时间
             update_fields["updateTime"] = get_formatted_time()
             
-            # 构建更新语句
-            set_clause = ", ".join([f"n.{k} = ${k}" for k in update_fields.keys()])
-            cypher = f"""
-            MATCH (n:DataResource)
-            WHERE id(n) = $resource_id
-            SET {set_clause}
-            RETURN n
-            """
+            # 构建更新语句，确保至少有 updateTime 字段要更新
+            if update_fields:
+                set_clause = ", ".join([f"n.{k} = ${k}" for k in update_fields.keys()])
+                cypher = f"""
+                MATCH (n:DataResource)
+                WHERE id(n) = $resource_id
+                SET {set_clause}
+                RETURN n
+                """
+                result = session.run(cypher, resource_id=int(resource_id), **update_fields)
+            else:
+                # 如果没有字段需要更新，只查询节点
+                cypher = """
+                MATCH (n:DataResource)
+                WHERE id(n) = $resource_id
+                RETURN n
+                """
+                result = session.run(cypher, resource_id=int(resource_id))
             
-            result = session.run(cypher, resource_id=int(resource_id), **update_fields)
             updated_node = result.single()
             
             if not updated_node:
@@ -1294,22 +1303,75 @@ def data_resource_edit(data):
             
             # 处理元数据关系
             parsed_data = data.get("parsed_data", [])
+            
+            # 首先删除旧的元数据关系和清洗资源关系（无论parsed_data是否为空都要执行）
+            delete_meta_cypher = """
+            MATCH (n:DataResource)-[r:contain]->()
+            WHERE id(n) = $resource_id
+            DELETE r
+            """
+            session.run(delete_meta_cypher, resource_id=int(resource_id))
+            
+            delete_clean_cypher = """
+            MATCH (n:DataResource)-[r:clean_resource]->()
+            WHERE id(n) = $resource_id
+            DELETE r
+            """
+            session.run(delete_clean_cypher, resource_id=int(resource_id))
+            
+            # 根据parsed_data是否为空来决定是否执行预处理和关系新建操作
             if parsed_data:
-                # 删除旧的元数据关系
-                delete_meta_cypher = """
-                MATCH (n:DataResource)-[r:contain]->()
-                WHERE id(n) = $resource_id
-                DELETE r
-                """
-                session.run(delete_meta_cypher, resource_id=int(resource_id))
-                
-                # 删除旧的清洗资源关系
-                delete_clean_cypher = """
-                MATCH (n:DataResource)-[r:clean_resource]->()
-                WHERE id(n) = $resource_id
-                DELETE r
-                """
-                session.run(delete_clean_cypher, resource_id=int(resource_id))
+                # 预处理 parsed_data，确保每个 metadata 都有有效的 ID
+                for meta in parsed_data:
+                    meta_id = meta.get("id")
+                    meta_name = meta.get("name")
+                    
+                    if not meta_id and meta_name:
+                        # 如果没有 ID 但有 name，先根据 name 查找是否存在对应的 DataMeta 节点
+                        find_meta_cypher = """
+                        MATCH (m:DataMeta {name: $meta_name})
+                        RETURN m
+                        """
+                        find_result = session.run(find_meta_cypher, meta_name=meta_name)
+                        existing_meta = find_result.single()
+                        
+                        if existing_meta:
+                            # 如果找到了，使用现有的 ID
+                            meta_id = existing_meta["m"].id
+                            meta["id"] = meta_id
+                            logger.info(f"找到现有的DataMeta节点: {meta_name}, ID: {meta_id}")
+                        else:
+                            # 如果没有找到，创建新的 DataMeta 节点
+                            create_meta_cypher = """
+                            CREATE (m:DataMeta {
+                                name: $name,
+                                en_name: $en_name,
+                                data_type: $data_type,
+                                createTime: $create_time,
+                                updateTime: $update_time
+                            })
+                            RETURN m
+                            """
+                            create_time = get_formatted_time()
+                            new_meta_result = session.run(
+                                create_meta_cypher,
+                                name=meta_name,
+                                en_name=meta.get("en_name", meta_name),
+                                data_type=meta.get("data_type", "varchar(255)"),
+                                create_time=create_time,
+                                update_time=create_time
+                            )
+                            new_meta = new_meta_result.single()
+                            if new_meta:
+                                meta_id = new_meta["m"].id
+                                meta["id"] = meta_id
+                                logger.info(f"创建新的DataMeta节点: {meta_name}, ID: {meta_id}")
+                            else:
+                                logger.error(f"创建DataMeta节点失败: {meta_name}")
+                                continue
+                    elif not meta_id:
+                        logger.warning(f"跳过没有ID和name的metadata: {meta}")
+                        continue
                 
                 # 创建新的元数据关系和相关关系
                 for meta in parsed_data:
@@ -1357,9 +1419,11 @@ def data_resource_edit(data):
                             RETURN r
                             """
                             session.run(create_resource_standard_cypher, resource_id=int(resource_id), standard_id=int(standard_id))
+            else:
+                logger.info(f"parsed_data为空，只删除旧的元数据关系，不创建新的关系")
             
             # 返回更新后的节点
-                            node_data = serialize_node_properties(updated_node["n"])
+            node_data = serialize_node_properties(updated_node["n"])
             node_data["id"] = updated_node["n"].id
             
             # 记录最终返回的describe字段