Python 3.13：基于列表相邻元素生成含parallelNode的JSON负载

核心思路 #

遍历请求列表时维护一个临时并行分组缓存，每处理一个请求就判断其是否符合当前分组的合并条件：

• 若当前无分组，直接初始化新分组
• 若请求与当前分组的command和command_options一致、且device不同，则加入当前分组
• 若不符合条件，先处理当前分组（多元素则包装为parallelNode，单元素则直接加入结果），再初始化新分组
• 遍历结束后处理最后剩余的分组

完整代码实现 #

from typing import List, Dict, Any

# 定义请求数据结构
class Request:
    def __init__(self, command: str, device: str, command_options: str, payload: Dict[str, Any]):
        self.command = command
        self.device = device
        self.command_options = command_options
        self.payload = payload

def build_parallel_nodes(requests: List[Request]) -> Dict[str, Any]:
    nodes = []
    current_group = None  # 存储格式：(分组键, payload列表)，分组键=(command, command_options)
    
    for req in requests:
        group_key = (req.command, req.command_options)
        if current_group is None:
            current_group = (group_key, [req.payload])
        else:
            current_key, current_payloads = current_group
            # 检查是否符合并行条件：键一致且device不重复
            if current_key == group_key and req.device not in [p['device'] for p in current_payloads]:
                current_payloads.append(req.payload)
            else:
                # 处理当前分组
                if len(current_payloads) > 1:
                    nodes.append({"parallelNode": current_payloads})
                else:
                    nodes.append(current_payloads[0])
                # 初始化新分组
                current_group = (group_key, [req.payload])
    
    # 处理最后一个分组
    if current_group is not None:
        _, payloads = current_group
        if len(payloads) > 1:
            nodes.append({"parallelNode": payloads})
        else:
            nodes.append(payloads[0])
    
    return {"nodes": nodes}

# 测试示例
if __name__ == "__main__":
    requests = [
        Request("C1", "D1", "C1O1", {"device": "D1", "data": "payload1"}),
        Request("C2", "D2", "C2O1", {"device": "D2", "data": "payload2"}),
        Request("C2", "D3", "C2O1", {"device": "D3", "data": "payload3"}),
        Request("C3", "D3", "C3O1", {"device": "D3", "data": "payload4"}),
    ]
    
    import json
    print(json.dumps(build_parallel_nodes(requests), indent=2))

关键细节说明 #

1. 分组键设计：用(command, command_options)作为分组唯一标识，确保只有核心参数一致的请求才会被合并。
2. Device去重：示例中假设payload包含device字段，若你的payload不存储该字段，可单独维护一个current_devices列表来判断重复。
3. 边界处理：遍历结束后必须处理剩余分组，避免遗漏最后一批请求。
4. 版本适配：代码未依赖Python 3.13专属特性，但可直接在该版本下运行；若要利用3.13的itertools优化，可调整遍历逻辑，当前实现已足够简洁高效。

输出结果 #

运行测试代码后将得到符合需求的JSON结构：

{
  "nodes": [
    {
      "device": "D1",
      "data": "payload1"
    },
    {
      "parallelNode": [
        {
          "device": "D2",
          "data": "payload2"
        },
        {
          "device": "D3",
          "data": "payload3"
        }
      ]
    },
    {
      "device": "D3",
      "data": "payload4"
    }
  ]
}

内容的提问来源于stack exchange，提问作者Jameson_uk