修复Pandas DataFrame转树形JSON的path与isCritial问题

问题概述 #

• 所有节点的path字段始终为空数组[]，正确逻辑应为根节点path为[]，子节点继承父节点路径（如二级节点path为["Total Tickets"]）
• 无法设置isCritial（拼写错误保留）标记，需将用户指定行（如Count为15的行）对应的全层级节点该标记设为True

输入数据 #

Total           Resolution  Category    Escalated   Count
Total Tickets   False       IT          False       4
Total Tickets   False       IT          True        3
Total Tickets   True        IT          False       1
Total Tickets   True        IT          True        15
Total Tickets   True        Unknown     True        1

当前输出 #

{"chart":{"data":{"path":[],"displayColumnLabel":"Total","displayValueLabel":"Total Tickets","values":[{"type":"ticket","value":24,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"Resolution","displayValueLabel":"True","values":[{"type":"ticket","value":17,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"Category","displayValueLabel":"IT","values":[{"type":"ticket","value":16,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"Escalated","displayValueLabel":"True","values":[{"type":"ticket","value":15,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"tickets","displayValueLabel":"15","values":[{"type":"ticket","value":15,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[]}]},{"path":[],"displayColumnLabel":"Escalated","displayValueLabel":"False","values":[{"type":"ticket","value":1,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"tickets","displayValueLabel":"1","values":[{"type":"ticket","value":1,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[]}]}]},{"path":[],"displayColumnLabel":"Category","displayValueLabel":"Unknown","values":[{"type":"ticket","value":1,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"Escalated","displayValueLabel":"True","values":[{"type":"ticket","value":1,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"tickets","displayValueLabel":"1","values":[{"type":"ticket","value":1,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[]}]}]}]},{"path":[],"displayColumnLabel":"Resolution","displayValueLabel":"False","values":[{"type":"ticket","value":7,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"Category","displayValueLabel":"IT","values":[{"type":"ticket","value":7,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"Escalated","displayValueLabel":"False","values":[{"type":"ticket","value":4,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"tickets","displayValueLabel":"4","values":[{"type":"ticket","value":4,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[]}]},{"path":[],"displayColumnLabel":"Escalated","displayValueLabel":"True","values":[{"type":"ticket","value":3,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[{"path":[],"displayColumnLabel":"tickets","displayValueLabel":"3","values":[{"type":"ticket","value":3,"change":0,"changeType":"neutral"}],"isCritial":false,"children":[]}]}]}]}]}}}}

当前函数代码 #

def create_hierarchical_json_new(df, value_column='Ticket Id'):
    # Extract column names
    columns = df.columns.tolist()
    

    # Build data and children sections
    def build_children(df, group_cols):
        if not group_cols:
            return []  # Base case: no more grouping columns, return empty list

        current_col = group_cols[0]
        grouped = df.groupby(current_col)
        children = []
        for value, group in grouped:
            is_terminal_node = len(group_cols) == 1
            child = {
                "path": [],
                "displayColumnLabel": current_col,
                "displayValueLabel": str(value),
                "values": [
                    {
                        "type": "ticket",
                        "value": group[value_column].sum(),
                        "change": 0,
                        "changeType": "neutral",
                    }
                ],
                "isCritial": False,
                "children": build_children(group, group_cols[1:]),
            }

            # Add one more node if this is the terminal node
            if is_terminal_node:
                child["children"].append({
                    "path": [],
                    "displayColumnLabel": "tickets",
                    "displayValueLabel": str(group[value_column].sum()),
                    "values": [
                        {
                            "type": "ticket",
                            "value": group[value_column].sum(),
                            "change": 0,
                            "changeType": "neutral",
                        }
                    ],
                    "isCritial": False,
                    "children": []
                })

            children.append(child)
        return children

    data = {
        "path": [],
        "displayColumnLabel": columns[0],
        "displayValueLabel": str(df[columns[0]].iloc[0]),
        "values": [
            {
                "type": "ticket",
                "value": df[value_column].sum(),
                "change": 0,
                "changeType": "neutral",
            }
        ],
        "isCritial": False,
        "children": build_children(df, columns[1:-1]),  # Group by all columns except first and last
    }

    # Combine meta and data sections
    result = {
        "chart": {
            "data": data
        }
    }

    return result

修改后的函数代码 #

def create_hierarchical_json_new(df, value_column='Count', critical_condition=None):
    columns = df.columns.tolist()
    
    # 预先获取所有关键路径（如果有指定条件）
    critical_paths = set()
    if critical_condition is not None:
        critical_rows = df[critical_condition]
        for _, row in critical_rows.iterrows():
            # 构建从根到当前行的完整路径
            path = [str(row[col]) for col in columns[:-1]]
            # 存储所有父路径（包含完整路径本身）
            for i in range(len(path)+1):
                critical_paths.add(tuple(path[:i]))

    def build_children(df, group_cols, parent_path):
        if not group_cols:
            return []

        current_col = group_cols[0]
        grouped = df.groupby(current_col)
        children = []
        for value, group in grouped:
            current_value_str = str(value)
            current_path = parent_path + [current_value_str]
            # 判断当前节点是否属于关键路径层级
            is_critical = tuple(current_path) in critical_paths or tuple(parent_path) in critical_paths
            is_terminal_node = len(group_cols) == 1
            
            child = {
                "path": parent_path.copy(),
                "displayColumnLabel": current_col,
                "displayValueLabel": current_value_str,
                "values": [
                    {
                        "type": "ticket",
                        "value": group[value_column].sum(),
                        "change": 0,
                        "changeType": "neutral",
                    }
                ],
                "isCritial": is_critical,
                "children": build_children(group, group_cols[1:], current_path),
            }

            if is_terminal_node:
                ticket_sum = group[value_column].sum()
                ticket_path = current_path.copy()
                ticket_is_critical = tuple(ticket_path) in critical_paths
                child["children"].append({
                    "path": ticket_path,
                    "displayColumnLabel": "tickets",
                    "displayValueLabel": str(ticket_sum),
                    "values": [
                        {
                            "type": "ticket",
                            "value": ticket_sum,
                            "change": 0,
                            "changeType": "neutral",
                        }
                    ],
                    "isCritial": ticket_is_critical,
                    "children": []
                })

            children.append(child)
        return children

    # 根节点处理
    root_path = []
    root_is_critical = tuple(root_path) in critical_paths
    data = {
        "path": root_path,
        "displayColumnLabel": columns[0],
        "displayValueLabel": str(df[columns[0]].iloc[0]),
        "values": [
            {
                "type": "ticket",
                "value": df[value_column].sum(),
                "change": 0,
                "changeType": "neutral",
            }
        ],
        "isCritial": root_is_critical,
        "children": build_children(df, columns[1:-1], root_path),
    }

    result = {
        "chart": {
            "data": data
        }
    }

    return result

使用示例 #

import pandas as pd
import json

# 构造输入数据
data = [
    ["Total Tickets", False, "IT", False, 4],
    ["Total Tickets", False, "IT", True, 3],
    ["Total Tickets", True, "IT", False, 1],
    ["Total Tickets", True, "IT", True, 15],
    ["Total Tickets", True, "Unknown", True, 1],
]
df = pd.DataFrame(data, columns=["Total", "Resolution", "Category", "Escalated", "Count"])

# 生成JSON，指定Count=15的行为关键行
result = create_hierarchical_json_new(df, value_column='Count', critical_condition=df['Count'] == 15)

# 打印格式化后的结果
print(json.dumps(result, indent=2))

修改说明 #

1. 路径继承实现：
   • 为build_children函数新增parent_path参数，传递父节点的路径信息
   • 子节点的path直接复制父节点路径，终端tickets节点继承上级完整路径
2. 关键节点标记实现：
   • 新增critical_condition参数，支持用户自定义关键行筛选条件
   • 预先计算所有关键行的完整路径及其所有父路径，存储为元组集合以实现快速查找
   • 每个节点生成时，检查自身路径或父路径是否属于关键路径集合，动态设置isCritial标记

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