完成datasource api开发和第一版的测试

app/api/data_resource/routes.py

@@ -165,15 +165,17 @@ def data_resource_save():
         if not receiver:
             return jsonify(failed("参数不完整：缺少receiver"))
         # 检查url是否存在
-        if 'url' not in receiver:
-            return jsonify(failed("参数不完整：缺少url"))
+        if 'url' not in receiver or not receiver['url']:
+            logger.debug(f"url 为空")
 
         additional_info = receiver.get('additional_info')
         if not additional_info:
             return jsonify(failed("参数不完整: 缺少additional_info"))
-              
-        file_extension = receiver['url'].split('.')[-1]
-        head_data = additional_info.get('head_data')           
+                      
+        head_data = additional_info.get('head_data') 
+
+        file_extension = receiver['url'].split('.')[-1] if receiver.get('url') else ""
+        
         
         if file_extension in ['xlsx', 'xls', 'csv']:
             # Excel/CSV文件必须有storage_location
@@ -184,7 +186,7 @@ def data_resource_save():
             # 调用业务逻辑处理数据资源创建，设置resource_type为structure
             resource_id = handle_node(receiver, head_data, data_source=None, resource_type='structure')
             
-        elif file_extension == 'sql':
+        elif file_extension == 'sql' or file_extension == "" or not file_extension:
             data_source = additional_info.get('data_source', '')
             storage_location = receiver.get('storage_location', '')
 

app/api/data_source/routes.py

@@ -243,23 +243,11 @@ def data_source_connstr_valid():
             existing_source = execute_cypher_query(check_query, {'en_name': data['en_name']})
             
             if existing_source:
-                return jsonify(success({
-                    "message": "连接信息验证通过，但该数据源的定义已经存在，如果保存则会更新该数据源",
-                    "valid": True,
-                    "exists": True
-                }))
+                return jsonify(success("连接信息验证通过，但该数据源的定义已经存在，如果保存则会更新该数据源"))
             else:
-                return jsonify(success({
-                    "message": "连接信息验证通过",
-                    "valid": True,
-                    "exists": False
-                }))
+                return jsonify(success("连接信息验证通过"))
         else:
-            return jsonify(failed({
-                "message": "连接信息验证失败",
-                "valid": False,
-                "exists": False
-            }))
+            return jsonify(failed("连接信息验证失败"))
 
     except Exception as e:
         logger.error(f"验证连接信息失败: {str(e)}")

app/core/data_resource/resource.py

@@ -105,7 +105,7 @@ def handle_node(receiver, head_data, data_source=None, resource_type=None):
             
         # 更新属性
         update_attributes = {
-            'en_name': receiver['en_name'],
+            'en_name': receiver.get('en_name', receiver.get('name', '')),
             'time': get_formatted_time(),
             'type': type_value  # 根据资源类型设置不同的type值
         }
@@ -232,7 +232,7 @@ def handle_node(receiver, head_data, data_source=None, resource_type=None):
                     if data_source_en_name:
                         # 创建 originates_from 关系
                         rel_data_source_cypher = """
-                        MATCH (a:data_resource), (b:data_source)
+                        MATCH (a:data_resource), (b:DataSource)
                         WHERE id(a) = $resource_id AND b.en_name = $ds_en_name
                         MERGE (a)-[r:originates_from]->(b)
                         RETURN r
@@ -252,7 +252,7 @@ def handle_node(receiver, head_data, data_source=None, resource_type=None):
                             logger.error(error_msg)
                             
                             # 检查数据源节点是否存在
-                            check_ds_cypher = "MATCH (b:data_source) WHERE b.en_name = $ds_en_name RETURN b"
+                            check_ds_cypher = "MATCH (b:DataSource) WHERE b.en_name = $ds_en_name RETURN b"
                             check_ds_result = session.run(check_ds_cypher, ds_en_name=data_source_en_name)
                             if not check_ds_result.single():
                                 logger.error(f"数据源节点不存在: en_name={data_source_en_name}")
@@ -1194,44 +1194,40 @@ def data_resource_edit(data):
         raise
 
 def handle_data_source(data_source):
-    """处理数据源的检查和创建
-    """
+    """处理数据源信息，创建或获取数据源节点"""
     try:
-        # 1. 检查en_name是否存在
-        ds_en_name = data_source.get("en_name")
-        if not ds_en_name:
-            raise ValueError("数据源信息不完整，缺少名称(en_name)")
-        
-        # 2. 处理name字段
-        if "name" not in data_source or not data_source["name"]:
-            data_source["name"] = ds_en_name
-            logger.debug(f"数据源name为空，使用en_name作为替代: {ds_en_name}")
-        
-        # 3. 检查是否为简单查询模式
-        required_fields = ["type", "host", "port", "database", "username"]
-        has_required_fields = all(data_source.get(field) for field in required_fields)
-        
         with neo4j_driver.get_session() as session:
-            # 简单查询模式：只通过en_name查找已有数据源
-            if not has_required_fields:
-                logger.info(f"简单数据源查询模式，查找en_name为: {ds_en_name}")
-                check_name_cypher = """
-                MATCH (ds:data_source {en_name: $en_name})
-                RETURN ds
-                """
-                check_result = session.run(check_name_cypher, en_name=ds_en_name)
-                existing_record = check_result.single()
+            # 获取英文名称作为唯一标识
+            ds_en_name = data_source.get("en_name")
+            if not ds_en_name:
+                logger.error("数据源缺少必要的en_name属性")
+                return None
                 
-                if existing_record:
-                    # 数据源已存在，返回其名称
-                    existing_data_source = dict(existing_record["ds"])
-                    logger.info(f"根据名称找到现有数据源: {existing_data_source.get('en_name')}")
-                    return existing_data_source.get("en_name")
-                else:
-                    # 数据源不存在，抛出异常
-                    raise ValueError(f"未找到名称为 {ds_en_name} 的数据源，请先创建该数据源或提供完整的数据源信息")
+            # 如果没有设置name，使用en_name作为name
+            if "name" not in data_source or not data_source["name"]:
+                data_source["name"] = ds_en_name
+            
+            # 检查必填字段
+            required_fields = ["type", "host", "port", "database", "username"]
+            has_required_fields = all(data_source.get(field) for field in required_fields)
+            
+            # 查询是否已存在相同en_name的数据源
+            existing_cypher = """
+            MATCH (ds:DataSource {en_name: $en_name})
+            RETURN ds
+            """
+            
+            existing_result = session.run(existing_cypher, en_name=ds_en_name)
+            existing_record = existing_result.single()
+            
+            if existing_record:
+                existing_data_source = dict(existing_record["ds"])
+                logger.info(f"根据名称找到现有数据源: {existing_data_source.get('en_name')}")
+                return existing_data_source.get("en_name")
+            else:
+                # 数据源不存在，抛出异常
+                raise ValueError(f"未找到名称为 {ds_en_name} 的数据源，请先创建该数据源或提供完整的数据源信息")
             
-              
     except Exception as e:
         logger.error(f"处理数据源失败: {str(e)}")
         raise RuntimeError(f"处理数据源失败: {str(e)}")

app/core/llm/ddl_parser.py

@@ -172,22 +172,24 @@ class DDLParser:
 1. 从DDL语句中识别所有表名，并在data对象中为每个表创建条目,表名请使用小写，可能会有多个表。
 2. 对于每个表，提取所有字段信息，包括名称、数据类型和注释。
    - 中文表名中不要出现标点符号
-3. 字段中文名称(name)的确定规则：
-   - 如有COMMENT注释，直接使用注释内容
-   - 如无注释但字段名有明确含义，将英文名翻译为中文
-   - 如字段名是无意义的拼音缩写，则name为空字符串
-   - 字段名中不要出现逗号，以及"主键"、"外键"、"索引"等字样
+   - 表中的字段对应输出json中的meta对象，en_name对应表的字段名，data_type对应表的字段类型.
+3. 返回结果的中文名称(name)的确定规则：
+   - 对于COMMENT注释，直接使用注释内容作为name
+   - 如sql中无注释但字段名en_name有明确含义，将英文名en_name翻译为中文
+   - 如字段名en_name是无意义的拼音缩写，则name为空字符串
+   - 中文字段名name中不要出现逗号，以及"主键"、"外键"、"索引"等字样
 4. 所有的表的定义信息，请放在tables对象中, tables对象的key为表名，value为表的定义信息。这里可能会有多个表，请一一识别。
-5. 忽略sql文件中除了表的定义和注释信息以外的内容。比如，忽略sql中的数据库的连接字符串。
+   - 对于每个表的字段都要检查它的en_name和name,name不能为空，首选字段的注释，如果没有注释，则尝试翻译en_name作为name。
+5. 忽略sql文件中除了表的定义和注释信息COMMIT以外的内容。比如，忽略sql中的数据库的连接字符串。
 6. 参考格式如下：
 {
     "users_table": { //表名
         "name": "用户表", //表的中文名，来自于COMMENT注释或LLM翻译，如果无法确定，则name为空字符串
         "schema": "public",
         "meta": [{
-                "en_name": "id",
-                "data_type": "integer",
-                "name": "用户ID"
+                "en_name": "id", //表的字段名
+                "data_type": "integer", //表的字段类型
+                "name": "用户ID" //表的中文名，来自于COMMENT注释或LLM翻译，如果无法确定，则name为空字符串
             },
             {
                 "en_name": "username",

app/core/production_line/production_line.py

@@ -221,7 +221,7 @@ def get_resource_storage_info(resource_id):
             metadata_query = """
             MATCH (n:data_resource)-[:contain]->(m:meta_data)
             WHERE id(n) = $resource_id
-            RETURN m.name as name, m.en_name as en_name, m.data_type as type
+            RETURN m.name as name, m.en_name as en_name, m.data_type as data_type
             """
             result = session.run(metadata_query, resource_id=int(resource_id))
             metadata_list = [dict(record) for record in result]
@@ -909,51 +909,35 @@ def get_resource_details(resource_id):
     return handle_id_resource(resource_id)
 
 def get_resource_data_source(resource_id):
-    """
-    获取数据资源关联的数据源信息
-    
-    Args:
-        resource_id: 数据资源ID
-        
-    Returns:
-        dict: 数据源连接信息
-    """
+    """获取数据资源关联的数据源信息"""
     try:
         with neo4j_driver.get_session() as session:
-            # 查询数据资源关联的数据源节点
+            # 查询数据资源节点连接的数据源节点
             cypher = """
-            MATCH (n:data_resource)-[:originates_from]->(ds:data_source)
+            MATCH (n:data_resource)-[:originates_from]->(ds:DataSource)
             WHERE id(n) = $resource_id
             RETURN ds
             """
+            
             result = session.run(cypher, resource_id=int(resource_id))
             record = result.single()
             
             if not record:
-                logger.warning(f"资源 {resource_id} 没有关联的数据源节点")
+                logger.warning(f"资源ID {resource_id} 没有关联的数据源")
                 return None
-                
-            data_source = dict(record["ds"])
             
-            # 构建连接信息
-            connection_info = {
+            # 构建数据源连接信息
+            data_source = dict(record["ds"])
+            return {
                 "type": data_source.get("type", "").lower(),
                 "host": data_source.get("host"),
                 "port": data_source.get("port"),
                 "database": data_source.get("database"),
                 "username": data_source.get("username"),
                 "password": data_source.get("password")
-                # 添加param参数,但是由于pyMySQL不支持一些参数，所以暂时不使用参数。
-               # "param": data_source.get("param")  
+                # 如果需要其他参数可以添加
+                # "param": data_source.get("param")
             }
-            
-            # 验证必要字段
-            if not all([connection_info["type"], connection_info["host"], 
-                        connection_info["database"], connection_info["username"]]):
-                logger.error(f"数据源信息不完整: {connection_info}")
-                return None
-                
-            return connection_info
     except Exception as e:
         logger.error(f"获取数据源信息失败: {str(e)}")
         return None