无环森林中叶子节点到所有可达节点关联关系高效提取方案咨询

我此前因信息不全分多次提问，现将完整问题说明如下：
我有存储两列数据的文本文件，第一列为前序节点、第二列为后继节点，我使用如下代码加载数据：

[line.split() for line in open('data.txt', encoding ='utf-8')]

示例输入数据 #

文件内的示例数据如下：

ANALYTICAL_BALANCE BFG_DEPOSIT 
CUSTOMER_DETAIL BALANCE 
BFG_2056 FFD_15 
BALANCE BFG_16 
BFG_16 STAT_HIST 
ANALYTICAL_BALANCE BFG_2056 
CUSTOM_DATA AND_11 
AND_11 DICT_DEAL 
DICT_DEAL BFG_2056 

加载后的数据格式如下：

[[ANALYTICAL_BALANCE,BFG_DEPOSIT],
[CUSTOMER_DETAIL,BALANCE],
[BFG_2056, FFD_15],
[BALANCE,BFG_16],
[BFG_16,STAT_HIST],
[ANALYTICAL_BALANCE,BFG_2056],
[CUSTOM_DATA,AND_11],
[AND_11,DICT_DEAL],
[DICT_DEAL,BFG_2056]]

现有实现逻辑 #

1. 邻接表构建代码 #

我构建邻接表关联数据，代码如下：

def create_adj(edges):
    adj = {}   # or use defaultdict(list) to avoid `if` in the loop below
    for a, b in edges:
        if not a in adj:
            adj[a] = []
        if not b in adj:
            adj[b] = []
        adj[a].append(b)
    return adj

2. 全路径递归遍历代码 #

随后通过递归遍历获取全路径，代码如下：

def all_paths(adj):
    def recur(path):
        node = path[-1]
        neighbors = [neighbor for neighbor in adj[node] if not neighbor in path]
        if not neighbors:
            yield path
        for neighbor in neighbors:
            yield from recur(path + [neighbor])

    for node in adj:
        yield from recur([node])

3. 运行输出示例 #

运行后输出的路径示例如下：

data = [
    ["ANALYTICAL_BALANCE","BFG_DEPOSIT"],
    ["CUSTOMER_DETAIL","BALANCE"],
    ["BFG_2056", "FFD_15"],
    ["BALANCE","BFG_16"],
    ["BFG_16","STAT_HIST"],
    ["ANALYTICAL_BALANCE","BFG_2056"],
    ["CUSTOM_DATA","AND_11"],
    ["AND_11","DICT_DEAL"],
    ["DICT_DEAL","BFG_2056"]
]

adj = create_adj(data)

print([path for path in all_paths(adj) if len(path) > 1])


[ANALYTICAL_BALANCE,BFG_DEPOSIT]
[CUSTOMER_DETAIL,BALANCE,BFG_16,STAT_HIST]
[BFG_2056,FFD_15]
[BALANCE,BFG_16,STAT_HIST]
[ANALYTICAL_BALANCE,BFG_2056,FFD_15]
[CUSTOM_DATA,AND_11,DICT_DEAL,BFG_2056,FFD_15]
[AND_11,DICT_DEAL,BFG_2056,FFD_15]
[DICT_DEAL,BFG_2056,FFD_15]

需求说明 #

上述关联关系可视为由多棵无环树组成的森林，输入数据特性保证无环。现在我的需求是：获取每棵树中所有叶子节点到其路径上所有节点的关联对，示例预期输出如下：

Tree1:
ANALYTICAL_BALANCE BFG_DEPOSIT

Tree2:
ANALYTICAL_BALANCE BFG_2056
ANALYTICAL_BALANCE FFD_15
CUSTOM_DATA AND_11
CUSTOM_DATA DICT_DEAL
CUSTOM_DATA BFG_2056
CUSTOM_DATA FFD_15

Tree3:
CUSTOMER_DETAIL BALANCE
CUSTOMER_DETAIL BFG_16
CUSTOMER_DETAIL STAT_HIST

性能问题与完整实现代码 #

我最初的实现是生成全路径后过滤非叶子节点作为起点的关联对，150行规模的输入可正常运行，但13k行的全量数据运行2天仍无结果。现寻求最高效的算法、适配的数据类型（如列表、DataFrame等），最终输出的关联对将通过openpyxl存入Excel，实现筛选后继节点时可查看所有关联的前置叶子节点的效果。以下是我的完整实现代码：

import itertools
from openpyxl import Workbook

# create adjacencies
def create_adj(edges):
    adj = {}
    for a, b in edges:
        if not a in adj:
            adj[a] = []
        if not b in adj:
            adj[b] = []
        adj[a].append(b)
    return adj
 
# find all paths
def all_paths(adj):
    def recur(path):
        node = path[-1]
        neighbors = [neighbor for neighbor in adj[node] if not neighbor in path]
        if not neighbors:
            yield path
        for neighbor in neighbors:
            yield from recur(path + [neighbor])
 
    for node in adj:
        yield from recur([node])
 
# delete the  connections from list 
def conn_deletion(list, list_deletion):
    after_del = [x for x in list if x[0] not in list_deletion]
    return after_del
 
 # get paths where len of path is > 2 and save them as a leaf to node. Also save connections to deletion.
def unpack_paths(my_list):
    list_of_more_succ = []
    to_deletion = []
    for item in my_list:
        if len(item) == 1:
            print("len 1",item)
            to_deletion.append(item[0])
        elif len(item) > 2:
            for i in range(1, len(item) - 1):
                to_deletion.append(item[i])
                print("len > 2", item[i])
                if [item[0], item[i]] in list_of_more_succ:
                    pass
                else:
                    list_of_more_succ.append([item[0], item[i]])
    list_concat = my_list + list_of_more_succ
    sorted_list = list(k for k, _ in itertools.groupby(list_concat))
    final = conn_deletion(sorted_list, list(dict.fromkeys(to_deletion)))
    return final
 
 
data = [line.split() for line in open('data.txt', encoding='utf-8')]
 
adj = create_adj(data)
print(adj)
 
workbook = Workbook()
sheet = workbook.active
sheet["A1"] = "Source"
sheet["B1"] = "Child"
 
loaded = list(all_paths(adj))
 
final_edited = unpack_paths(loaded)
 
# Save data to excel file. We don't want paths with len == 1 or > 2.
for row, item in enumerate(final_edited, start=2):
    if len(item) > 2:
        pass
    elif len(item) == 1:
        pass
    else:
        sheet[f"A{row}"] = item[0]
        sheet[f"B{row}"] = item[1]
workbook.save("DataMap.xlsx")

优化方案 #

原代码性能瓶颈 #

原代码生成全路径的逻辑会产生大量冗余路径，同时判断节点是否在路径中的操作为O(n)复杂度，面对13k行规模的输入时间复杂度指数级上升，是运行卡顿的核心原因。

高效实现思路 #

因为输入是确定无环的森林结构，不需要生成全路径，直接从根节点（入度为0的节点）向下遍历，遍历过程中记录当前根节点到路径上所有节点的关联对即可，整体时间复杂度为O(N)，N为节点总数：

• 第一步统计所有节点的入度，入度为0的节点就是每棵树的根节点
• 第二步对每个根节点进行BFS/DFS遍历，每访问到一个节点就输出「根节点, 当前节点」的关联对
• 第三步收集所有关联对直接写入Excel

优化后代码 #

from openpyxl import Workbook
from collections import deque, defaultdict

def main():
    # 加载数据
    edges = [line.strip().split() for line in open('data.txt', encoding='utf-8') if line.strip()]
    
    # 构建邻接表和入度统计
    adj = defaultdict(list)
    in_degree = defaultdict(int)
    all_nodes = set()
    for a, b in edges:
        adj[a].append(b)
        in_degree[b] += 1
        all_nodes.add(a)
        all_nodes.add(b)
    
    # 找所有根节点（入度为0）
    roots = [node for node in all_nodes if in_degree.get(node, 0) == 0]
    
    # BFS遍历每个根节点，收集关联对
    result = []
    for root in roots:
        q = deque()
        q.append(root)
        while q:
            cur = q.popleft()
            # 排除自己到自己的关联对
            if cur != root:
                result.append([root, cur])
            # 遍历子节点
            for neighbor in adj[cur]:
                q.append(neighbor)
    
    # 写入Excel
    workbook = Workbook()
    sheet = workbook.active
    sheet["A1"] = "Source"
    sheet["B1"] = "Child"
    for row, item in enumerate(result, start=2):
        sheet[f"A{row}"] = item[0]
        sheet[f"B{row}"] = item[1]
    workbook.save("DataMap.xlsx")

if __name__ == "__main__":
    main()

效果说明 #

• 13k行输入数据可在几秒内处理完成，完全避免原代码的指数级复杂度问题
• 不需要额外去重和过滤，遍历逻辑本身保证生成的关联对完全符合需求
• 内存占用极低，不需要存储大量中间路径

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