如何高效识别模拟企业网络中二阶以上的邮件循环链？

场景背景 #

某企业存在num_employees名员工，总计交换num_emails封邮件，通信规则如下：

• 每位员工最多向另一位员工发送一封邮件
• 严格遵循一对一通信模式，禁止同时向多个收件人发送邮件

测试环境搭建（Python + Faker） #

使用Python和Faker库模拟上述场景，代码实现如下：

from faker import Faker
from random import Random
rand = Random()
fake = Faker().unique

num_employees = 200
num_emails = 2000

employees = [fake.email() for _ in range(num_employees)]

emails = []
for i in range(num_emails):
    email = (rand.choice(employees), rand.choice(employees))
    # 确保所有发件人-收件人组合唯一
    while email in emails:
        email = (rand.choice(employees), rand.choice(employees))

    emails.append(email)

注：emails列表的索引代表邮件的发送时间顺序

现有回复关系识别代码 #

目前已实现二阶以内回复关系的识别（即A→B后出现B→A），代码如下：

received_a_reply = {}
for i, (sender, recipient) in enumerate(emails):

    if (recipient, sender) in emails[i:]:
        received_a_reply[i] = [sender, recipient]

print(f"{round(len(received_a_reply)/len(emails) * 100)}% of emails are replies")
print(f"{len(emails)} emails")
print(f"{len(received_a_reply)} emails received a reply")
print(f"{len(emails) - len(received_a_reply)} emails weren't in a thread")

核心问题 #

上述代码仅能识别二阶回复循环，现需检测三阶及以上的循环链（例如：Matthew→Mark→John→Matthew）。实际数据集包含约4000名员工与40000封邮件，要求同时优化时间与内存占用，避免通过穷举所有组合再过滤非循环实例的低效方式。

解决方案 #

1. 构建有向图模型 #

将员工视为图的节点，每封邮件（发件人→收件人）视为一条有向边，构建邻接表形式的有向图。由于同一发件人对同一收件人最多发送一封邮件，图中不会存在重复的同方向边。

2. 高效环检测算法 #

针对大规模图，采用深度优先搜索(DFS)结合回溯的方式检测环，同时做以下优化：

• 节点状态标记：给每个节点标记三种状态：未访问、正在访问（当前DFS栈中）、已访问，避免重复遍历，时间复杂度可控制在O(V+E)（V为员工数，E为邮件数）。
• 路径长度限制：仅关注三阶及以上的环，在DFS过程中记录路径长度，达到阈值后停止递归，减少无效计算。
• 提前剪枝：若当前节点后续无出边，直接回溯，跳过无效路径。

3. 代码实现示例 #

def find_cycles(graph, min_length=3):
    cycles = set()
    visited = {}  # 节点状态：0=未访问，1=正在访问，2=已访问

    def dfs(node, path):
        if visited.get(node, 0) == 1:
            # 找到环，截取从当前节点开始的路径
            idx = path.index(node)
            cycle = tuple(path[idx:])
            if len(cycle) >= min_length:
                # 标准化环（避免重复记录同一环的不同起点，如A→B→C与B→C→A视为同一环）
                sorted_cycle = tuple(sorted(cycle, key=lambda x: cycle.index(x)))
                cycles.add(sorted_cycle)
            return
        if visited.get(node, 0) == 2:
            return
        
        visited[node] = 1
        for neighbor in graph.get(node, []):
            dfs(neighbor, path + [neighbor])
        visited[node] = 2

    # 遍历所有未访问节点启动DFS
    for node in graph:
        if visited.get(node, 0) == 0:
            dfs(node, [node])
    
    return cycles

# 将邮件列表转换为有向图邻接表
graph = {}
for sender, recipient in emails:
    if sender not in graph:
        graph[sender] = []
    graph[sender].append(recipient)

# 查找三阶及以上的循环链
cycles = find_cycles(graph, min_length=3)
print(f"找到{len(cycles)}个三阶及以上的循环链")

4. 内存与时间优化补充 #

• 图结构优化：使用字典存储邻接表，相比列表更节省内存，且节点查找效率更高。
• 环去重处理：通过标准化环的方式减少重复记录，降低内存占用。
• 分批处理适配：若数据集过大，可分批次处理邮件，但需保留节点访问状态以避免遗漏跨批次的环。

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