import logging from typing import Dict, List, Optional, Any, Union from datetime import datetime import json from app.core.llm.llm_service import llm_sql from app.core.graph.graph_operations import ( connect_graph, create_or_get_node, get_node, relationship_exists, ) from app.core.meta_data import translate_and_parse, get_formatted_time from app import db from sqlalchemy import text logger = logging.getLogger(__name__) class DataFlowService: """数据流服务类，处理数据流相关的业务逻辑""" @staticmethod def get_dataflows( page: int = 1, page_size: int = 10, search: str = '', ) -> Dict[str, Any]: """ 获取数据流列表 Args: page: 页码 page_size: 每页大小 search: 搜索关键词 Returns: 包含数据流列表和分页信息的字典 """ try: # 从图数据库查询数据流列表 skip_count = (page - 1) * page_size # 构建搜索条件 where_clause = "" params: Dict[str, Union[int, str]] = { 'skip': skip_count, 'limit': page_size, } if search: where_clause = ( "WHERE n.name_zh CONTAINS $search OR " "n.description CONTAINS $search" ) params['search'] = search # 查询数据流列表（包含标签数组） # 使用WITH子句先分页，再聚合标签，避免分页结果不准确 query = f""" MATCH (n:DataFlow) {where_clause} WITH n ORDER BY n.created_at DESC SKIP $skip LIMIT $limit OPTIONAL MATCH (n)-[:LABEL]->(label:DataLabel) RETURN n, id(n) as node_id, n.created_at as created_at, collect({{ id: id(label), name_zh: label.name_zh, name_en: label.name_en }}) as tags """ # 获取Neo4j驱动（如果连接失败会抛出ConnectionError异常） try: with connect_graph().session() as session: list_result = session.run(query, params).data() # 查询总数 count_query = f""" MATCH (n:DataFlow) {where_clause} RETURN count(n) as total """ count_params = {'search': search} if search else {} count_result = session.run( count_query, count_params).single() total = count_result['total'] if count_result else 0 except Exception as e: # 确保 driver 被正确关闭，避免资源泄漏 - 这里不再需要手动关闭 # driver，因为connect_graph可能返回单例或新实例。如果是新实例， # 我们没有引用它去关闭；若connect_graph每次返回新实例且需要关闭， # 之前的代码是对的。如果connect_graph返回单例，则不应关闭。 # 用户反馈：The driver.close() call prematurely closes a shared # driver instance，所以直接使用 session，并不关闭 driver。 logger.error(f"查询数据流失败: {str(e)}") raise e # 格式化结果 dataflows = [] for record in list_result: node = record['n'] dataflow = dict(node) dataflow['id'] = record['node_id'] # 使用查询返回的node_id # 处理标签数组，过滤掉空标签 tags = record.get('tags', []) dataflow['tag'] = [ tag for tag in tags if tag.get('id') is not None ] dataflows.append(dataflow) return { 'list': dataflows, 'pagination': { 'page': page, 'page_size': page_size, 'total': total, 'total_pages': (total + page_size - 1) // page_size } } except Exception as e: logger.error(f"获取数据流列表失败: {str(e)}") raise e @staticmethod def get_dataflow_by_id(dataflow_id: int) -> Optional[Dict[str, Any]]: """ 根据ID获取数据流详情 Args: dataflow_id: 数据流ID Returns: 数据流详情字典，如果不存在则返回None """ try: # 从Neo4j获取DataFlow节点的所有属性（包含标签数组） neo4j_query = """ MATCH (n:DataFlow) WHERE id(n) = $dataflow_id OPTIONAL MATCH (n)-[:LABEL]->(label:DataLabel) RETURN n, id(n) as node_id, collect({ id: id(label), name_zh: label.name_zh, name_en: label.name_en }) as tags """ with connect_graph().session() as session: neo4j_result = session.run( neo4j_query, dataflow_id=dataflow_id).data() if not neo4j_result: logger.warning(f"未找到ID为 {dataflow_id} 的DataFlow节点") return None record = neo4j_result[0] node = record['n'] # 将节点属性转换为字典 dataflow = dict(node) dataflow['id'] = record['node_id'] # 处理标签数组，过滤掉空标签 tags = record.get('tags', []) dataflow['tag'] = [ tag for tag in tags if tag.get('id') is not None ] # 处理 script_requirement：如果是JSON字符串，解析为对象 script_requirement_str = dataflow.get('script_requirement', '') if script_requirement_str: try: # 尝试解析JSON字符串 script_requirement_obj = json.loads( script_requirement_str) dataflow['script_requirement'] = script_requirement_obj logger.debug( "成功解析script_requirement: %s", script_requirement_obj, ) except (json.JSONDecodeError, TypeError) as e: logger.warning(f"script_requirement解析失败，保持原值: {e}") # 保持原值（字符串） dataflow['script_requirement'] = script_requirement_str else: # 如果为空，设置为None dataflow['script_requirement'] = None logger.info( "成功获取DataFlow详情，ID: %s, 名称: %s", dataflow_id, dataflow.get('name_zh'), ) return dataflow except Exception as e: logger.error(f"获取数据流详情失败: {str(e)}") raise e @staticmethod def create_dataflow(data: Dict[str, Any]) -> Dict[str, Any]: """ 创建新的数据流 Args: data: 数据流配置数据 Returns: 创建的数据流信息 """ try: # 验证必填字段 required_fields = ['name_zh', 'describe'] for field in required_fields: if field not in data: raise ValueError(f"缺少必填字段: {field}") dataflow_name = data['name_zh'] # 使用LLM翻译名称生成英文名 try: result_list = translate_and_parse(dataflow_name) name_en = ( result_list[0] if result_list else dataflow_name.lower().replace(' ', '_') ) except Exception as e: logger.warning(f"翻译失败，使用默认英文名: {str(e)}") name_en = dataflow_name.lower().replace(' ', '_') # 处理 script_requirement，将其转换为 JSON 字符串 script_requirement = data.get('script_requirement', None) if script_requirement is not None: # 如果是字典或列表，转换为 JSON 字符串 if isinstance(script_requirement, (dict, list)): script_requirement_str = json.dumps( script_requirement, ensure_ascii=False) else: # 如果已经是字符串，直接使用 script_requirement_str = str(script_requirement) else: script_requirement_str = '' # 准备节点数据（tag不作为节点属性存储，而是通过LABEL关系关联） node_data = { 'name_zh': dataflow_name, 'name_en': name_en, 'category': data.get('category', ''), 'organization': data.get('organization', ''), 'leader': data.get('leader', ''), 'frequency': data.get('frequency', ''), 'describe': data.get('describe', ''), 'status': data.get('status', 'inactive'), 'update_mode': data.get('update_mode', 'append'), 'script_type': data.get('script_type', 'python'), 'script_requirement': script_requirement_str, 'created_at': get_formatted_time(), 'updated_at': get_formatted_time() } # 创建或获取数据流节点 dataflow_id = get_node('DataFlow', name=dataflow_name) if dataflow_id: raise ValueError(f"数据流 '{dataflow_name}' 已存在") dataflow_id = create_or_get_node('DataFlow', **node_data) # 处理标签关系（支持多标签数组） tag_list = data.get('tag', []) if tag_list: try: DataFlowService._handle_tag_relationships( dataflow_id, tag_list) except Exception as e: logger.warning(f"处理标签关系时出错: {str(e)}") # 成功创建图数据库节点后，写入PG数据库 try: DataFlowService._save_to_pg_database( data, dataflow_name, name_en) logger.info(f"数据流信息已写入PG数据库: {dataflow_name}") # PG数据库记录成功写入后，在neo4j图数据库中创建script关系 try: DataFlowService._handle_script_relationships( data, dataflow_name, name_en) logger.info(f"脚本关系创建成功: {dataflow_name}") except Exception as script_error: logger.warning(f"创建脚本关系失败: {str(script_error)}") except Exception as pg_error: logger.error(f"写入PG数据库失败: {str(pg_error)}") # 注意：这里可以选择回滚图数据库操作，但目前保持图数据库数据 # 在实际应用中，可能需要考虑分布式事务 # 返回创建的数据流信息 # 查询创建的节点获取完整信息 query = ( "MATCH (n:DataFlow {name_zh: $name_zh}) " "RETURN n, id(n) as node_id" ) with connect_graph().session() as session: id_result = session.run(query, name_zh=dataflow_name).single() if id_result: dataflow_node = id_result['n'] node_id = id_result['node_id'] # 将节点属性转换为字典 result = dict(dataflow_node) result['id'] = node_id else: # 如果查询失败，返回基本信息 result = { 'id': ( dataflow_id if isinstance(dataflow_id, int) else None ), 'name_zh': dataflow_name, 'name_en': name_en, 'created_at': get_formatted_time(), } logger.info(f"创建数据流成功: {dataflow_name}") return result except Exception as e: logger.error(f"创建数据流失败: {str(e)}") raise e @staticmethod def _save_to_pg_database( data: Dict[str, Any], script_name: str, name_en: str, ): """ 将脚本信息保存到PG数据库 Args: data: 包含脚本信息的数据 script_name: 脚本名称 name_en: 英文名称 """ try: # 提取脚本相关信息 # 处理 script_requirement，确保保存为 JSON 字符串 script_requirement_raw = data.get('script_requirement', None) # 用于保存从 script_requirement 中提取的 rule rule_from_requirement = '' if script_requirement_raw is not None: # 如果是字典，提取 rule 字段 if isinstance(script_requirement_raw, dict): rule_from_requirement = script_requirement_raw.get( 'rule', '') script_requirement = json.dumps( script_requirement_raw, ensure_ascii=False) elif isinstance(script_requirement_raw, list): script_requirement = json.dumps( script_requirement_raw, ensure_ascii=False) else: # 如果已经是字符串，尝试解析以提取 rule script_requirement = str(script_requirement_raw) try: parsed_req = json.loads(script_requirement) if isinstance(parsed_req, dict): rule_from_requirement = parsed_req.get('rule', '') except (json.JSONDecodeError, TypeError): pass else: script_requirement = '' # 处理 script_content：优先使用前端传入的值，如果为空则使用从 script_requirement 提取的 rule script_content = data.get('script_content', '') if not script_content and rule_from_requirement: script_content = rule_from_requirement logger.info( "script_content为空，使用从script_requirement提取的rule: %s", rule_from_requirement, ) # 安全处理 source_table 和 target_table（避免 None 值导致的 'in' 操作错误） source_table_raw = data.get('source_table') or '' source_table = ( source_table_raw.split(':')[-1] if ':' in source_table_raw else source_table_raw ) target_table_raw = data.get('target_table') or '' target_table = ( target_table_raw.split(':')[-1] if ':' in target_table_raw else (target_table_raw or name_en) ) script_type = data.get('script_type', 'python') user_name = data.get('created_by', 'system') target_dt_column = data.get('target_dt_column', '') # 验证必需字段 if not target_table: target_table = name_en if not script_name: raise ValueError("script_name不能为空") # 构建插入SQL insert_sql = text( """ INSERT INTO dags.data_transform_scripts (source_table, target_table, script_name, script_type, script_requirement, script_content, user_name, create_time, update_time, target_dt_column) VALUES (:source_table, :target_table, :script_name, :script_type, :script_requirement, :script_content, :user_name, :create_time, :update_time, :target_dt_column) ON CONFLICT (target_table, script_name) DO UPDATE SET source_table = EXCLUDED.source_table, script_type = EXCLUDED.script_type, script_requirement = EXCLUDED.script_requirement, script_content = EXCLUDED.script_content, user_name = EXCLUDED.user_name, update_time = EXCLUDED.update_time, target_dt_column = EXCLUDED.target_dt_column """ ) # 准备参数 current_time = datetime.now() params = { 'source_table': source_table, 'target_table': target_table, 'script_name': script_name, 'script_type': script_type, 'script_requirement': script_requirement, 'script_content': script_content, 'user_name': user_name, 'create_time': current_time, 'update_time': current_time, 'target_dt_column': target_dt_column } # 执行插入操作 db.session.execute(insert_sql, params) # 新增：保存到task_list表 try: # 1. 解析script_requirement并构建详细的任务描述 task_description_md = script_requirement try: # 尝试解析JSON try: req_json = json.loads(script_requirement) except (json.JSONDecodeError, TypeError): req_json = None if isinstance(req_json, dict): # 1. 从script_requirement中提取rule字段作为request_content_str request_content_str = req_json.get('rule', '') # 2. 从script_requirement中提取source_table和 # target_table字段信息 source_table_ids = req_json.get('source_table', []) target_table_ids = req_json.get('target_table', []) # 确保是列表格式 if not isinstance(source_table_ids, list): source_table_ids = [ source_table_ids] if source_table_ids else [] if not isinstance(target_table_ids, list): target_table_ids = [ target_table_ids] if target_table_ids else [] # 合并所有BusinessDomain ID all_bd_ids = source_table_ids + target_table_ids # 4. 从data参数中提取update_mode update_mode = data.get('update_mode', 'append') # 生成Business Domain DDLs source_ddls = [] target_ddls = [] data_source_info = None if all_bd_ids: try: with connect_graph().session() as session: # 处理source tables for bd_id in source_table_ids: ddl_info = ( DataFlowService ._generate_businessdomain_ddl( session, bd_id, is_target=False, ) ) if ddl_info: source_ddls.append(ddl_info['ddl']) # 3. 如果BELONGS_TO关系连接的是 # "数据资源"，获取数据源信息 if ( ddl_info.get('data_source') and not data_source_info ): data_source_info = ddl_info[ 'data_source' ] # 处理target tables（5. 目标表缺省要有create_time字段） for bd_id in target_table_ids: ddl_info = ( DataFlowService ._generate_businessdomain_ddl( session, bd_id, is_target=True, update_mode=update_mode, ) ) if ddl_info: target_ddls.append(ddl_info['ddl']) # 同样检查BELONGS_TO关系，获取数据源信息 if ( ddl_info.get('data_source') and not data_source_info ): data_source_info = ddl_info[ 'data_source' ] except Exception as neo_e: logger.error( f"获取BusinessDomain DDL失败: {str(neo_e)}") # 构建Markdown格式的任务描述 task_desc_parts = [f"# Task: {script_name}\n"] # 添加数据源信息 if data_source_info: task_desc_parts.append("## Data Source") task_desc_parts.append( f"- **Type**: " f"{data_source_info.get('type', 'N/A')}" ) task_desc_parts.append( f"- **Host**: " f"{data_source_info.get('host', 'N/A')}" ) task_desc_parts.append( f"- **Port**: " f"{data_source_info.get('port', 'N/A')}" ) task_desc_parts.append( f"- **Database**: " f"{data_source_info.get('database', 'N/A')}\n" ) # 添加源表DDL if source_ddls: task_desc_parts.append("## Source Tables (DDL)") for ddl in source_ddls: task_desc_parts.append(f"```sql\n{ddl}\n```\n") # 添加目标表DDL if target_ddls: task_desc_parts.append("## Target Tables (DDL)") for ddl in target_ddls: task_desc_parts.append(f"```sql\n{ddl}\n```\n") # 添加更新模式说明 task_desc_parts.append("## Update Mode") if update_mode == 'append': task_desc_parts.append("- **Mode**: Append (追加模式)") task_desc_parts.append( "- **Description**: 新数据将追加到目标表，不删除现有数据\n") else: task_desc_parts.append( "- **Mode**: Full Refresh (全量更新)") task_desc_parts.append( "- **Description**: 目标表将被清空后重新写入数据\n") # 添加请求内容（rule） if request_content_str: task_desc_parts.append("## Request Content") task_desc_parts.append(f"{request_content_str}\n") # 添加实施步骤（根据任务类型优化） task_desc_parts.append("## Implementation Steps") # 判断是否为远程数据源导入任务 if data_source_info: # 从远程数据源导入数据的简化步骤 task_desc_parts.append( "1. Create an n8n workflow to execute the " "data import task" ) task_desc_parts.append( "2. Configure the workflow to call " "`import_resource_data.py` Python script" ) task_desc_parts.append( "3. Pass the following parameters to the " "Python execution node:" ) task_desc_parts.append( " - `--source-config`: JSON configuration " "for the remote data source" ) task_desc_parts.append( " - `--target-table`: Target table name " "(data resource English name)" ) task_desc_parts.append( f" - `--update-mode`: {update_mode}") task_desc_parts.append( "4. The Python script will automatically:") task_desc_parts.append( " - Connect to the remote data source") task_desc_parts.append( " - Extract data from the source table") task_desc_parts.append( f" - Write data to target table using " f"{update_mode} mode" ) else: # 数据转换任务的完整步骤 task_desc_parts.append( "1. Extract data from source tables as " "specified in the DDL" ) task_desc_parts.append( "2. Apply transformation logic according " "to the rule:" ) if request_content_str: task_desc_parts.append( f" - Rule: {request_content_str}") task_desc_parts.append( "3. Generate Python program to implement the " "data transformation logic" ) task_desc_parts.append( f"4. Write transformed data to target table " f"using {update_mode} mode" ) task_desc_parts.append( "5. Create an n8n workflow to schedule and " "execute the Python program" ) task_description_md = "\n".join(task_desc_parts) except Exception as parse_e: logger.warning(f"解析任务描述详情失败，使用原始描述: {str(parse_e)}") task_description_md = script_requirement # 假设运行根目录为项目根目录，dataflows.py在app/core/data_flow/ code_path = 'app/core/data_flow' task_insert_sql = text( "INSERT INTO public.task_list\n" "(task_name, task_description, status, code_name, " "code_path, create_by, create_time, update_time)\n" "VALUES\n" "(:task_name, :task_description, :status, :code_name, " ":code_path, :create_by, :create_time, :update_time)" ) task_params = { 'task_name': script_name, 'task_description': task_description_md, 'status': 'pending', 'code_name': script_name, 'code_path': code_path, 'create_by': 'cursor', 'create_time': current_time, 'update_time': current_time } # 使用嵌套事务，确保task_list插入失败不影响主流程 with db.session.begin_nested(): db.session.execute(task_insert_sql, task_params) logger.info(f"成功将任务信息写入task_list表: task_name={script_name}") except Exception as task_error: # 记录错误但不中断主流程 logger.error(f"写入task_list表失败: {str(task_error)}") # 如果要求必须成功写入任务列表，则这里应该raise task_error # raise task_error db.session.commit() logger.info( "成功将脚本信息写入PG数据库: target_table=%s, script_name=%s", target_table, script_name, ) except Exception as e: db.session.rollback() logger.error(f"写入PG数据库失败: {str(e)}") raise e @staticmethod def _handle_children_relationships(dataflow_node, children_ids): """处理子节点关系""" logger.debug( "处理子节点关系，原始children_ids: %s, 类型: %s", children_ids, type(children_ids), ) # 确保children_ids是列表格式 if not isinstance(children_ids, (list, tuple)): if children_ids is not None: children_ids = [children_ids] # 如果是单个值，转换为列表 logger.debug(f"将单个值转换为列表: {children_ids}") else: children_ids = [] # 如果是None，转换为空列表 logger.debug("将None转换为空列表") for child_id in children_ids: try: # 查找子节点 query = "MATCH (n) WHERE id(n) = $child_id RETURN n" with connect_graph().session() as session: result = session.run(query, child_id=child_id).data() if result: # 获取dataflow_node的ID dataflow_id = getattr(dataflow_node, 'identity', None) if dataflow_id is None: # 如果没有identity属性，从名称查询ID query_id = ( "MATCH (n:DataFlow) WHERE n.name_zh = " "$name_zh RETURN id(n) as node_id" ) id_result = session.run( query_id, name_zh=dataflow_node.get('name_zh'), ).single() dataflow_id = ( id_result['node_id'] if id_result else None ) # 创建关系 - 使用ID调用relationship_exists if dataflow_id and not relationship_exists( dataflow_id, 'child', child_id ): session.run( "MATCH (a), (b) WHERE id(a) = $dataflow_id " "AND id(b) = $child_id " "CREATE (a)-[:child]->(b)", dataflow_id=dataflow_id, child_id=child_id, ) logger.info( f"创建子节点关系: {dataflow_id} -> {child_id}") except Exception as e: logger.warning(f"创建子节点关系失败 {child_id}: {str(e)}") @staticmethod def _handle_tag_relationships(dataflow_id, tag_list): """ 处理多标签关系 Args: dataflow_id: 数据流节点ID tag_list: 标签列表，可以是ID数组或包含id字段的对象数组 """ # 确保tag_list是列表格式 if not isinstance(tag_list, list): tag_list = [tag_list] if tag_list else [] for tag_item in tag_list: tag_id = None if isinstance(tag_item, dict) and 'id' in tag_item: tag_id = int(tag_item['id']) elif isinstance(tag_item, (int, str)): try: tag_id = int(tag_item) except (ValueError, TypeError): pass if tag_id: DataFlowService._handle_single_tag_relationship( dataflow_id, tag_id) @staticmethod def _handle_single_tag_relationship(dataflow_id, tag_id): """处理单个标签关系""" try: # 查找标签节点 query = "MATCH (n:DataLabel) WHERE id(n) = $tag_id RETURN n" with connect_graph().session() as session: result = session.run(query, tag_id=tag_id).data() if result: # 创建关系 - 使用ID调用relationship_exists if dataflow_id and not relationship_exists( dataflow_id, 'LABEL', tag_id ): session.run( "MATCH (a), (b) WHERE id(a) = $dataflow_id " "AND id(b) = $tag_id " "CREATE (a)-[:LABEL]->(b)", dataflow_id=dataflow_id, tag_id=tag_id, ) logger.info(f"创建标签关系: {dataflow_id} -> {tag_id}") except Exception as e: logger.warning(f"创建标签关系失败 {tag_id}: {str(e)}") @staticmethod def update_dataflow( dataflow_id: int, data: Dict[str, Any], ) -> Optional[Dict[str, Any]]: """ 更新数据流 Args: dataflow_id: 数据流ID data: 更新的数据 Returns: 更新后的数据流信息，如果不存在则返回None """ try: # 提取 tag 数组（不作为节点属性存储） tag_list = data.pop('tag', None) # 查找节点 query = "MATCH (n:DataFlow) WHERE id(n) = $dataflow_id RETURN n" with connect_graph().session() as session: result = session.run(query, dataflow_id=dataflow_id).data() if not result: return None # 更新节点属性 update_fields = [] params: Dict[str, Any] = {'dataflow_id': dataflow_id} for key, value in data.items(): if key not in ['id', 'created_at']: # 保护字段 # 复杂对象序列化为 JSON 字符串 if key in ['config', 'script_requirement']: if isinstance(value, dict): value = json.dumps(value, ensure_ascii=False) update_fields.append(f"n.{key} = ${key}") params[key] = value if update_fields: params['updated_at'] = get_formatted_time() update_fields.append("n.updated_at = $updated_at") update_query = f""" MATCH (n:DataFlow) WHERE id(n) = $dataflow_id SET {', '.join(update_fields)} RETURN n, id(n) as node_id """ result = session.run(update_query, params).data() # 处理 tag 关系（支持多标签数组） if tag_list is not None: # 确保是列表格式 if not isinstance(tag_list, list): tag_list = [tag_list] if tag_list else [] # 先删除现有的 LABEL 关系 delete_query = """ MATCH (n:DataFlow)-[r:LABEL]->(:DataLabel) WHERE id(n) = $dataflow_id DELETE r """ session.run(delete_query, dataflow_id=dataflow_id) logger.info(f"删除数据流 {dataflow_id} 的现有标签关系") # 为每个 tag 创建新的 LABEL 关系 for tag_item in tag_list: tag_id = None if isinstance(tag_item, dict) and 'id' in tag_item: tag_id = int(tag_item['id']) elif isinstance(tag_item, (int, str)): try: tag_id = int(tag_item) except (ValueError, TypeError): pass if tag_id: DataFlowService._handle_single_tag_relationship( dataflow_id, tag_id ) if result: node = result[0]['n'] updated_dataflow = dict(node) # 使用查询返回的node_id updated_dataflow['id'] = result[0]['node_id'] # 查询并添加标签数组到返回数据 tags_query = """ MATCH (n:DataFlow) WHERE id(n) = $dataflow_id OPTIONAL MATCH (n)-[:LABEL]->(label:DataLabel) RETURN collect({ id: id(label), name_zh: label.name_zh, name_en: label.name_en }) as tags """ tags_result = session.run( tags_query, dataflow_id=dataflow_id ).single() if tags_result: tags = tags_result.get('tags', []) updated_dataflow['tag'] = [ tag for tag in tags if tag.get('id') is not None ] else: updated_dataflow['tag'] = [] logger.info(f"更新数据流成功: ID={dataflow_id}") return updated_dataflow return None except Exception as e: logger.error(f"更新数据流失败: {str(e)}") raise e @staticmethod def delete_dataflow(dataflow_id: int) -> bool: """ 删除数据流 Args: dataflow_id: 数据流ID Returns: 删除是否成功 """ try: # 删除节点及其关系 query = """ MATCH (n:DataFlow) WHERE id(n) = $dataflow_id DETACH DELETE n RETURN count(n) as deleted_count """ with connect_graph().session() as session: delete_result = session.run( query, dataflow_id=dataflow_id).single() result = delete_result['deleted_count'] if delete_result else 0 if result and result > 0: logger.info(f"删除数据流成功: ID={dataflow_id}") return True return False except Exception as e: logger.error(f"删除数据流失败: {str(e)}") raise e @staticmethod def execute_dataflow( dataflow_id: int, params: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """ 执行数据流 Args: dataflow_id: 数据流ID params: 执行参数 Returns: 执行结果信息 """ try: # 检查数据流是否存在 query = "MATCH (n:DataFlow) WHERE id(n) = $dataflow_id RETURN n" with connect_graph().session() as session: result = session.run(query, dataflow_id=dataflow_id).data() if not result: raise ValueError(f"数据流不存在: ID={dataflow_id}") execution_id = ( f"exec_{dataflow_id}_{int(datetime.now().timestamp())}" ) # TODO: 这里应该实际执行数据流 # 目前返回模拟结果 result = { 'execution_id': execution_id, 'dataflow_id': dataflow_id, 'status': 'running', 'started_at': datetime.now().isoformat(), 'params': params or {}, 'progress': 0 } logger.info( "开始执行数据流: ID=%s, execution_id=%s", dataflow_id, execution_id, ) return result except Exception as e: logger.error(f"执行数据流失败: {str(e)}") raise e @staticmethod def get_dataflow_status(dataflow_id: int) -> Dict[str, Any]: """ 获取数据流执行状态 Args: dataflow_id: 数据流ID Returns: 执行状态信息 """ try: # TODO: 这里应该查询实际的执行状态 # 目前返回模拟状态 query = "MATCH (n:DataFlow) WHERE id(n) = $dataflow_id RETURN n" with connect_graph().session() as session: result = session.run(query, dataflow_id=dataflow_id).data() if not result: raise ValueError(f"数据流不存在: ID={dataflow_id}") status = ['running', 'completed', 'failed', 'pending'][ dataflow_id % 4 ] return { 'dataflow_id': dataflow_id, 'status': status, 'progress': ( 100 if status == 'completed' else (dataflow_id * 10) % 100 ), 'started_at': datetime.now().isoformat(), 'completed_at': ( datetime.now().isoformat() if status == 'completed' else None ), 'error_message': ( '执行过程中发生错误' if status == 'failed' else None ), } except Exception as e: logger.error(f"获取数据流状态失败: {str(e)}") raise e @staticmethod def get_dataflow_logs( dataflow_id: int, page: int = 1, page_size: int = 50, ) -> Dict[str, Any]: """ 获取数据流执行日志 Args: dataflow_id: 数据流ID page: 页码 page_size: 每页大小 Returns: 执行日志列表和分页信息 """ try: # TODO: 这里应该查询实际的执行日志 # 目前返回模拟日志 query = "MATCH (n:DataFlow) WHERE id(n) = $dataflow_id RETURN n" with connect_graph().session() as session: result = session.run(query, dataflow_id=dataflow_id).data() if not result: raise ValueError(f"数据流不存在: ID={dataflow_id}") mock_logs = [ { 'id': i, 'timestamp': datetime.now().isoformat(), 'level': ['INFO', 'WARNING', 'ERROR'][i % 3], 'message': f'数据流执行日志消息 {i}', 'component': ['source', 'transform', 'target'][i % 3] } for i in range(1, 101) ] # 分页处理 total = len(mock_logs) start = (page - 1) * page_size end = start + page_size logs = mock_logs[start:end] return { 'logs': logs, 'pagination': { 'page': page, 'page_size': page_size, 'total': total, 'total_pages': (total + page_size - 1) // page_size } } except Exception as e: logger.error(f"获取数据流日志失败: {str(e)}") raise e @staticmethod def create_script(request_data: Union[Dict[str, Any], str]) -> str: """ 使用Deepseek模型生成SQL脚本 Args: request_data: 包含input, output, request_content的请求数据字典，或JSON字符串 Returns: 生成的SQL脚本内容 """ try: logger.info(f"开始处理脚本生成请求: {request_data}") logger.info(f"request_data类型: {type(request_data)}") # 类型检查和处理 if isinstance(request_data, str): logger.warning(f"request_data是字符串，尝试解析为JSON: {request_data}") try: import json request_data = json.loads(request_data) except json.JSONDecodeError as e: raise ValueError(f"无法解析request_data为JSON: {str(e)}") if not isinstance(request_data, dict): raise ValueError( f"request_data必须是字典类型，实际类型: {type(request_data)}") # 1. 从传入的request_data中解析input, output, request_content内容 input_data = request_data.get('input', '') output_data = request_data.get('output', '') request_content = request_data.get('request_data', '') # 如果request_content是HTML格式，提取纯文本 if request_content and ( request_content.startswith('

') or '<' in request_content ): # 简单的HTML标签清理 import re request_content = re.sub( r'<[^>]+>', '', request_content).strip() if not input_data or not output_data or not request_content: raise ValueError( "缺少必要参数：input='{}', output='{}', " "request_content='{}' 不能为空".format( input_data, output_data, request_content[:100] if request_content else '', ) ) logger.info( "解析得到 - input: %s, output: %s, request_content: %s", input_data, output_data, request_content, ) # 2. 解析input中的多个数据表并生成源表DDL source_tables_ddl = [] input_tables = [] if input_data: tables = [table.strip() for table in input_data.split(',') if table.strip()] for table in tables: ddl = DataFlowService._parse_table_and_get_ddl( table, 'input') if ddl: input_tables.append(table) source_tables_ddl.append(ddl) else: logger.warning(f"无法获取输入表 {table} 的DDL结构") # 3. 解析output中的数据表并生成目标表DDL target_table_ddl = "" if output_data: target_table_ddl = DataFlowService._parse_table_and_get_ddl( output_data.strip(), 'output') if not target_table_ddl: logger.warning(f"无法获取输出表 {output_data} 的DDL结构") # 4. 按照Deepseek-prompt.txt的框架构建提示语 prompt_parts = [] # 开场白 - 角色定义 prompt_parts.append( "你是一名数据库工程师，正在构建一个PostgreSQL数据中的汇总逻辑。" "请为以下需求生成一段标准的 PostgreSQL SQL 脚本：" ) # 动态生成源表部分（第1点） for i, (table, ddl) in enumerate( zip(input_tables, source_tables_ddl), 1 ): table_name = table.split(':')[-1] if ':' in table else table prompt_parts.append(f"{i}.有一个源表: {table_name}，它的定义语句如下：") prompt_parts.append(ddl) prompt_parts.append("") # 添加空行分隔 # 动态生成目标表部分（第2点） if target_table_ddl: target_table_name = output_data.split( ':')[-1] if ':' in output_data else output_data next_index = len(input_tables) + 1 prompt_parts.append( f"{next_index}.有一个目标表：{target_table_name}，它的定义语句如下：") prompt_parts.append(target_table_ddl) prompt_parts.append("") # 添加空行分隔 # 动态生成处理逻辑部分（第3点） next_index = ( len(input_tables) + 2 if target_table_ddl else len(input_tables) + 1 ) prompt_parts.append(f"{next_index}.处理逻辑为：{request_content}") prompt_parts.append("") # 添加空行分隔 # 固定的技术要求部分（第4-8点） tech_requirements = [ ( f"{next_index + 1}.脚本应使用标准的 PostgreSQL 语法，" "适合在 Airflow、Python 脚本、或调度系统中调用；" ), f"{next_index + 2}.无需使用 UPSERT 或 ON CONFLICT", f"{next_index + 3}.请直接输出SQL，无需进行解释。", ( f"{next_index + 4}.请给这段sql起个英文名，不少于三个英文单词，使用\"_\"分隔，" "采用蛇形命名法。把sql的名字作为注释写在返回的sql中。" ), ( f"{next_index + 5}.生成的sql在向目标表插入数据的时候，向create_time字段写入当前日期" "时间now(),不用处理update_time字段" ), ] prompt_parts.extend(tech_requirements) # 组合完整的提示语 full_prompt = "\n".join(prompt_parts) logger.info(f"构建的完整提示语长度: {len(full_prompt)}") logger.info(f"完整提示语内容: {full_prompt}") # 5. 调用LLM生成SQL脚本 logger.info("开始调用Deepseek模型生成SQL脚本") script_content = llm_sql(full_prompt) if not script_content: raise ValueError("Deepseek模型返回空内容") # 确保返回的是文本格式 if not isinstance(script_content, str): script_content = str(script_content) logger.info(f"SQL脚本生成成功，内容长度: {len(script_content)}") return script_content except Exception as e: logger.error(f"生成SQL脚本失败: {str(e)}") raise e @staticmethod def _parse_table_and_get_ddl(table_str: str, table_type: str) -> str: """ 解析表格式（A:B）并从Neo4j查询元数据生成DDL Args: table_str: 表格式字符串，格式为"label:name_en" table_type: 表类型，用于日志记录（input/output） Returns: DDL格式的表结构字符串 """ try: # 解析A:B格式 if ':' not in table_str: logger.error(f"表格式错误，应为'label:name_en'格式: {table_str}") return "" parts = table_str.split(':', 1) if len(parts) != 2: logger.error(f"表格式解析失败: {table_str}") return "" label = parts[0].strip() name_en = parts[1].strip() if not label or not name_en: logger.error(f"标签或英文名为空: label={label}, name_en={name_en}") return "" logger.info(f"开始查询{table_type}表: label={label}, name_en={name_en}") # 从Neo4j查询节点及其关联的元数据 with connect_graph().session() as session: # 查询节点及其关联的元数据 cypher = f""" MATCH (n:{label} {{name_en: $name_en}}) OPTIONAL MATCH (n)-[:INCLUDES]->(m:DataMeta) RETURN n, collect(m) as metadata """ result = session.run( cypher, # type: ignore[arg-type] {'name_en': name_en}, ) record = result.single() if not record: logger.error(f"未找到节点: label={label}, name_en={name_en}") return "" node = record['n'] metadata = record['metadata'] logger.info(f"找到节点，关联元数据数量: {len(metadata)}") # 生成DDL格式的表结构 ddl_lines = [] ddl_lines.append(f"CREATE TABLE {name_en} (") if metadata: column_definitions = [] for meta in metadata: if meta: # 确保meta不为空 meta_props = dict(meta) column_name = meta_props.get( 'name_en', meta_props.get('name_zh', 'unknown_column'), ) data_type = meta_props.get( 'data_type', 'VARCHAR(255)') comment = meta_props.get('name_zh', '') # 构建列定义 column_def = f" {column_name} {data_type}" if comment: column_def += f" COMMENT '{comment}'" column_definitions.append(column_def) if column_definitions: ddl_lines.append(",\n".join(column_definitions)) else: ddl_lines.append( " id BIGINT PRIMARY KEY COMMENT '主键ID'") else: # 如果没有元数据，添加默认列 ddl_lines.append( " id BIGINT PRIMARY KEY COMMENT '主键ID'") ddl_lines.append(");") # 添加表注释 node_props = dict(node) table_comment = node_props.get( 'name_zh', node_props.get('describe', name_en)) if table_comment and table_comment != name_en: ddl_lines.append( f"COMMENT ON TABLE {name_en} IS '{table_comment}';") ddl_content = "\n".join(ddl_lines) logger.info(f"{table_type}表DDL生成成功: {name_en}") logger.debug(f"生成的DDL: {ddl_content}") return ddl_content except Exception as e: logger.error(f"解析表格式和生成DDL失败: {str(e)}") return "" @staticmethod def _generate_businessdomain_ddl( session, bd_id: int, is_target: bool = False, update_mode: str = 'append', ) -> Optional[Dict[str, Any]]: """ 根据BusinessDomain节点ID生成DDL Args: session: Neo4j session对象 bd_id: BusinessDomain节点ID is_target: 是否为目标表（目标表需要添加create_time字段） update_mode: 更新模式（append或full） Returns: 包含ddl和data_source信息的字典，如果节点不存在则返回None """ try: # 查询BusinessDomain节点、元数据、标签关系和数据源关系 cypher = """ MATCH (bd:BusinessDomain) WHERE id(bd) = $bd_id OPTIONAL MATCH (bd)-[:INCLUDES]->(m:DataMeta) OPTIONAL MATCH (bd)-[:LABEL]->(label:DataLabel) OPTIONAL MATCH (bd)-[:COME_FROM]->(ds:DataSource) RETURN bd, collect(DISTINCT m) as metadata, collect(DISTINCT { id: id(label), name_zh: label.name_zh, name_en: label.name_en }) as labels, ds.type as ds_type, ds.host as ds_host, ds.port as ds_port, ds.database as ds_database """ result = session.run(cypher, bd_id=bd_id).single() if not result or not result['bd']: logger.warning(f"未找到ID为 {bd_id} 的BusinessDomain节点") return None node = result['bd'] metadata = result['metadata'] # 处理标签数组，获取第一个有效标签名称用于判断 labels = result.get('labels', []) valid_labels = [ label for label in labels if label.get('id') is not None ] label_name = ( valid_labels[0].get('name_zh') if valid_labels else None ) # 生成DDL node_props = dict(node) table_name = node_props.get('name_en', f'table_{bd_id}') ddl_lines = [] ddl_lines.append(f"CREATE TABLE {table_name} (") column_definitions = [] # 添加元数据列 if metadata: for meta in metadata: if meta: meta_props = dict(meta) column_name = meta_props.get( 'name_en', meta_props.get('name_zh', 'unknown_column'), ) data_type = meta_props.get('data_type', 'VARCHAR(255)') comment = meta_props.get('name_zh', '') column_def = f" {column_name} {data_type}" if comment: column_def += f" COMMENT '{comment}'" column_definitions.append(column_def) # 如果没有元数据，添加默认主键 if not column_definitions: column_definitions.append( " id BIGINT PRIMARY KEY COMMENT '主键ID'") # 5. 如果是目标表，添加create_time字段 if is_target: column_definitions.append( " create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP " "COMMENT '数据创建时间'" ) ddl_lines.append(",\n".join(column_definitions)) ddl_lines.append(");") # 添加表注释 table_comment = node_props.get( 'name_zh', node_props.get('describe', table_name)) if table_comment and table_comment != table_name: ddl_lines.append( f"COMMENT ON TABLE {table_name} IS '{table_comment}';") ddl_content = "\n".join(ddl_lines) # 3. 检查BELONGS_TO关系是否连接"数据资源"，如果是则返回数据源信息 data_source = None if label_name == '数据资源' and result['ds_type']: data_source = { 'type': result['ds_type'], 'host': result['ds_host'], 'port': result['ds_port'], 'database': result['ds_database'] } logger.info(f"获取到数据源信息: {data_source}") logger.debug( f"生成BusinessDomain DDL成功: {table_name}, is_target={is_target}") return { 'ddl': ddl_content, 'table_name': table_name, 'data_source': data_source } except Exception as e: logger.error(f"生成BusinessDomain DDL失败，ID={bd_id}: {str(e)}") return None @staticmethod def _handle_script_relationships( data: Dict[str, Any], dataflow_name: str, name_en: str, ): """ 处理脚本关系，在Neo4j图数据库中创建从source_table到target_table之间的 DERIVED_FROM关系 Args: data: 包含脚本信息的数据字典，应包含script_name, script_type, schedule_status, source_table, target_table, update_mode """ try: # 从data中读取键值对 script_name = dataflow_name, script_type = data.get('script_type', 'sql') schedule_status = data.get('status', 'inactive') source_table_full = data.get('source_table', '') target_table_full = data.get('target_table', '') update_mode = data.get('update_mode', 'full') # 处理source_table和target_table的格式 source_table = source_table_full.split( ':')[-1] if ':' in source_table_full else source_table_full target_table = target_table_full.split( ':')[-1] if ':' in target_table_full else target_table_full source_label = source_table_full.split( ':')[0] if ':' in source_table_full else source_table_full target_label = target_table_full.split( ':')[0] if ':' in target_table_full else target_table_full # 验证必要字段 if not source_table or not target_table: logger.warning( "source_table或target_table为空，跳过关系创建: " "source_table=%s, target_table=%s", source_table, target_table, ) return logger.info(f"开始创建脚本关系: {source_table} -> {target_table}") with connect_graph().session() as session: # 创建或获取source和target节点 create_nodes_query = f""" MERGE (source:{source_label} {{name: $source_table}}) ON CREATE SET source.created_at = $created_at, source.type = 'source' WITH source MERGE (target:{target_label} {{name: $target_table}}) ON CREATE SET target.created_at = $created_at, target.type = 'target' RETURN source, target, id(source) as source_id, id(target) as target_id """ # 执行创建节点的查询 result = session.run( create_nodes_query, # type: ignore[arg-type] { 'source_table': source_table, 'target_table': target_table, 'created_at': get_formatted_time(), }, ).single() if result: source_id = result['source_id'] target_id = result['target_id'] # 检查并创建关系 create_relationship_query = f""" MATCH (source:{source_label}), (target:{target_label}) WHERE id(source) = $source_id AND id(target) = $target_id AND NOT EXISTS((target)-[:DERIVED_FROM]->(source)) CREATE (target)-[r:DERIVED_FROM]->(source) SET r.script_name = $script_name, r.script_type = $script_type, r.schedule_status = $schedule_status, r.update_mode = $update_mode, r.created_at = $created_at, r.updated_at = $created_at RETURN r """ relationship_result = session.run( create_relationship_query, # type: ignore[arg-type] { 'source_id': source_id, 'target_id': target_id, 'script_name': script_name, 'script_type': script_type, 'schedule_status': schedule_status, 'update_mode': update_mode, 'created_at': get_formatted_time(), }, ).single() if relationship_result: logger.info( "成功创建DERIVED_FROM关系: %s -> %s (script: %s)", target_table, source_table, script_name, ) else: logger.info( "DERIVED_FROM关系已存在: %s -> %s", target_table, source_table, ) else: logger.error( "创建表节点失败: source_table=%s, target_table=%s", source_table, target_table, ) except Exception as e: logger.error(f"处理脚本关系失败: {str(e)}") raise e @staticmethod def get_business_domain_list() -> List[Dict[str, Any]]: """ 获取BusinessDomain节点列表 Returns: BusinessDomain节点列表，每个节点包含 id, name_zh, name_en, tag """ try: logger.info("开始查询BusinessDomain节点列表") with connect_graph().session() as session: # 查询所有BusinessDomain节点及其LABEL关系指向的标签（支持多标签） query = """ MATCH (bd:BusinessDomain) OPTIONAL MATCH (bd)-[:LABEL]->(label:DataLabel) RETURN id(bd) as id, bd.name_zh as name_zh, bd.name_en as name_en, bd.create_time as create_time, collect({ id: id(label), name_zh: label.name_zh, name_en: label.name_en }) as tags ORDER BY create_time DESC """ result = session.run(query) bd_list = [] for record in result: # 处理标签数组，过滤掉空标签 tags = record.get("tags", []) tag_list = [ tag for tag in tags if tag.get('id') is not None ] bd_item = { "id": record["id"], "name_zh": record.get("name_zh", "") or "", "name_en": record.get("name_en", "") or "", "tag": tag_list, } bd_list.append(bd_item) logger.info(f"成功查询到 {len(bd_list)} 个BusinessDomain节点") return bd_list except Exception as e: logger.error(f"查询BusinessDomain节点列表失败: {str(e)}") raise e