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

import contextlib
import copy
import json
import logging
import os
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Union

from sqlalchemy import text

from app import db
from app.core.common.identifiers import ensure_governance_uid
from app.core.data_rules.contracts import validate_dataflow_spec
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.meta_data import get_formatted_time, translate_and_parse

logger = logging.getLogger(__name__)

# 项目根目录
PROJECT_ROOT = Path(__file__).parent.parent.parent.parent


class DataFlowService:
    """数据流服务类，处理数据流相关的业务逻辑"""

    _GOVERNED_REQUIREMENT_KEYS = {
        "dataflow_spec",
        "dataset_edges",
        "migration_metadata",
    }
    _MIGRATION_METADATA_KEYS = {
        "status",
        "legacy_fields_present",
        "preserved_for_read_only",
        "governed_semantics",
    }
    _GOVERNED_PAYLOAD_FIELDS = {
        "draft_reservation",
        "dataflow_spec",
        "dataset_edges",
        "migration_metadata",
    }
    _GOVERNED_FORBIDDEN_FIELDS = {
        "code",
        "generated_code",
        "legacy_script_path",
        "n8n_workflow_id",
        "task_list",
        "workflow",
    }
    _GOVERNED_NODE_KEYS = {
        "uid",
        "name_zh",
        "name_en",
        "category",
        "organization",
        "leader",
        "frequency",
        "describe",
        "status",
        "update_mode",
        "script_type",
        "script_requirement",
        "script_path",
    }

    @staticmethod
    def _decode_script_requirement(value: Any) -> Any:
        if not isinstance(value, str):
            return copy.deepcopy(value)
        try:
            return json.loads(value)
        except (TypeError, json.JSONDecodeError):
            return value

    @classmethod
    def _signals_governed(cls, data: dict[str, Any], requirement: Any) -> bool:
        return (
            data.get("script_type") == "governed"
            or bool(set(data) & cls._GOVERNED_PAYLOAD_FIELDS)
            or (
                isinstance(requirement, dict)
                and bool(set(requirement) & cls._GOVERNED_REQUIREMENT_KEYS)
            )
        )

    @classmethod
    def _reject_governed_execution_fields(cls, data: dict[str, Any]) -> None:
        if set(data) & cls._GOVERNED_FORBIDDEN_FIELDS:
            raise ValueError(
                "governed DataFlow cannot contain legacy execution fields"
            )
        if data.get("script_type") not in {None, "governed"}:
            raise ValueError("governed DataFlow script_type must be governed")
        if data.get("script_path") not in {None, ""}:
            raise ValueError("governed DataFlow cannot define script_path")

    @classmethod
    def validate_governed_requirement(
        cls, value: Any, *, repository
    ) -> Dict[str, Any]:
        """Validate the closed, published-asset DataFlow save envelope."""
        requirement = cls._decode_script_requirement(value)
        if not isinstance(requirement, dict):
            raise ValueError("governed script_requirement must be an object")
        if set(requirement) != cls._GOVERNED_REQUIREMENT_KEYS:
            raise ValueError(
                "governed script_requirement contains unsupported fields"
            )

        flow = validate_dataflow_spec(requirement.get("dataflow_spec"))
        edges = requirement.get("dataset_edges")
        if not isinstance(edges, dict) or set(edges) != {
            "source_table",
            "target_table",
        }:
            raise ValueError("dataset edges must be a closed object")
        if (
            edges.get("source_table") != flow["input_schema_refs"]
            or edges.get("target_table") != flow["output_schema_ref"]
        ):
            raise ValueError("dataset edges must match the DataFlow schema refs")

        metadata = requirement.get("migration_metadata")
        if (
            not isinstance(metadata, dict)
            or set(metadata) != cls._MIGRATION_METADATA_KEYS
        ):
            raise ValueError("migration metadata must be a closed object")
        if metadata.get("status") not in {"migrated", "unmigrated"}:
            raise ValueError("unsupported migration status")
        if type(metadata.get("legacy_fields_present")) is not bool:
            raise ValueError("legacy_fields_present must be a boolean")
        if metadata.get("preserved_for_read_only") is not True:
            raise ValueError("legacy fields must be preserved for read-only use")
        if metadata.get("governed_semantics") != "dataflow_spec":
            raise ValueError("unsupported governed semantics")

        repository.load_published_assets(flow)
        return {
            "dataflow_spec": flow,
            "dataset_edges": {
                "source_table": list(flow["input_schema_refs"]),
                "target_table": flow["output_schema_ref"],
            },
            "migration_metadata": copy.deepcopy(metadata),
        }

    @staticmethod
    def _merge_governed_dataflow(node_data: dict[str, Any]) -> tuple[int, dict]:
        """Idempotently create by stable UID and reject immutable conflicts."""
        node_data = copy.deepcopy(node_data)
        node_data.setdefault("created_at", get_formatted_time())
        node_data.setdefault("updated_at", node_data["created_at"])
        immutable = {
            key: node_data[key]
            for key in ("uid", "name_zh", "script_type", "script_requirement")
        }
        driver = connect_graph()
        try:
            with driver.session() as session:
                session.run(
                    "CREATE CONSTRAINT data_flow_uid IF NOT EXISTS "
                    "FOR (n:DataFlow) REQUIRE n.uid IS UNIQUE"
                )
                conflict = session.run(
                    "MATCH (n:DataFlow {name_zh: $name_zh}) "
                    "WHERE n.uid IS NULL OR n.uid <> $uid "
                    "RETURN n.uid AS uid LIMIT 1",
                    {"name_zh": node_data["name_zh"], "uid": node_data["uid"]},
                ).single()
                if conflict is not None:
                    raise ValueError("dataflow_uid_conflict")
                record = session.run(
                    "MERGE (n:DataFlow {uid: $uid}) "
                    "ON CREATE SET n = $properties "
                    "RETURN n, id(n) AS node_id",
                    {"uid": node_data["uid"], "properties": node_data},
                ).single()
                if record is None:
                    raise RuntimeError(
                        "governed DataFlow MERGE returned no node"
                    )
                persisted = dict(record["n"])
                for key, expected in immutable.items():
                    if persisted.get(key) != expected:
                        raise ValueError("dataflow_uid_conflict")
                node_id = record["node_id"]
                if isinstance(node_id, bool) or not isinstance(node_id, int):
                    raise RuntimeError("governed DataFlow node id is invalid")
                result = dict(persisted)
                result["id"] = node_id
                return node_id, result
        finally:
            driver.close()

    @classmethod
    def validate_governed_create_intent(
        cls, value: Any, *, repository
    ) -> dict[str, Any]:
        if not isinstance(value, dict) or set(value) != {
            "dataflow_uid",
            "node",
            "tags",
        }:
            raise ValueError("dataflow create intent is invalid")
        node = value.get("node")
        tags = value.get("tags")
        if not isinstance(node, dict) or set(node) != cls._GOVERNED_NODE_KEYS:
            raise ValueError("dataflow create intent node is invalid")
        if not isinstance(tags, list) or len(tags) > 100:
            raise ValueError("dataflow create intent tags are invalid")
        flow_uid = ensure_governance_uid(
            {"uid": str(value.get("dataflow_uid"))}
        )
        if node.get("uid") != flow_uid:
            raise ValueError("dataflow create intent UID mismatch")
        if node.get("script_type") != "governed" or node.get(
            "script_path"
        ) not in {"", None}:
            raise ValueError("dataflow create intent execution mode is invalid")
        requirement = cls.validate_governed_requirement(
            node.get("script_requirement"), repository=repository
        )
        if requirement["dataflow_spec"]["dataflow_uid"] != flow_uid:
            raise ValueError("dataflow create intent requirement UID mismatch")
        normalized_node = copy.deepcopy(node)
        normalized_node["script_requirement"] = json.dumps(
            requirement,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        )
        for key in cls._GOVERNED_NODE_KEYS - {
            "script_path",
            "script_requirement",
        }:
            if not isinstance(normalized_node.get(key), str):
                raise ValueError("dataflow create intent node field is invalid")
        return {
            "dataflow_uid": flow_uid,
            "node": normalized_node,
            "tags": copy.deepcopy(tags),
        }

    @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], *, repository=None, actor_uid=None
    ) -> 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"]

            script_requirement = data.get("script_requirement")
            decoded_requirement = DataFlowService._decode_script_requirement(
                script_requirement
            )
            governed = DataFlowService._signals_governed(
                data, decoded_requirement
            )
            name_en = data.get("name_en")
            if governed and (
                not isinstance(name_en, str) or not name_en.strip()
            ):
                flow_value = (
                    decoded_requirement.get("dataflow_spec")
                    if isinstance(decoded_requirement, dict)
                    else None
                )
                flow_uid = (
                    flow_value.get("dataflow_uid")
                    if isinstance(flow_value, dict)
                    else None
                )
                name_en = f"dataflow_{str(flow_uid).replace('-', '')}"
            elif not isinstance(name_en, str) or not name_en.strip():
                # 使用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(" ", "_")
            name_en = name_en.strip()

            saga_claim = None
            receipt = None
            if governed:
                DataFlowService._reject_governed_execution_fields(data)
                if repository is None:
                    from app.core.data_rules.repository import DataRuleRepository

                    repository = DataRuleRepository(db.session)
                decoded_requirement = (
                    DataFlowService.validate_governed_requirement(
                        decoded_requirement, repository=repository
                    )
                )
                if actor_uid is None:
                    raise ValueError(
                        "governed DataFlow requires an authenticated actor"
                    )
                receipt = data.get("draft_reservation")
                script_requirement = decoded_requirement

            # 处理 script_requirement，将其转换为 JSON 字符串
            if script_requirement is not None:
                # 如果是字典或列表，转换为 JSON 字符串
                if isinstance(script_requirement, (dict, list)):
                    script_requirement_str = json.dumps(
                        script_requirement,
                        ensure_ascii=False,
                        sort_keys=governed,
                        separators=(",", ":") if governed else None,
                    )
                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,
                "script_path": "",  # 脚本路径，任务完成后更新
            }
            if not governed:
                node_data["created_at"] = get_formatted_time()
                node_data["updated_at"] = get_formatted_time()
            if governed:
                node_data["uid"] = decoded_requirement["dataflow_spec"][
                    "dataflow_uid"
                ]
                node_data["script_type"] = "governed"
            ensure_governance_uid(node_data)

            # Governed nodes use the reservation UID as the only create key.
            if governed:
                tag_list = copy.deepcopy(data.get("tag", []))
                if not isinstance(tag_list, list):
                    tag_list = [tag_list] if tag_list else []
                create_intent = {
                    "dataflow_uid": node_data["uid"],
                    "node": copy.deepcopy(node_data),
                    "tags": tag_list,
                }
                request_payload = copy.deepcopy(data)
                request_payload.pop("draft_reservation", None)
                request_payload["script_requirement"] = decoded_requirement
                create_request = {
                    "payload": request_payload,
                    "intent": create_intent,
                }
                saga_claim = repository.begin_dataflow_create(
                    receipt,
                    actor_uid=actor_uid,
                    create_request=create_request,
                    create_intent=create_intent,
                )
                if saga_claim["status"] == "completed":
                    return copy.deepcopy(saga_claim["result"])
                reserved_uid = saga_claim["dataflow_uid"]
                if reserved_uid != node_data["uid"]:
                    raise ValueError(
                        "draft reservation does not match dataflow_uid"
                    )
                stored_intent = DataFlowService.validate_governed_create_intent(
                    saga_claim["create_intent"], repository=repository
                )
                node_data = stored_intent["node"]
                tag_list = stored_intent["tags"]
                repository.commit_dataflow_create_claim()
                try:
                    dataflow_id, result = (
                        DataFlowService._merge_governed_dataflow(node_data)
                    )
                except Exception as graph_error:
                    try:
                        repository.commit_dataflow_create_failure(
                            reservation_id=receipt["reservation_id"],
                            lease_token=saga_claim["lease_token"],
                            error_code="neo4j_create_failed",
                        )
                    except Exception:
                        logger.exception(
                            "failed to persist governed DataFlow saga failure"
                        )
                    raise graph_error
            else:
                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 = tag_list if governed else data.get("tag", [])
            if tag_list:
                try:
                    DataFlowService._handle_tag_relationships(
                        dataflow_id, tag_list, strict=governed
                    )
                except Exception as e:
                    if governed:
                        try:
                            repository.commit_dataflow_create_failure(
                                reservation_id=receipt["reservation_id"],
                                lease_token=saga_claim["lease_token"],
                                error_code="neo4j_tag_merge_failed",
                            )
                        except Exception:
                            logger.exception(
                                "failed to persist governed DataFlow tag failure"
                            )
                        raise
                    logger.warning(f"处理标签关系时出错: {str(e)}")

            # Governed definitions are control-plane assets. Data Factory owns
            # all later task, code, workflow and product deployment side effects.
            if not governed:
                try:
                    DataFlowService._save_to_pg_database(
                        data, dataflow_name, name_en
                    )
                    logger.info(
                        f"数据流信息已写入PG数据库: {dataflow_name}"
                    )

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

            if governed:
                result = repository.complete_dataflow_create(
                    reservation_id=receipt["reservation_id"],
                    dataflow_uid=node_data["uid"],
                    lease_token=saga_claim["lease_token"],
                    request_digest=saga_claim["request_digest"],
                    dataflow_node_id=dataflow_id,
                    result=result,
                )
            else:
                # 查询创建的节点获取完整信息
                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
                            ),
                            "uid": node_data["uid"],
                            "name_zh": dataflow_name,
                            "name_en": name_en,
                            "created_at": get_formatted_time(),
                        }

            if not governed:
                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数据库的task_list表

        Args:
            data: 包含脚本信息的数据
            script_name: 脚本名称
            name_en: 英文名称
        """
        try:
            # 提取脚本相关信息
            # 处理 script_requirement，确保保存为 JSON 字符串
            script_requirement_raw = data.get("script_requirement")

            if script_requirement_raw is not None:
                if isinstance(script_requirement_raw, (dict, list)):
                    script_requirement = json.dumps(
                        script_requirement_raw, ensure_ascii=False
                    )
                else:
                    script_requirement = str(script_requirement_raw)
            else:
                script_requirement = ""

            # 验证必需字段
            if not script_name:
                raise ValueError("script_name不能为空")

            current_time = datetime.now()

            # 保存到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 []
                            )

                        # 从data参数中提取update_mode
                        update_mode = data.get("update_mode", "append")

                        # 生成Business Domain DDLs和数据源信息
                        source_tables_info = []
                        target_tables_info = []

                        if source_table_ids or target_table_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_tables_info.append(ddl_info)

                                    # 处理target tables（目标表缺省要有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_tables_info.append(ddl_info)

                            except Exception as neo_e:
                                logger.error(
                                    f"获取BusinessDomain DDL失败: {str(neo_e)}"
                                )

                        # 构建Markdown格式的任务描述
                        task_desc_parts = [f"# Task: {script_name}\n"]

                        # 添加源表信息（DDL和数据源）
                        if source_tables_info:
                            task_desc_parts.append("## Source Tables")
                            for info in source_tables_info:
                                task_desc_parts.append(f"### {info['table_name']}")
                                if info.get("data_source"):
                                    ds = info["data_source"]
                                    task_desc_parts.append("**Data Source**")
                                    task_desc_parts.append(
                                        f"- **Type**: {ds.get('type', 'N/A')}"
                                    )
                                    task_desc_parts.append(
                                        f"- **Host**: {ds.get('host', 'N/A')}"
                                    )
                                    task_desc_parts.append(
                                        f"- **Port**: {ds.get('port', 'N/A')}"
                                    )
                                    task_desc_parts.append(
                                        f"- **Database**: {ds.get('database', 'N/A')}"
                                    )
                                    task_desc_parts.append(
                                        f"- **Schema**: {ds.get('schema', 'N/A')}\n"
                                    )
                                task_desc_parts.append("**DDL**")
                                task_desc_parts.append(f"```sql\n{info['ddl']}\n```\n")

                        # 添加目标表信息（DDL和数据源）
                        if target_tables_info:
                            task_desc_parts.append("## Target Tables")
                            for info in target_tables_info:
                                task_desc_parts.append(f"### {info['table_name']}")
                                if info.get("data_source"):
                                    ds = info["data_source"]
                                    task_desc_parts.append("**Data Source**")
                                    task_desc_parts.append(
                                        f"- **Type**: {ds.get('type', 'N/A')}"
                                    )
                                    task_desc_parts.append(
                                        f"- **Host**: {ds.get('host', 'N/A')}"
                                    )
                                    task_desc_parts.append(
                                        f"- **Port**: {ds.get('port', 'N/A')}"
                                    )
                                    task_desc_parts.append(
                                        f"- **Database**: {ds.get('database', 'N/A')}"
                                    )
                                    task_desc_parts.append(
                                        f"- **Schema**: {ds.get('schema', 'N/A')}\n"
                                    )
                                task_desc_parts.append("**DDL**")
                                task_desc_parts.append(f"```sql\n{info['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")
                        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_description_md = "\n".join(task_desc_parts)

                except Exception as parse_e:
                    logger.warning(
                        f"解析任务描述详情失败，使用原始描述: {str(parse_e)}"
                    )
                    task_description_md = script_requirement

                # 设置 code_path（不包含文件名）
                # code_name 需要在获取 task_id 后生成
                code_path = "datafactory/scripts"

                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)\n"
                    "RETURNING task_id"
                )

                task_params = {
                    "task_name": script_name,
                    "task_description": task_description_md,
                    "status": "pending",
                    "code_name": "",  # 暂时为空，等获取 task_id 后更新
                    "code_path": code_path,
                    "create_by": "cursor",
                    "create_time": current_time,
                    "update_time": current_time,
                }

                result = db.session.execute(task_insert_sql, task_params)
                row = result.fetchone()
                task_id = row[0] if row else None

                # 根据 task_id 生成脚本文件名
                # 格式: task_{task_id}_{task_name}.py（与 auto_execute_tasks 生成的一致）
                code_name = f"task_{task_id}_{script_name}.py"

                # 更新 code_name 字段
                if task_id:
                    update_sql = text(
                        "UPDATE public.task_list SET code_name = :code_name "
                        "WHERE task_id = :task_id"
                    )
                    db.session.execute(
                        update_sql, {"code_name": code_name, "task_id": task_id}
                    )

                db.session.commit()

                logger.info(
                    f"成功将任务信息写入task_list表: "
                    f"task_id={task_id}, task_name={script_name}, code_name={code_name}"
                )

                # 自动生成 n8n 工作流 JSON 文件
                try:
                    DataFlowService._generate_n8n_workflow(
                        script_name=script_name,
                        code_name=code_name,
                        code_path=code_path,
                        update_mode=update_mode,
                        task_id=task_id,
                    )
                except Exception as wf_error:
                    logger.warning(f"生成n8n工作流文件失败: {str(wf_error)}")
                    # 不影响主流程

            except Exception as task_error:
                db.session.rollback()
                logger.error(f"写入task_list表失败: {str(task_error)}")
                raise task_error

        except Exception as e:
            db.session.rollback()
            logger.error(f"保存到PG数据库失败: {str(e)}")
            raise e

    @staticmethod
    def _generate_n8n_workflow(
        script_name: str,
        code_name: str,
        code_path: str,
        update_mode: str = "append",
        task_id: Optional[int] = None,
    ) -> Optional[str]:
        """
        自动生成 n8n 工作流 JSON 文件

        Args:
            script_name: 脚本/任务名称
            code_name: 代码文件名（如 task_42_DF_DO202601210001.py）
            code_path: 代码路径（如 datafactory/scripts）
            update_mode: 更新模式
            task_id: 关联的任务 ID

        Returns:
            生成的工作流文件路径，失败返回 None
        """
        try:
            # 确保工作流目录存在
            workflows_dir = PROJECT_ROOT / "datafactory" / "workflows"
            workflows_dir.mkdir(parents=True, exist_ok=True)

            # 生成工作流文件名（使用任务ID以便于关联）
            if task_id:
                workflow_filename = f"task_{task_id}_{script_name}_workflow.json"
            else:
                workflow_filename = f"{script_name}_workflow.json"
            workflow_path = workflows_dir / workflow_filename

            # 生成唯一ID
            def gen_id():
                return str(uuid.uuid4())

            # 构建完整的 SSH 命令，包含激活 venv
            # 注意：由于 n8n 服务器与应用服务器分离，必须使用 SSH 节点
            # code_name 已经包含 .py 后缀（如 task_42_DF_DO202601210001.py）
            ssh_command = (
                f"cd /opt/dataops-platform && source venv/bin/activate && "
                f"python {code_path}/{code_name}"
            )

            workflow_json = {
                "name": f"{script_name}_工作流",
                "nodes": [
                    {
                        "parameters": {
                            "rule": {
                                "interval": [
                                    {
                                        "field": "days",
                                        "daysInterval": 1,
                                        "triggerAtHour": 1,
                                        "triggerAtMinute": 0,
                                    }
                                ]
                            }
                        },
                        "id": gen_id(),
                        "name": "Schedule Trigger",
                        "type": "n8n-nodes-base.scheduleTrigger",
                        "typeVersion": 1.2,
                        "position": [250, 300],
                    },
                    {
                        "parameters": {
                            "resource": "command",
                            "operation": "execute",
                            "command": ssh_command,
                            "cwd": "/opt/dataops-platform",
                        },
                        "id": gen_id(),
                        "name": "Execute Script",
                        "type": "n8n-nodes-base.ssh",
                        "typeVersion": 1,
                        "position": [450, 300],
                        "credentials": {
                            "sshPassword": {
                                "id": "pYTwwuyC15caQe6y",
                                "name": "SSH Password account",
                            }
                        },
                    },
                    {
                        "parameters": {
                            "conditions": {
                                "options": {
                                    "caseSensitive": True,
                                    "leftValue": "",
                                    "typeValidation": "strict",
                                },
                                "conditions": [
                                    {
                                        "id": "condition-success",
                                        "leftValue": "={{ $json.code }}",
                                        "rightValue": 0,
                                        "operator": {
                                            "type": "number",
                                            "operation": "equals",
                                        },
                                    }
                                ],
                                "combinator": "and",
                            }
                        },
                        "id": gen_id(),
                        "name": "Check Result",
                        "type": "n8n-nodes-base.if",
                        "typeVersion": 2,
                        "position": [650, 300],
                    },
                    {
                        "parameters": {
                            "assignments": {
                                "assignments": [
                                    {
                                        "id": "result-success",
                                        "name": "status",
                                        "value": "success",
                                        "type": "string",
                                    },
                                    {
                                        "id": "result-message",
                                        "name": "message",
                                        "value": f"{script_name} 执行成功",
                                        "type": "string",
                                    },
                                    {
                                        "id": "result-output",
                                        "name": "output",
                                        "value": "={{ $json.stdout }}",
                                        "type": "string",
                                    },
                                    {
                                        "id": "result-time",
                                        "name": "executionTime",
                                        "value": "={{ $now.toISO() }}",
                                        "type": "string",
                                    },
                                ]
                            }
                        },
                        "id": gen_id(),
                        "name": "Success Response",
                        "type": "n8n-nodes-base.set",
                        "typeVersion": 3.4,
                        "position": [850, 200],
                    },
                    {
                        "parameters": {
                            "assignments": {
                                "assignments": [
                                    {
                                        "id": "error-status",
                                        "name": "status",
                                        "value": "error",
                                        "type": "string",
                                    },
                                    {
                                        "id": "error-message",
                                        "name": "message",
                                        "value": f"{script_name} 执行失败",
                                        "type": "string",
                                    },
                                    {
                                        "id": "error-output",
                                        "name": "error",
                                        "value": "={{ $json.stderr }}",
                                        "type": "string",
                                    },
                                    {
                                        "id": "error-code",
                                        "name": "exitCode",
                                        "value": "={{ $json.code }}",
                                        "type": "number",
                                    },
                                    {
                                        "id": "error-time",
                                        "name": "executionTime",
                                        "value": "={{ $now.toISO() }}",
                                        "type": "string",
                                    },
                                ]
                            }
                        },
                        "id": gen_id(),
                        "name": "Error Response",
                        "type": "n8n-nodes-base.set",
                        "typeVersion": 3.4,
                        "position": [850, 400],
                    },
                ],
                "connections": {
                    "Schedule Trigger": {
                        "main": [
                            [
                                {
                                    "node": "Execute Script",
                                    "type": "main",
                                    "index": 0,
                                }
                            ]
                        ]
                    },
                    "Execute Script": {
                        "main": [
                            [
                                {
                                    "node": "Check Result",
                                    "type": "main",
                                    "index": 0,
                                }
                            ]
                        ]
                    },
                    "Check Result": {
                        "main": [
                            [
                                {
                                    "node": "Success Response",
                                    "type": "main",
                                    "index": 0,
                                }
                            ],
                            [
                                {
                                    "node": "Error Response",
                                    "type": "main",
                                    "index": 0,
                                }
                            ],
                        ]
                    },
                },
                "active": False,
                "settings": {"executionOrder": "v1"},
                "versionId": "1",
                "meta": {
                    "templateCredsSetupCompleted": False,
                    "instanceId": "dataops-platform",
                },
                "tags": [
                    {
                        "createdAt": datetime.now().isoformat() + "Z",
                        "updatedAt": datetime.now().isoformat() + "Z",
                        "id": "1",
                        "name": "数据流程",
                    }
                ],
            }

            # 写入文件
            with open(workflow_path, "w", encoding="utf-8") as f:
                json.dump(workflow_json, f, ensure_ascii=False, indent=2)

            logger.info(f"成功生成n8n工作流文件: {workflow_path}")
            return str(workflow_path)

        except Exception as e:
            logger.error(f"生成n8n工作流失败: {str(e)}")
            return None

    @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, *, strict=False):
        """
        处理多标签关系

        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, strict=strict
                )
            elif strict:
                raise ValueError("dataflow create intent tag is invalid")

    @staticmethod
    def _handle_single_tag_relationship(dataflow_id, tag_id, *, strict=False):
        """处理单个标签关系"""
        driver = None
        try:
            driver = connect_graph()
            with driver.session() as session:
                merged = session.run(
                    "MATCH (a:DataFlow), (b:DataLabel) "
                    "WHERE id(a) = $dataflow_id AND id(b) = $tag_id "
                    "MERGE (a)-[:LABEL]->(b) "
                    "RETURN count(*) AS merged",
                    dataflow_id=dataflow_id,
                    tag_id=tag_id,
                ).single()
                if merged is None or merged["merged"] != 1:
                    raise ValueError("dataflow tag target does not exist")
                logger.info(f"合并标签关系: {dataflow_id} -> {tag_id}")
        except Exception as e:
            if strict:
                raise
            logger.warning(f"创建标签关系失败 {tag_id}: {str(e)}")
        finally:
            if driver is not None:
                driver.close()

    @staticmethod
    def update_dataflow_script_path(
        dataflow_name: str,
        script_path: str,
    ) -> bool:
        """
        更新 DataFlow 节点的脚本路径

        当任务完成后，将创建的 Python 脚本路径更新到 DataFlow 节点

        Args:
            dataflow_name: 数据流名称（中文名）
            script_path: Python 脚本的完整路径

        Returns:
            是否更新成功
        """
        try:
            query = """
            MATCH (n:DataFlow {name_zh: $name_zh})
            SET n.script_path = $script_path, n.updated_at = $updated_at
            RETURN n
            """
            with connect_graph().session() as session:
                result = session.run(
                    query,
                    name_zh=dataflow_name,
                    script_path=script_path,
                    updated_at=get_formatted_time(),
                ).single()

                if result:
                    logger.info(
                        f"已更新 DataFlow 脚本路径: {dataflow_name} -> {script_path}"
                    )
                    return True
                else:
                    logger.warning(f"未找到 DataFlow 节点: {dataflow_name}")
                    return False

        except Exception as e:
            logger.error(f"更新 DataFlow 脚本路径失败: {str(e)}")
            return False

    @staticmethod
    def get_script_content(dataflow_id: int) -> Dict[str, Any]:
        """
        根据 DataFlow ID 获取关联的脚本内容

        Args:
            dataflow_id: 数据流ID

        Returns:
            包含脚本内容和元信息的字典:
            - script_path: 脚本路径
            - script_content: 脚本内容
            - script_type: 脚本类型（如 python）
            - dataflow_name: 数据流名称

        Raises:
            ValueError: 当 DataFlow 不存在或脚本路径为空时
            FileNotFoundError: 当脚本文件不存在时
        """
        from pathlib import Path

        try:
            # 从 Neo4j 获取 DataFlow 节点
            query = """
            MATCH (n:DataFlow)
            WHERE id(n) = $dataflow_id
            RETURN n, id(n) as node_id
            """

            with connect_graph().session() as session:
                result = session.run(query, dataflow_id=dataflow_id).single()

                if not result:
                    raise ValueError(f"未找到 ID 为 {dataflow_id} 的 DataFlow 节点")

                node = result["n"]
                node_props = dict(node)

                # 获取脚本路径
                script_path = node_props.get("script_path", "")
                if not script_path:
                    raise ValueError(
                        f"DataFlow (ID: {dataflow_id}) 的 script_path 属性为空"
                    )

                # 确定脚本文件的完整路径
                # script_path 可能是相对路径或绝对路径
                script_file = Path(script_path)

                # 如果是相对路径，相对于项目根目录
                if not script_file.is_absolute():
                    # 获取项目根目录（假设 app 目录的父目录是项目根）
                    project_root = Path(__file__).parent.parent.parent.parent
                    script_file = project_root / script_path

                # 检查文件是否存在
                if not script_file.exists():
                    raise FileNotFoundError(f"脚本文件不存在: {script_file}")

                # 读取脚本内容
                with script_file.open("r", encoding="utf-8") as f:
                    script_content = f.read()

                # 确定脚本类型
                suffix = script_file.suffix.lower()
                script_type_map = {
                    ".py": "python",
                    ".js": "javascript",
                    ".ts": "typescript",
                    ".sql": "sql",
                    ".sh": "shell",
                }
                script_type = script_type_map.get(suffix, "text")

                logger.info(
                    f"成功读取脚本内容: DataFlow ID={dataflow_id}, "
                    f"路径={script_path}, 类型={script_type}"
                )

                return {
                    "script_path": script_path,
                    "script_content": script_content,
                    "script_type": script_type,
                    "dataflow_id": dataflow_id,
                    "dataflow_name": node_props.get("name_zh", ""),
                    "dataflow_name_en": node_props.get("name_en", ""),
                }

        except (ValueError, FileNotFoundError):
            raise
        except Exception as e:
            logger.error(f"获取脚本内容失败: {str(e)}")
            raise

    @staticmethod
    def update_dataflow(
        dataflow_id: int,
        data: Dict[str, Any],
        *,
        repository=None,
    ) -> Optional[Dict[str, Any]]:
        """
        更新数据流

        Args:
            dataflow_id: 数据流ID
            data: 更新的数据

        Returns:
            更新后的数据流信息，如果不存在则返回None
        """
        try:
            data = copy.deepcopy(data)
            requirement = DataFlowService._decode_script_requirement(
                data.get("script_requirement")
            )

            # 查找节点
            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
                existing = dict(result[0]["n"])
                existing_requirement = (
                    DataFlowService._decode_script_requirement(
                        existing.get("script_requirement")
                    )
                )
                governed = DataFlowService._signals_governed(
                    data, requirement
                ) or DataFlowService._signals_governed(
                    existing, existing_requirement
                )
                if governed:
                    if "draft_reservation" in data:
                        raise ValueError(
                            "existing DataFlow cannot consume a new draft"
                        )
                    DataFlowService._reject_governed_execution_fields(data)
                    if repository is None:
                        from app.core.data_rules.repository import (
                            DataRuleRepository,
                        )

                        repository = DataRuleRepository(db.session)
                    requirement = (
                        DataFlowService.validate_governed_requirement(
                            requirement, repository=repository
                        )
                    )
                    data["script_requirement"] = requirement
                    data["script_type"] = "governed"
                    data["script_path"] = ""
                    data["uid"] = requirement["dataflow_spec"][
                        "dataflow_uid"
                    ]
                    existing_uid = existing.get("uid")
                    requested_uid = data["uid"]
                    if not existing_uid or str(existing_uid) != requested_uid:
                        raise ValueError(
                            "dataflow_uid cannot replace an existing identity"
                        )

                # 提取 tag 数组（不作为节点属性存储）
                tag_list = data.pop("tag", 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

    # 无 COME_FROM 关系时使用显式环境配置；默认仅指向本机，禁止回退内网地址。
    DEFAULT_DATA_SOURCE = {
        "type": "postgresql",
        "host": os.environ.get("DATAFLOW_DEFAULT_DB_HOST", "127.0.0.1"),
        "port": int(os.environ.get("DATAFLOW_DEFAULT_DB_PORT", "5432")),
        "database": os.environ.get("DATAFLOW_DEFAULT_DB_NAME", "dataops"),
        "schema": os.environ.get("DATAFLOW_SCHEMA", "dags"),
    }

    @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
            data_source始终返回，如果没有COME_FROM关系则使用显式环境配置
        """
        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,
                   ds.schema as ds_schema
            """
            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"]

            # 生成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'")

            # 如果是目标表，添加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)

            # 始终返回数据源信息
            # 如果通过COME_FROM关系找到了数据源，使用该数据源
            # 否则使用默认的生产环境数据库配置
            if result["ds_type"]:
                data_source = {
                    "type": result["ds_type"],
                    "host": result["ds_host"],
                    "port": result["ds_port"],
                    "database": result["ds_database"],
                    "schema": result["ds_schema"],
                }

                # 端口验证：确保数据库类型使用正确的端口
                # PostgreSQL 默认端口 5432，MySQL 默认端口 3306
                # 5678 是 n8n 服务端口，不是数据库端口
                ds_type_lower = (result["ds_type"] or "").lower()
                current_port = data_source.get("port")

                # 定义数据库类型与默认端口的映射
                db_default_ports = {
                    "postgresql": 5432,
                    "postgres": 5432,
                    "mysql": 3306,
                    "mariadb": 3306,
                    "sqlserver": 1433,
                    "mssql": 1433,
                    "oracle": 1521,
                }

                # 常见的非数据库端口（需要修正）
                invalid_db_ports = {5678, 8080, 80, 443, 8000, 3000}

                if ds_type_lower in db_default_ports:
                    expected_port = db_default_ports[ds_type_lower]
                    if current_port in invalid_db_ports:
                        logger.warning(
                            f"检测到数据源端口配置异常: type={ds_type_lower}, "
                            f"port={current_port}（疑似非数据库端口），"
                            f"已自动修正为默认端口 {expected_port}"
                        )
                        data_source["port"] = expected_port
                    elif current_port is None:
                        logger.info(f"数据源端口为空，使用默认端口: {expected_port}")
                        data_source["port"] = expected_port

                logger.info(f"通过COME_FROM关系获取到数据源信息: {data_source}")
            else:
                # 使用默认生产环境数据源配置
                data_source = DataFlowService.DEFAULT_DATA_SOURCE.copy()
                logger.info(
                    f"未找到COME_FROM关系，使用默认生产环境数据源: {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，以及关联数据源的schema
            with connect_graph().session() as session:
                for bd_id in target_bd_ids:
                    try:
                        # 查询BusinessDomain节点信息及其关联的数据源schema
                        query = """
                        MATCH (bd:BusinessDomain)
                        WHERE id(bd) = $bd_id
                        OPTIONAL MATCH (bd)-[:COME_FROM]->(ds:DataSource)
                        RETURN bd.name_zh as name_zh,
                               bd.name_en as name_en,
                               bd.describe as describe,
                               ds.schema as ds_schema
                        """
                        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 = result.get("ds_schema") or "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