如何在Python Pandas中检测Excel导入数据的循环引用并抛出错误

一、预处理与基础校验 #

先把Excel数据读入DataFrame，统一处理空值，同时先过滤掉非法父部件（非NULL但不在部件编号列表里的），避免无效遍历：

• 用pd.read_excel读取文件，将父部件列的空单元格、空字符串统一转为None/pd.NA。
• 提取所有合法部件编号的集合，筛选出父部件存在但不在合法集合里的行，直接抛出错误。

二、循环引用核心检测逻辑 #

循环引用的本质是遍历路径中出现重复节点，用带路径跟踪的DFS就能精准检测：

• 对每个根节点（父部件为NULL的部件），维护两个数据结构：
  • visited：全局集合，记录所有已处理过的部件，避免重复遍历。
  • current_path：当前遍历的路径列表，用来检测是否出现循环（如果当前部件已经在current_path里，说明形成闭环）。
• 遍历流程：
  1. 检查当前部件是否在current_path中：是则抛出Circular Reference Found错误，同时输出循环路径方便排查。
  2. 如果部件已在visited中，直接返回（已处理过）。
  3. 把部件加入visited和current_path，查找它的所有子部件。
  4. 递归遍历每个子部件，遍历完成后从current_path中移除当前部件（回溯）。

三、层级构建与检测整合代码示例 #

把层级构建和循环检测合并到同一个遍历过程，代码如下：

import pandas as pd

def build_hierarchy_with_cycle_check(df):
    # 预处理：统一空值
    df['父部件编号（Parent Part Number）'] = df['父部件编号（Parent Part Number）'].replace({pd.NA: None, '': None})
    
    # 基础校验：父部件必须存在（除了NULL）
    valid_parts = set(df['部件编号（Part Number）'].unique())
    invalid_rows = df[(df['父部件编号（Parent Part Number）'].notna()) & (~df['父部件编号（Parent Part Number）'].isin(valid_parts))]
    if not invalid_rows.empty:
        invalid_parents = invalid_rows['父部件编号（Parent Part Number）'].unique().tolist()
        raise ValueError(f"Invalid Parent Part Numbers: {invalid_parents}")
    
    # 初始化层级结构和访问记录
    hierarchy = {}
    visited = set()
    
    def dfs(part_num, current_path):
        # 检测循环
        if part_num in current_path:
            cycle = current_path[current_path.index(part_num):] + [part_num]
            raise ValueError(f"Circular Reference Found: {' → '.join(map(str, cycle))}")
        # 已处理过的节点直接返回
        if part_num in visited:
            return
        # 标记访问
        visited.add(part_num)
        current_path.append(part_num)
        # 获取子部件
        children = df[df['父部件编号（Parent Part Number）'] == part_num]['部件编号（Part Number）'].tolist()
        hierarchy[part_num] = children
        # 遍历子部件
        for child in children:
            dfs(child, current_path)
        # 回溯路径
        current_path.pop()
    
    # 遍历所有根节点（父部件为NULL）
    root_parts = df[df['父部件编号（Parent Part Number）'].isna()]['部件编号（Part Number）'].tolist()
    for root in root_parts:
        dfs(root, [])
    
    return hierarchy

# 调用示例
if __name__ == "__main__":
    try:
        df = pd.read_excel('product_hierarchy.xlsx')
        hierarchy = build_hierarchy_with_cycle_check(df)
        print("Hierarchy built successfully:", hierarchy)
    except ValueError as e:
        print(e)

四、大数据量优化 #

如果部件数量很多，递归DFS可能触发栈溢出，改用迭代式DFS替代递归：

def iterative_dfs(root, df, valid_parts, hierarchy, visited):
    stack = [(root, [])]
    while stack:
        part_num, current_path = stack.pop()
        if part_num in current_path:
            cycle = current_path[current_path.index(part_num):] + [part_num]
            raise ValueError(f"Circular Reference Found: {' → '.join(map(str, cycle))}")
        if part_num in visited:
            continue
        visited.add(part_num)
        new_path = current_path + [part_num]
        children = df[df['父部件编号（Parent Part Number）'] == part_num]['部件编号（Part Number）'].tolist()
        hierarchy[part_num] = children
        # 逆序入栈，保证遍历顺序和递归一致
        for child in reversed(children):
            stack.append((child, new_path))

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