压缩JSON的格式化与结构统
  13608564384 19天前 87 0

工作流JSON实战技巧

一、压缩JSON的格式化与结构统一

从PandaAI导出的JSON文件有时是压缩格式（单行），在IDE中编辑极其困难。同时，某些导出格式可能包含冗余的 nodes 和 litegraph.nodes 双重结构，需要统一处理。

问题现象：

  • JSON文件被压缩成单行，无法阅读和编辑
  • 存在 litegraph.nodes 和顶层 nodes 两套节点数据，结构不一致
  • 需要保留 litegraph 结构，清理冗余数据

格式化脚本：

import json
import sys

def format_workflow_json(input_file, output_file=None):
    """
    格式化工作流JSON文件，统一结构
    
    功能：
    1. 格式化压缩的JSON为可读格式
    2. 统一节点结构（只保留litegraph.nodes）
    3. 清理冗余的顶层nodes和links
    4. 验证JSON有效性
    """
    # 读取文件
    try:
        with open(input_file, 'r', encoding='utf-8') as f:
            content = f.read()
            workflow = json.loads(content)
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e}")
        return False
    
    # 统一结构：确保只使用litegraph结构
    if 'litegraph' not in workflow:
        print("错误：未找到litegraph结构")
        return False
    
    litegraph = workflow['litegraph']
    
    # 如果存在顶层nodes/links，检查是否需要合并
    if 'nodes' in workflow and workflow['nodes']:
        print("警告：发现顶层nodes结构，将使用litegraph.nodes")
        # 可以选择合并，但通常litegraph.nodes是标准格式
        # 这里选择保留litegraph.nodes，忽略顶层nodes
    
    # 清理冗余字段（可选）
    # 移除顶层nodes和links（如果存在且与litegraph重复）
    if 'nodes' in workflow:
        del workflow['nodes']
    if 'links' in workflow:
        del workflow['links']
    
    # 确保litegraph结构完整
    if 'nodes' not in litegraph:
        litegraph['nodes'] = []
    if 'links' not in litegraph:
        litegraph['links'] = []
    if 'version' not in litegraph:
        litegraph['version'] = 0.4
    
    # 输出文件
    if output_file is None:
        output_file = input_file.replace('.json', '_格式化.json')
    
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(workflow, f, indent=2, ensure_ascii=False)
    
    print(f"格式化完成: {output_file}")
    print(f"节点数量: {len(litegraph['nodes'])}")
    print(f"连接数量: {len(litegraph['links'])}")
    
    return True

# 使用示例
if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: python format_json.py <输入文件> [输出文件]")
        sys.exit(1)
    
    input_file = sys.argv[1]
    output_file = sys.argv[2] if len(sys.argv) > 2 else None
    
    format_workflow_json(input_file, output_file)

IDE快速格式化（VS Code）：

  1. 安装扩展：JSON Tools 或使用内置格式化（Shift+Alt+F）
  2. 对于压缩JSON，先使用 JSON Tools: Minify 再 Format Document
  3. 使用命令面板：Format Document With... → 选择 JSON

命令行一键格式化：

# 使用Python
python -c "import json, sys; json.dump(json.load(open(sys.argv[1])), open(sys.argv[2], 'w'), indent=2, ensure_ascii=False)" input.json output.json

# 使用jq（需要安装）
jq . input.json > output.json

# 使用Node.js
node -e "const fs=require('fs'); const data=JSON.parse(fs.readFileSync(process.argv[2])); fs.writeFileSync(process.argv[3], JSON.stringify(data, null, 2))" input.json output.json

二、JSON结构差异的兼容处理

不同版本或不同导出方式产生的JSON结构可能不同，需要编写兼容性处理脚本，确保工作流能在各种环境下正常运行。

常见结构差异：

  1. 节点ID类型：有些是数字，有些是字符串
  2. 连接格式：数组格式 [id, src, slot, tgt, slot, type] vs 对象格式 {id, source, target, ...}
  3. 字段缺失：某些节点缺少 widgets_values、flags 等字段
  4. 嵌套结构：litegraph.nodes vs 顶层 nodes

兼容性处理脚本：

import json
from typing import Any, Dict, List

def normalize_workflow(workflow: Dict[str, Any]) -> Dict[str, Any]:
    """
    标准化工作流JSON结构，处理各种格式差异
    
    返回:
        标准化后的工作流JSON
    """
    # 1. 统一节点结构
    if 'litegraph' not in workflow:
        # 如果没有litegraph，尝试从顶层nodes构建
        if 'nodes' in workflow:
            workflow['litegraph'] = {
                'id': workflow.get('id', '00000000-0000-0000-0000-000000000000'),
                'version': workflow.get('version', 0.4),
                'nodes': workflow['nodes'],
                'links': workflow.get('links', []),
                'groups': workflow.get('groups', []),
                'config': workflow.get('config', {}),
                'extra': workflow.get('extra', {})
            }
        else:
            raise ValueError("无法找到节点数据")
    
    litegraph = workflow['litegraph']
    nodes = litegraph.get('nodes', [])
    links = litegraph.get('links', [])
    
    # 2. 标准化节点ID（确保都是整数）
    node_id_map = {}
    for i, node in enumerate(nodes, start=1):
        old_id = node.get('id')
        if isinstance(old_id, str):
            # 字符串ID转为整数
            try:
                new_id = int(old_id)
            except ValueError:
                new_id = i
        else:
            new_id = old_id if old_id else i
        
        node_id_map[old_id] = new_id
        node['id'] = new_id
        node['order'] = node.get('order', new_id)
    
    # 3. 标准化连接格式
    normalized_links = []
    for link in links:
        if isinstance(link, list):
            # 数组格式：[link_id, src_id, src_slot, tgt_id, tgt_slot, type]
            link_id, src_id, src_slot, tgt_id, tgt_slot, link_type = link
            # 更新节点ID引用
            src_id = node_id_map.get(src_id, src_id)
            tgt_id = node_id_map.get(tgt_id, tgt_id)
            normalized_links.append([link_id, src_id, src_slot, tgt_id, tgt_slot, link_type])
        elif isinstance(link, dict):
            # 对象格式：转换为数组格式
            link_id = link.get('id', len(normalized_links) + 1)
            src_id = node_id_map.get(link.get('source'), link.get('source'))
            src_slot = link.get('source_slot', 0)
            tgt_id = node_id_map.get(link.get('target'), link.get('target'))
            tgt_slot = link.get('target_slot', 0)
            link_type = link.get('type', 'dataframe')
            normalized_links.append([link_id, src_id, src_slot, tgt_id, tgt_slot, link_type])
    
    litegraph['links'] = normalized_links
    
    # 4. 补充缺失的必需字段
    for node in nodes:
        # 确保有flags字段
        if 'flags' not in node:
            node['flags'] = {}
        if 'uuid' not in node['flags']:
            import uuid
            node['flags']['uuid'] = str(uuid.uuid4())
        
        # 确保有order字段
        if 'order' not in node:
            node['order'] = node.get('id', 0)
        
        # 确保有mode字段
        if 'mode' not in node:
            node['mode'] = 0
        
        # CodeControl节点确保有widgets_values
        if node.get('type') == 'CodeControl':
            if 'widgets_values' not in node:
                # 从properties.策略代码复制
                code = node.get('properties', {}).get('策略代码', '')
                node['widgets_values'] = [code] if code else []
    
    # 5. 清理冗余的顶层结构
    if 'nodes' in workflow and 'litegraph' in workflow:
        del workflow['nodes']
    if 'links' in workflow and 'litegraph' in workflow:
        del workflow['links']
    
    return workflow

# 使用示例
if __name__ == "__main__":
    import sys
    
    input_file = sys.argv[1] if len(sys.argv) > 1 else '工作流.json'
    output_file = sys.argv[2] if len(sys.argv) > 2 else input_file.replace('.json', '_标准化.json')
    
    with open(input_file, 'r', encoding='utf-8') as f:
        workflow = json.load(f)
    
    normalized = normalize_workflow(workflow)
    
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(normalized, f, indent=2, ensure_ascii=False)
    
    print(f"标准化完成: {output_file}")

使用场景：

  • 合并来自不同版本导出的工作流
  • 修复格式不规范的JSON文件
  • 批量处理多个工作流文件
  • 迁移旧版本工作流到新版本

验证脚本：

def validate_workflow_structure(workflow: Dict) -> tuple[bool, List[str]]:
    """
    验证工作流结构完整性
    
    返回:
        (是否有效, 错误列表)
    """
    errors = []
    
    # 检查必需字段
    if 'litegraph' not in workflow:
        errors.append("缺少litegraph字段")
        return False, errors
    
    litegraph = workflow['litegraph']
    nodes = litegraph.get('nodes', [])
    links = litegraph.get('links', [])
    
    # 检查节点ID唯一性
    node_ids = [node.get('id') for node in nodes]
    if len(node_ids) != len(set(node_ids)):
        errors.append("节点ID不唯一")
    
    # 检查连接有效性
    node_id_set = set(node_ids)
    for link in links:
        if isinstance(link, list) and len(link) >= 5:
            src_id, tgt_id = link[1], link[3]
            if src_id not in node_id_set:
                errors.append(f"连接引用不存在的源节点: {src_id}")
            if tgt_id not in node_id_set:
                errors.append(f"连接引用不存在的目标节点: {tgt_id}")
    
    return len(errors) == 0, errors
最后一次编辑于 19天前 0

暂无评论

推荐阅读
  13608564384   2025年12月16日   75   0   0 编程与工具
  19520938347   2025年04月07日   1007   1   0 量化策略编程与工具Python历史数据C++
  18958283423   2025年07月13日   868   0   0 编程与工具
  19520938347   2025年04月07日   975   1   1 量化策略编程与工具PythonC++
  18501669716   26天前   113   0   0 量化策略编程与工具经验分享低频交易
  18621003097   2025年12月25日   79   0   0 量化策略编程与工具
  18958283423   2025年06月16日   1016   1   1 编程与工具Python新手入门趋势跟踪代码分享
  13608564384   19天前   88   0   0 编程与工具
  13608564384   24天前   59   0   0 编程与工具
  18501669716   2025年12月19日   138   1   0 编程与工具数据可视化数据API
  13608564384   2025年12月27日   58   0   0 编程与工具
  AlphaSmith   2025年06月04日   664   1   4 编程与工具Python新手入门经验分享
  13654149687   26天前   70   0   0 中频交易编程与工具PythonAPI接口第一期线下课
13608564384
13608564384
作者其他文章 更多
压缩JSON的格式化与结构统

2026-01-12

工作流JSON高级编辑技巧

2026-01-06

工作流导出与IDE编辑适配小技巧-2

2025-12-27

工作流导出与IDE编辑适配小技巧

2025-12-16

常用标签
因子大赛第零期线下课C机器学习C++Python中频交易低频交易量化策略波动率策略动量策略统计套利事件驱动策略机器学习策略多因子模型高频交易均值回归趋势跟踪机器学习模型历史数据组合优化杠杆管理风险模型数据可视化数据API数据存储数据清洗实时数据API接口数据库clickhousemysqlredismongodb编程与工具活动与比赛第一期线下课第三期线下课第四期线下课第五期线下课python入门基础线上课线下课经验分享新手入门论坛公告学习资源书籍推荐代码分享策略讨论c#gonode.js