用并查集归并日志中的多标识账号

问题场景

日志里同一用户可能留下多个标识：注册邮箱、手机号、设备 ID。单看一条记录，往往只有其中两个标识；但多条记录通过共同标识串联后，可能属于同一账号。例如一条记录同时出现邮箱和手机号，另一条记录同时出现手机号和设备 ID，三者应合并。若逐条比较，容易漏掉传递关系。

本文把每个标识视为图节点，把一条记录中出现的两个标识视为一条边，用并查集求连通分量。每条记录执行一次 union；若一条记录包含两个以上标识，则用第一个标识与其余标识依次 union，效果等同于把记录内标识连成一组。处理完所有记录后，做路径压缩，再按根节点分组输出。

最小可运行实现

下面的脚本包含标识归一化、并查集、分组输出和示例数据。示例数据中的空字符串表示该字段缺失。

from collections import defaultdict
import json


class UnionFind:
    def __init__(self):
        self.parent = {}
        self.rank = {}

    def add(self, x):
        if x not in self.parent:
            self.parent[x] = x
            self.rank[x] = 0

    def find(self, x):
        root = x
        while self.parent[root] != root:
            root = self.parent[root]
        while self.parent[x] != x:
            p = self.parent[x]
            self.parent[x] = root
            x = p
        return root

    def union(self, a, b):
        ra, rb = self.find(a), self.find(b)
        if ra == rb:
            return False
        if self.rank[ra] < self.rank[rb]:
            ra, rb = rb, ra
        self.parent[rb] = ra
        if self.rank[ra] == self.rank[rb]:
            self.rank[ra] += 1
        return True


def normalize(record):
    email = (record.get('email') or '').strip().lower()
    phone = (record.get('phone') or '').strip().replace(' ', '').replace('-', '')
    device = (record.get('device_id') or '').strip()
    result = []
    if email:
        result.append('email:' + email)
    if phone:
        result.append('phone:' + phone)
    if device:
        result.append('device:' + device)
    return result


def merge_accounts(records):
    uf = UnionFind()
    parsed = []
    no_identifier = []

    for rec in records:
        rid = rec.get('record_id')
        ids = normalize(rec)
        for item in ids:
            uf.add(item)
        if len(ids) >= 2:
            base = ids[0]
            for other in ids[1:]:
                uf.union(base, other)
        if not ids:
            no_identifier.append(rid)
        parsed.append((rid, ids))

    groups = defaultdict(list)
    for item in uf.parent:
        groups[uf.find(item)].append(item)

    record_map = defaultdict(list)
    for rid, ids in parsed:
        if ids:
            record_map[uf.find(ids[0])].append(rid)

    accounts = []
    for root, members in groups.items():
        accounts.append({
            'root': root,
            'identifiers': sorted(members),
            'record_ids': record_map.get(root, [])
        })
    accounts.sort(key=lambda x: x['root'])
    return accounts, no_identifier


if __name__ == '__main__':
    records = [
        {'record_id': 'r1', 'email': 'A@Example.com', 'phone': '138-0000-0001', 'device_id': ''},
        {'record_id': 'r2', 'email': '', 'phone': '13800000001', 'device_id': 'dev-1'},
        {'record_id': 'r3', 'email': '', 'phone': '', 'device_id': 'dev-2'},
        {'record_id': 'r4', 'email': 'b@example.com', 'phone': '', 'device_id': ''},
        {'record_id': 'r5', 'email': '', 'phone': '', 'device_id': ''},
    ]
    accounts, no_identifier = merge_accounts(records)
    print(json.dumps({'accounts': accounts, 'no_identifier': no_identifier}, ensure_ascii=False, indent=2))

运行方法

将代码保存为 union_log_accounts.py，在 Python 3 环境执行：

python3 union_log_accounts.py

示例中 r1 的邮箱统一为小写，手机号去掉分隔符后与 r2 的手机号相同，因此 r1、r2 合并；r3 只有设备 ID，r4 只有邮箱，二者分别形成独立账号；r5 没有任何标识，进入 no_identifier。实际使用时，把 records 替换为日志解析后的字典列表即可，字段名按业务调整。

边界与限制

• 归一化规则要先确认。邮箱通常转小写、去首尾空白；手机号可去空格和短横线；设备 ID 去空白。不要随意删除邮箱本地部分的点号，也不要把不同国家的手机号规则混用，否则会把不同用户误合并。
• 标识要加类型前缀，如 email:、phone:、device:，避免不同字段恰好同值造成碰撞。
• 无标识记录不进入并查集，单独返回，便于补采或人工处理；单标识记录会形成只含一个标识的账号，后续批次出现新边时才能合并。本文是单批静态处理，跨批次合并需要持久化并查集结构或重新计算。
• 并查集只保证等价关系的传递闭包，不能判断共享设备、公共邮箱、手机号换绑等业务语义，也不能自动识别脏数据。若同一设备被多人使用，合并结果可能偏大，需要业务规则或人工复核。
• 根节点取决于 union 顺序，不保证稳定。对外输出账号 ID 时，建议使用组内最小标识、业务主键或另行生成的稳定 ID，不要直接把根节点当账号 ID。
• 代码用迭代方式实现 find 并配合按秩合并，避免递归过深；标识数量很大时，内存占用与标识总数同阶，需关注日志规模。

小结

并查集适合处理「多条记录把多个标识连起来」的账号归并问题。实现重点不在 union 本身，而在标识归一化、单标识与无标识记录的去向，以及根节点不稳定带来的输出约定。先把这些边界定义清楚，再接入日志解析流程，结果更容易核对。