使用petl处理ETL流程中重复slug字段问题求助

我来帮你拆解下这个问题——你遇到的核心问题是petl的惰性求值特性以及循环逻辑的漏洞，导致重复slug没有被彻底处理。

为什么你的代码没生效？ #

1. petl Table是不可变且惰性的：每次调用etl.convert()都会返回一个新的Table对象，但所有转换操作都是延迟执行的，直到你实际迭代或输出Table时才会计算。你一开始获取的duplicates是基于原始Table的迭代结果，后续循环中修改Table后，并没有重新评估重复项，而且多次转换的惰性叠加可能导致修改没有真正应用到最终结果里。
2. 循环逻辑覆盖不全：etl.duplicates()返回的是原始表中所有重复slug的行，但你的循环只逐个修改这些行的slug。如果同一个slug对应多个行，你只修改了其中部分行（或者说，基于原始重复项的id修改），但可能剩下的行仍然保留着原始slug，导致重复依然存在。

正确的处理方式 #

推荐先统计所有slug的出现次数，然后一次性批量处理所有行，这样既避免惰性求值的坑，也能确保所有重复slug都被处理。

方案1：加载到内存处理（适合小数据集）

先把所有行加载到内存，统计每个slug的出现次数，再逐个修改：

from collections import defaultdict

# 先把表数据加载到内存，方便统计重复
rows = list(table)

# 统计每个slug的出现次数
slug_counts = defaultdict(int)
for row in rows:
    slug_counts[row.slug] += 1

# 定义转换函数：重复的slug追加id后缀
def modify_slug(row):
    if slug_counts[row.slug] > 1:
        return f"{slugify_unicode(row.slug)}-{str(row.id).encode('hex')}"
    return row.slug

# 重新构建处理后的table
table = etl.fromrows(rows).convert('slug', modify_slug, pass_row=True)

方案2：用petl原生操作处理（适合大数据集，无需全量加载）

利用petl的aggregate和join操作，在不加载全量数据的情况下统计并处理重复：

# 统计每个slug的出现次数
slug_counts = etl.aggregate(table, 'slug', len)

# 将统计结果与原表关联，得到每行对应的slug出现次数
table_with_counts = etl.join(
    table, 
    slug_counts, 
    key='slug', 
    lprefix='', 
    rprefix='count_'
)

# 转换slug：如果出现次数>1则追加id后缀
table = etl.convert(
    table_with_counts,
    'slug',
    lambda v, row: f"{slugify_unicode(v)}-{str(row.id).encode('hex')}" if row.count_slug > 1 else v,
    pass_row=True
)

# 可选：移除临时的count_slug字段
table = etl.cutout(table, 'count_slug')

关键要点 #

• 避免在循环中多次修改petl Table，惰性求值会让这种操作的结果不可预测。
• 先统计所有slug的出现情况，再批量处理，能确保所有重复项都被覆盖，不管是2个还是多个重复行。

内容的提问来源于stack exchange，提问作者Mo J. Mughrabi