Method generate_yaml

src/domain/services/node_code_generator.py:399–446

生成 YAML 定义 参数： node_name: 节点名称 description: 描述 language: 编程语言 parameters: 参数列表 returns: 返回值定义 返回： YAML 生成结果

(
        self,
        node_name: str,
        description: str,
        language: str,
        parameters: list[dict[str, Any]],
        returns: dict[str, Any],
    )

    """

    def generate_yaml(
        self,
        node_name: str,
        description: str,
        language: str,
        parameters: list[dict[str, Any]],
        returns: dict[str, Any],
    ) -> YamlGenerationResult:
        """生成 YAML 定义

        参数：
            node_name: 节点名称
            description: 描述
            language: 编程语言
            parameters: 参数列表
            returns: 返回值定义

        返回：
            YAML 生成结果
        """
        # 规范化节点名称
        sanitized_name = self._sanitize_name(node_name)

        # 构建 YAML 结构
        yaml_dict = {
            "name": sanitized_name,
            "kind": "node",
            "description": description,
            "version": "1.0.0",
            "parameters": self._format_parameters(parameters),
            "returns": returns,
            "executor_type": "code",
            "language": language,
        }

        try:
            dumped = yaml.dump(
                yaml_dict, allow_unicode=True, default_flow_style=False, sort_keys=False
            )
            if dumped is None:
                yaml_content = ""
            elif isinstance(dumped, bytes):
                yaml_content = dumped.decode("utf-8", errors="replace")
            else:
                yaml_content = dumped
            return YamlGenerationResult(yaml_content=yaml_content, is_valid=True)
        except Exception as e:
            return YamlGenerationResult(yaml_content="", is_valid=False, errors=[str(e)])

    def _sanitize_name(self, name: str) -> str:
        """规范化节点名称"""