DataOps-platform/app/core/data_flow/dataflows.py

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

            # 查询数据流列表
            query = f"""
            MATCH (n:DataFlow)
            {where_clause}
            RETURN n, id(n) as node_id
            ORDER BY n.created_at DESC
            SKIP $skip
            LIMIT $limit
            """

            # 获取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
                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
            RETURN n, id(n) as node_id
            """

            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']

                # 处理 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 = ''

            # 准备节点数据
            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', ''),
                'tag': data.get('tag', ''),
                '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_id = data.get('tag')
            if tag_id is not None:
                try:
                    DataFlowService._handle_tag_relationship(
                        dataflow_id, tag_id)
                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_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 and isinstance(tag_list, list):
                    # 先删除现有的 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_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']

                    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('<p>') 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)-[:BELONGS_TO]->(label:DataLabel)
            OPTIONAL MATCH (bd)-[:COME_FROM]->(ds:DataSource)
            RETURN bd,
                   collect(DISTINCT m) as metadata,
                   label.name_zh as label_name,
                   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']
            label_name = result['label_name']

            # 生成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节点及其BELONGS_TO关系指向的标签
                query = """
                MATCH (bd:BusinessDomain)
                OPTIONAL MATCH (bd)-[:BELONGS_TO]->(label:DataLabel)
                RETURN id(bd) as id,
                       bd.name_zh as name_zh,
                       bd.name_en as name_en,
                       label.name_zh as tag
                ORDER BY bd.create_time DESC
                """

                result = session.run(query)

                bd_list = []
                for record in result:
                    bd_item = {
                        "id": record["id"],
                        "name_zh": record.get("name_zh", "") or "",
                        "name_en": record.get("name_en", "") or "",
                        "tag": record.get("tag", "") or "",
                    }
                    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