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

import contextlib
import json
import logging
from datetime import datetime
from typing import Any, Dict, List, Optional, Union

from sqlalchemy import text

from app import db
from app.core.data_service.data_product_service import DataProductService
from app.core.graph.graph_operations import (
    connect_graph,
    create_or_get_node,
    get_node,
    relationship_exists,
)
from app.core.llm.llm_service import llm_sql
from app.core.meta_data import get_formatted_time, translate_and_parse

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")
            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(),
                    }

            # 注册数据产品到数据服务
            try:
                DataFlowService._register_data_product(
                    data=data,
                    dataflow_name=dataflow_name,
                    name_en=name_en,
                    dataflow_id=result.get("id"),
                )
                logger.info(f"数据产品注册成功: {dataflow_name}")
            except Exception as product_error:
                logger.warning(f"注册数据产品失败: {str(product_error)}")
                # 不影响主流程，仅记录警告

            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")
            # 用于保存从 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**: {data_source_info.get('type', 'N/A')}"
                            )
                            task_desc_parts.append(
                                f"- **Host**: {data_source_info.get('host', 'N/A')}"
                            )
                            task_desc_parts.append(
                                f"- **Port**: {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)):
                with contextlib.suppress(ValueError, TypeError):
                    tag_id = int(tag_item)

            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()

                # 创建关系 - 使用ID调用relationship_exists
                if (
                    result
                    and 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"] and 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)):
                            with contextlib.suppress(ValueError, TypeError):
                                tag_id = int(tag_item)

                        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)}") from 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)-[: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 BusinessDomain到DataFlow的
        INPUT关系，以及从DataFlow到target BusinessDomain的OUTPUT关系。

        关系模型：
        - (source:BusinessDomain)-[:INPUT]->(dataflow:DataFlow)
        - (dataflow:DataFlow)-[:OUTPUT]->(target:BusinessDomain)

        Args:
            data: 包含脚本信息的数据字典，应包含script_name, script_type,
                  schedule_status, source_table, target_table, update_mode
        """
        try:
            # 从data中读取键值对
            source_table_full = data.get("source_table", "")
            target_table_full = data.get("target_table", "")

            # 处理source_table和target_table的格式
            # 格式: "label:name" 或 直接 "name"
            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 "BusinessDomain"
            )
            target_label = (
                target_table_full.split(":")[0]
                if ":" in target_table_full
                else "BusinessDomain"
            )

            # 验证必要字段
            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(
                "开始创建INPUT/OUTPUT关系: %s -[INPUT]-> %s -[OUTPUT]-> %s",
                source_table,
                dataflow_name,
                target_table,
            )

            with connect_graph().session() as session:
                # 步骤1：获取DataFlow节点ID
                dataflow_query = """
                MATCH (df:DataFlow {name_zh: $dataflow_name})
                RETURN id(df) as dataflow_id
                """
                df_result = session.run(
                    dataflow_query,  # type: ignore[arg-type]
                    {"dataflow_name": dataflow_name},
                ).single()

                if not df_result:
                    logger.error(f"未找到DataFlow节点: {dataflow_name}")
                    return

                dataflow_id = df_result["dataflow_id"]

                # 步骤2：获取或创建source节点
                # 优先通过name_en匹配，其次通过name匹配
                source_query = f"""
                MATCH (source:{source_label})
                WHERE source.name_en = $source_table OR source.name = $source_table
                RETURN id(source) as source_id
                LIMIT 1
                """
                source_result = session.run(
                    source_query,  # type: ignore[arg-type]
                    {"source_table": source_table},
                ).single()

                if not source_result:
                    logger.warning(
                        "未找到source节点: %s，将创建新节点",
                        source_table,
                    )
                    # 创建source节点
                    create_source_query = f"""
                    CREATE (source:{source_label} {{
                        name: $source_table,
                        name_en: $source_table,
                        created_at: $created_at,
                        type: 'source'
                    }})
                    RETURN id(source) as source_id
                    """
                    source_result = session.run(
                        create_source_query,  # type: ignore[arg-type]
                        {
                            "source_table": source_table,
                            "created_at": get_formatted_time(),
                        },
                    ).single()

                source_id = source_result["source_id"] if source_result else None

                # 步骤3：获取或创建target节点
                target_query = f"""
                MATCH (target:{target_label})
                WHERE target.name_en = $target_table OR target.name = $target_table
                RETURN id(target) as target_id
                LIMIT 1
                """
                target_result = session.run(
                    target_query,  # type: ignore[arg-type]
                    {"target_table": target_table},
                ).single()

                if not target_result:
                    logger.warning(
                        "未找到target节点: %s，将创建新节点",
                        target_table,
                    )
                    # 创建target节点
                    create_target_query = f"""
                    CREATE (target:{target_label} {{
                        name: $target_table,
                        name_en: $target_table,
                        created_at: $created_at,
                        type: 'target'
                    }})
                    RETURN id(target) as target_id
                    """
                    target_result = session.run(
                        create_target_query,  # type: ignore[arg-type]
                        {
                            "target_table": target_table,
                            "created_at": get_formatted_time(),
                        },
                    ).single()

                target_id = target_result["target_id"] if target_result else None

                if not source_id or not target_id:
                    logger.error(
                        "无法获取source或target节点ID: source_id=%s, target_id=%s",
                        source_id,
                        target_id,
                    )
                    return

                # 步骤4：创建 INPUT 关系 (source)-[:INPUT]->(dataflow)
                create_input_query = """
                MATCH (source), (dataflow:DataFlow)
                WHERE id(source) = $source_id AND id(dataflow) = $dataflow_id
                MERGE (source)-[r:INPUT]->(dataflow)
                ON CREATE SET r.created_at = $created_at
                ON MATCH SET r.updated_at = $created_at
                RETURN r
                """
                input_result = session.run(
                    create_input_query,  # type: ignore[arg-type]
                    {
                        "source_id": source_id,
                        "dataflow_id": dataflow_id,
                        "created_at": get_formatted_time(),
                    },
                ).single()

                if input_result:
                    logger.info(
                        "成功创建INPUT关系: %s -> %s",
                        source_table,
                        dataflow_name,
                    )
                else:
                    logger.warning(
                        "INPUT关系创建失败或已存在: %s -> %s",
                        source_table,
                        dataflow_name,
                    )

                # 步骤5：创建 OUTPUT 关系 (dataflow)-[:OUTPUT]->(target)
                create_output_query = """
                MATCH (dataflow:DataFlow), (target)
                WHERE id(dataflow) = $dataflow_id AND id(target) = $target_id
                MERGE (dataflow)-[r:OUTPUT]->(target)
                ON CREATE SET r.created_at = $created_at
                ON MATCH SET r.updated_at = $created_at
                RETURN r
                """
                output_result = session.run(
                    create_output_query,  # type: ignore[arg-type]
                    {
                        "dataflow_id": dataflow_id,
                        "target_id": target_id,
                        "created_at": get_formatted_time(),
                    },
                ).single()

                if output_result:
                    logger.info(
                        "成功创建OUTPUT关系: %s -> %s",
                        dataflow_name,
                        target_table,
                    )
                else:
                    logger.warning(
                        "OUTPUT关系创建失败或已存在: %s -> %s",
                        dataflow_name,
                        target_table,
                    )

                logger.info(
                    "血缘关系创建完成: %s -[INPUT]-> %s -[OUTPUT]-> %s",
                    source_table,
                    dataflow_name,
                    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

    @staticmethod
    def _register_data_product(
        data: Dict[str, Any],
        dataflow_name: str,
        name_en: str,
        dataflow_id: Optional[int] = None,
    ) -> None:
        """
        注册数据产品到数据服务

        当数据流创建成功后，自动将其注册为数据产品，
        以便在数据服务模块中展示和管理。

        从 script_requirement.target_table 中获取 BusinessDomain ID，
        然后查询 Neo4j 获取对应节点的 name_zh 和 name_en 作为数据产品名称。

        Args:
            data: 数据流配置数据
            dataflow_name: 数据流名称（中文）
            name_en: 数据流英文名
            dataflow_id: 数据流ID（Neo4j节点ID）
        """
        try:
            # 从script_requirement中获取target_table（BusinessDomain ID列表）
            script_requirement = data.get("script_requirement")
            description = data.get("describe", "")

            # 解析 script_requirement
            req_json: Optional[Dict[str, Any]] = None
            if script_requirement:
                if isinstance(script_requirement, dict):
                    req_json = script_requirement
                elif isinstance(script_requirement, str):
                    try:
                        parsed = json.loads(script_requirement)
                        if isinstance(parsed, dict):
                            req_json = parsed
                    except (json.JSONDecodeError, TypeError):
                        pass

            # 获取target_table中的BusinessDomain ID列表
            target_bd_ids: List[int] = []
            if req_json:
                target_table_ids = req_json.get("target_table", [])
                if isinstance(target_table_ids, list):
                    target_bd_ids = [
                        int(bid) for bid in target_table_ids if bid is not None
                    ]
                elif target_table_ids is not None:
                    target_bd_ids = [int(target_table_ids)]

                # 如果有rule字段，添加到描述中
                rule = req_json.get("rule", "")
                if rule and not description:
                    description = rule

            # 如果没有target_table ID，则不注册数据产品
            if not target_bd_ids:
                logger.warning(
                    f"数据流 {dataflow_name} 没有指定target_table，跳过数据产品注册"
                )
                return

            # 从Neo4j查询每个BusinessDomain节点的name_zh和name_en
            with connect_graph().session() as session:
                for bd_id in target_bd_ids:
                    try:
                        query = """
                        MATCH (bd:BusinessDomain)
                        WHERE id(bd) = $bd_id
                        RETURN bd.name_zh as name_zh,
                               bd.name_en as name_en,
                               bd.describe as describe
                        """
                        result = session.run(query, bd_id=bd_id).single()

                        if not result:
                            logger.warning(
                                f"未找到ID为 {bd_id} 的BusinessDomain节点，跳过"
                            )
                            continue

                        # 使用BusinessDomain节点的name_zh和name_en
                        product_name = result.get("name_zh") or ""
                        product_name_en = result.get("name_en") or ""

                        # 如果没有name_zh，使用name_en
                        if not product_name:
                            product_name = product_name_en

                        # 如果没有name_en，使用name_zh转换
                        if not product_name_en:
                            product_name_en = product_name.lower().replace(" ", "_")

                        # 目标表名使用BusinessDomain的name_en
                        target_table = product_name_en

                        # 如果BusinessDomain有describe且当前description为空，使用它
                        bd_describe = result.get("describe") or ""
                        if bd_describe and not description:
                            description = bd_describe

                        # 解析目标schema（默认为public）
                        target_schema = "public"

                        # 调用数据产品服务进行注册
                        DataProductService.register_data_product(
                            product_name=product_name,
                            product_name_en=product_name_en,
                            target_table=target_table,
                            target_schema=target_schema,
                            description=description,
                            source_dataflow_id=dataflow_id,
                            source_dataflow_name=dataflow_name,
                            created_by=data.get("created_by", "dataflow"),
                        )

                        logger.info(
                            f"数据产品注册成功: {product_name} -> "
                            f"{target_schema}.{target_table}"
                        )

                    except Exception as bd_error:
                        logger.error(
                            f"处理BusinessDomain {bd_id} 失败: {str(bd_error)}"
                        )
                        # 继续处理下一个

        except Exception as e:
            logger.error(f"注册数据产品失败: {str(e)}")
            raise