影刀RPA进阶教程_数据采集去重策略与增量更新方案

影刀RPA进阶教程：数据采集中的去重策略与增量更新方案

数据采集最怕两件事：重复采集和全量重跑。

采集了 5000 条数据，跑完发现有 2000 条是重复的。或者流程崩了，只能把全部数据重新跑一遍，跑完发现数据翻倍了。

这篇文章讲去重和增量采集的实战方案。

一、去重的三个层面

最佳实践：三层都用。采集前过滤是最省资源的，采集后清洗是最后一道防线。

二、采集前去重——边跑边记

在采集循环里维护一个"已采集 ID 集合"，每次判断。

拼多多店群自动化上架方案

# 用 Python set 实现采集前去重
已采集ID = set() # 存为全局变量或流程参数

def should_collect(item_id):
"""判断是否应该采集这条数据"""
if item_id in 已采集ID:
return False
已采集ID.add(item_id)
return True

# 在采集循环中使用
遍历商品列表:
商品ID = 获取元素文本("商品ID元素")

if not should_collect(商品ID):
continue # 跳过重复

# 正常采集逻辑
名称 = 获取元素文本("商品名称")
价格 = 获取元素文本("商品价格")
写入行数据(结果表, [商品ID, 名称, 价格])

set 的优点：查找是 O(1)，5000 条数据去重和 50 条一样快。

set 的缺点：只在内存里，流程重启后就丢了。如果要做跨次运行的增量去重，需要用文件。

三、文件持久化——跨次运行也不重复

import os

HISTORY_FILE = r"D:\\采集记录\\已采集ID.txt"

def load_collected_ids():
"""从文件加载已采集的 ID 集合"""
if not os.path.exists(HISTORY_FILE):
return set()

with open(HISTORY_FILE, "r", encoding="utf-8") as f:
ids = [line.strip() for line in f if line.strip()]
return set(ids)

def save_collected_id(item_id):
"""追加一个 ID 到文件"""
with open(HISTORY_FILE, "a", encoding="utf-8") as f:
f.write(f"{item_id}\\n")

# 流程开始时加载
已采集ID = load_collected_ids()
print(f"已加载 {len(已采集ID)} 条历史记录")

# 采集循环中
遍历商品列表:
商品ID = 获取元素文本()
if 商品ID in 已采集ID:
continue

采集商品数据()
已采集ID.add(商品ID)
save_collected_id(商品ID)

四、采集后去重——Pandas 一键清洗

import pandas as pd

df = pd.read_excel(r"D:\\采集数据_原始.xlsx")

# 打印去重前统计
print(f"去重前：{len(df)} 条")

# 按单个字段去重（保留第一条）
df = df.drop_duplicates(subset=["商品ID"], keep="first")

# 按多个字段组合去重
df = df.drop_duplicates(subset=["商品ID", "采集时间"], keep="last")

# 删除所有完全重复的行
df = df.drop_duplicates()

print(f"去重后：{len(df)} 条")

# 保存
df.to_excel(r"D:\\采集数据_去重后.xlsx", index=False)

keep 参数含义：

• "first" → 保留第一次出现的（推荐）
• "last" → 保留最后一次出现的
• False → 全删（一个不留，慎用）

五、增量更新——只采集新增的数据

全量采集效率低，特别是数据量大的时候。增量采集是新老数据对比，只取新增的。

方案A：基于时间戳

如果数据源有时间字段（上架时间、更新时间），只采集今天更新的。

import datetime

today = datetime.date.today().strftime("%Y-%m-%d")

# 在搜索或筛选中限定时间范围

选择下拉项("时间筛选", "今天")
# 或在 API 参数中加时间过滤

方案B：基于 ID 比对（最可靠）

# 1. 读取历史数据的所有 ID
历史数据 = pd.read_excel(r"D:\\采集历史.xlsx")
历史ID集 = set(历史数据["商品ID"].astype(str))

# 2. 采集新数据
新数据 = [] # 本次采集的结果

遍历商品列表:
商品ID = 获取元素文本()
if 商品ID in 历史ID集:
continue # 已存在，跳过

采集一条商品数据 → 新数据.append(结果)

# 3. 新数据追加到历史文件
if len(新数据) > 0:
新数据df = pd.DataFrame(新数据)
新数据df.to_excel(r"D:\\采集历史.xlsx",
mode="a", # 追加模式
header=False, # 不加标题行
index=False)

print(f"本次新增：{len(新数据)}条，累计：{len(历史ID集)+len(新数据)}条")

六、模糊去重——相似但不完全相同

TEMU店群如何管理运营？

标题差不多但有一个标点不同，怎么去重？

from difflib import SequenceMatcher

def is_similar(a, b, threshold=0.85):
"""判断两个字符串是否相似"""
return SequenceMatcher(None, a, b).ratio() > threshold

# 示例
is_similar("韩版气质连衣裙", "韩版 气质 连衣裙") # True
is_similar("韩版连衣裙", "男士运动短袖T恤") # False

模糊去重只适合数据量小的时候（几百条），多了性能很差。大量数据建议用 ID 精确匹配 + 人工抽查。

七、增量的完整流程

1. 加载历史数据 → 获取已有 ID 集合
2. 开始采集 → 每条检查 ID 是否已有
3. 已有 → 跳过；没有 → 采集并加入集合
4. 每采集 N 条（比如 50 条）→ 写一次文件（防中断丢数据）
5. 

6. 采集结束 → 统计新增数量 → 通知

# 增量采集的防中断策略
BUFFER_SIZE = 50 # 每 50 条写一次
buffer = []
已保存ID = load_collected_ids()

for item in items:
if item.id in 已保存ID:
continue

data = collect_one(item)
buffer.append(data)

if len(buffer) >= BUFFER_SIZE:
# 写入文件
pd.DataFrame(buffer).to_csv(r"D:\\增量数据.csv", mode="a", header=False, index=False)
# 更新 ID 文件
for d in buffer:
save_collected_id(d["商品ID"])
已保存ID.add(d["商品ID"])

print(f"已写入 {len(buffer)} 条，累计 {len(已保存ID)} 条")
buffer = []

# 最后一批不足 BUFFER_SIZE 的数据
if buffer:
pd.DataFrame(buffer).to_csv(r"D:\\增量数据.csv", mode="a", header=False, index=False)
for d in buffer:
save_collected_id(d["商品ID"])

八、常见坑

ID 不稳定：有些平台商品 ID 会变（编辑后重新生成），依赖 ID 去重可能失效。观察一下你采集的平台 ID 到底稳不稳。

去重后数据量对不上：drop_duplicates 不会告诉你具体删了哪些重复行。排查时建议在去重前导出原始数据的重复项统计。

增量 ID 文件过大：采集了 10 万条后，TXT 文件里有 10 万行 ID，加载会变慢。改用 set 存内存，文件只保留最近 N 天的，历史 ID 转存数据库。

mode="a" 追加 Excel 的坑：Pandas 追加 Excel 需要 openpyxl 引擎，且目标文件必须存在。如果文件不存在要先新建。

作者：林焱

本文为《影刀RPA学习手册》系列文章之一，内容源于实操经验的整理与分享。