影刀RPA避坑指南_数据重复的5个来源与去重方案翻页重叠流程重跑与DOM重复

影刀RPA避坑指南：数据重复的5个来源与去重方案——别再因为翻页重叠多采一倍的量

影刀RPA数据重复是最隐蔽的坑。流程没报错、数据写进去了、Excel也能打开——你以为一切正常，直到运营说："这个商品的销量怎么是两个不同数字？"一查才发现同一商品被采集了两次。本文拆解重复数据的5个来源，每种都给去重方案。

作者：林焱 | 影刀RPA避坑系列

重复数据的5个来源

来源一：翻页重叠

场景：很多网站翻页时有1-2条数据会跨页重复。列表页按销量排序，销量变化后新页面顶掉了旧页面的首条。

解决方案：用唯一的业务ID去重

def deduplicate_by_id(data_list, id_field='商品ID'):
"""
按业务ID去重，保留最新一条
"""
seen = {}
for item in data_list:
item_id = item.get(id_field)
if item_id:
# 如果ID重复，保留后面那条（更可能是最新数据）
seen[item_id] = item

return list(seen.values())

如果是翻页实时的，在每页采集后跟已有数据合并去重：

all_data = []
for page in range(1, 21):
page_data = collect_page(page)

# 合并并去重
all_data.extend(page_data)

[video(video–YHt9Xvua–1781590448071)(type–csdn)(url–https://live.csdn.net/v/embed/526818)(image–https://v–blog.csdnimg.cn/asset/582d14c3bd0451c5399cd990b56e2a0d/cover/Cover0.jpg)(title–拼多多店群自动化报活动上架！)]

all_data = deduplicate_by_id(all_data)

print(f"第{page}页：采集{len(page_data)}条，去重后累计{len(all_data)}条")

来源二：流程重跑导致重复

场景：流程采集到一半挂了，重启后从头开始，前面已入库的数据重复写入。

解决方案一：任务断点记录

import json

def load_checkpoint(flow_name):
"""读取上次执行断点"""
checkpoint_file = f"checkpoint_{flow_name}.json"
try:
with open(checkpoint_file, 'r') as f:
return json.load(f)
except:
return {'last_keyword': None, 'last_page': 0, 'collected_ids': set()}

def save_checkpoint(flow_name, checkpoint):
"""保存当前进度"""
checkpoint['collected_ids'] = list(checkpoint['collected_ids'])
with open(f"checkpoint_{flow_name}.json", 'w') as f:
json.dump(checkpoint, f, ensure_ascii=False)

重跑时从断点恢复，跳过已处理的部分。

解决方案二：写入前去重

def save_with_dedup(new_data, existing_file, unique_key='商品ID'):
"""
写入前和已有数据去重
"""
import pandas as pd

# 读取已有数据
try:
existing_df = pd.read_excel(existing_file)
existing_ids = set(existing_df[unique_key].dropna().values)
except:
existing_df = pd.DataFrame()
existing_ids = set()

# 过滤重复
df_new = pd.DataFrame(new_data)
if unique_key in df_new.columns:
df_new = df_new[~df_new[unique_key].isin(existing_ids)]

# 合并保存
df_merged = pd.concat([existing_df, df_new], ignore_index=True)
df_merged.to_excel(existing_file, index=False)

return len(df_new), len(existing_ids)

来源三：数据排序变化

场景：昨天采集的"蓝牙耳机"第1页是商品A排第一，今天销量变了，商品B排第一，A掉到第3页。如果每天存独立文件没问题，做历史汇总时就会出现同一商品被不同日期记录覆盖。

解决方案：按日期+ID做历史快照

# 每天的数据存独立文件，不做合并覆盖
today = datetime.now().strftime("%Y-%m-%d")
daily_file = f"data/price_{today}.xlsx"
df.to_excel(daily_file, index=False)

# 分析时跨文件按ID+日期查最新
def get_latest_snapshot(data_dir, days=30):
"""获取最近N天每个商品的最新数据"""
all_files = sorted(glob.glob(f"{data_dir}/price_*.xlsx"))
latest_files = all_files[–days:]

all_data = []
for f in latest_files:
df = pd.read_excel(f)
all_data.append(df)

combined = pd.concat(all_data, ignore_index=True)
# 按商品ID分组，保留每组最新日期那条
return combined.sort_values('采集日期').groupby('商品ID').last().reset_index()

来源四：DOM重复渲染

场景：滚动加载时DOM被重复渲染，采集脚本把同一条数据抓了两遍。

解决方案：在采集时实时去重

collected_ids = set()
results = []

def collect_with_dedup(card_elements):
for card in card_elements:
# 提取唯一标识（如商品链接或ID）
product_url = card.get_attribute('href') or ''
product_id = extract_id_from_url(product_url)

if product_id in collected_ids:
continue # 已采集过，跳过

collected_ids.add(product_id)

# 采集数据
data = extract_card_data(card)
results.append(data)

来源五：并发写入冲突

场景：开了3个采集实例，都往同一个Excel写，互相覆盖。

解决方案：每实例独立文件 + 定时合并

参考之前并发避坑篇的"独立文件+最终合并"方案。

去重策略速查表

TEMU店群矩阵自动化运营核价报活动

| DOM重复 | 采集时 | 元素唯一属性 | 采集循环内去重 | | 并发冲突 | 写入中 | 任务分片 | 独立文件互不干扰 |

去重不是万能的——什么数据不应该去重

有些场景去重反而是错的：

• 价格监控：同一商品每天采集一次，价格不同，不能按ID去重
• 评论采集：同一商品的新评论和旧评论是不同的数据
• 订单流水：每条订单都有独立订单号，本身不会重复

去重的原则是：数据本身是幂等的（同一ID多次采集内容相同），才去重。内容可能变化的，区分维度（如采集时间），不要硬去重。

小结

数据去重要在两个层面做：

• 采集层：循环中实时去重（DOM重复、翻页重叠）
• 存储层：写入前去重（流程重跑、历史数据）

两层防线都做，重复数据的漏网之鱼基本为零。

内容标签：影刀RPA / 数据去重 / 翻页重叠 / 断点续传 / 数据质量

作者：林焱

本文是「影刀RPA从零到实战」系列文章之一，持续更新实操技巧与避坑经验。