如何检查Python DataFrame中员工汇报关系的循环引用？

核心思路 #

员工汇报关系可抽象为有向图：每个员工ID是节点，id指向parentid的箭头代表汇报关系。我们需要遍历每个节点，沿着汇报链路向上追踪，检查是否会回到起始节点——若出现这种情况，就说明存在循环引用。

实现方法 #

可以通过迭代追踪的方式，对每个员工的汇报链路逐一验证，记录已访问节点，避免重复计算。

基础迭代实现代码

import pandas as pd

def check_report_cycle(df):
    # 构建员工ID到经理ID的映射字典，提升查询效率
    parent_map = df.set_index('id')['parentid'].to_dict()
    # 记录已检查过的节点，避免重复处理
    checked_nodes = set()

    for emp_id in parent_map.keys():
        if emp_id in checked_nodes:
            continue
        current_path = []
        current_id = emp_id
        while current_id is not None:
            if current_id in current_path:
                # 找到循环，返回循环状态和具体路径
                cycle_start = current_path.index(current_id)
                return True, current_path[cycle_start:]
            if current_id in checked_nodes:
                # 该链路已验证过无循环，终止追踪
                break
            current_path.append(current_id)
            checked_nodes.add(current_id)
            # 跳转到上级ID，若上级不存在则终止（比如顶级管理者）
            current_id = parent_map.get(current_id)
    # 所有节点遍历完成，无循环
    return False, None

# 测试示例数据
test_df = pd.DataFrame({"id":[111,112,113],"parentid":[112,113,111]})
has_cycle, cycle_chain = check_report_cycle(test_df)
if has_cycle:
    print(f"发现循环引用：{' → '.join(map(str, cycle_chain))} → {cycle_chain[0]}")
else:
    print("未发现循环引用")

代码说明

• parent_map字典：快速通过员工ID找到其经理ID，避免反复查询DataFrame。
• checked_nodes集合：记录已验证过的节点，防止重复处理同一链路的节点，提升效率。
• 追踪逻辑：对每个未验证的节点，沿着parentid向上遍历，用current_path记录当前路径：
  • 若当前节点已在current_path中，说明形成循环，返回循环路径。
  • 若当前节点已在checked_nodes中，说明该链路之前已验证无循环，直接终止。

大数据量优化方案：拓扑排序 #

如果员工数据量较大，拓扑排序是更高效的方法——通过不断移除无上级（入度为0）的节点，若最终仍有节点未被处理，则说明存在循环。

def check_cycle_topological(df):
    parent_map = df.set_index('id')['parentid'].to_dict()
    # 统计每个节点的入度（即有多少员工向其汇报）
    in_degree = {emp_id: 0 for emp_id in parent_map.keys()}
    for manager_id in parent_map.values():
        if manager_id in in_degree:
            in_degree[manager_id] += 1

    # 初始化队列，放入所有入度为0的节点（顶级管理者）
    from collections import deque
    process_queue = deque([emp_id for emp_id, cnt in in_degree.items() if cnt == 0])
    processed_count = 0

    while process_queue:
        current_emp = process_queue.popleft()
        processed_count += 1
        manager = parent_map.get(current_emp)
        if manager in in_degree:
            in_degree[manager] -= 1
            if in_degree[manager] == 0:
                process_queue.append(manager)

    # 处理的节点数不等于总节点数，说明存在循环
    return processed_count != len(parent_map)

# 测试
print(check_cycle_topological(test_df))  # 输出True，代表存在循环

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