影刀RPA 数据对比分析：两表差异查找
作者：林焱
什么情况用什么
本月数据和上月数据对比找变化、实际数据和计划数据对比找差异、两份名单对比找出新增和删除的人。在影刀RPA里用pandas的merge和compare可以快速找出两个数据集之间的差异。
适用场景：月度数据对比、库存盘点对比、人员变动检测、价格变动监控、数据一致性校验。
怎么做
店群矩阵自动化突破运营极限！
两表对比找差异
importpandasaspd# 读取两个表df_old=pd.read_excel(r"C:\Data\june_data.xlsx")df_new=pd.read_excel(r"C:\Data\july_data.xlsx")# 方法1：merge + indicatormerged=df_new.merge(df_old,on='订单ID',how='outer',indicator=True# 添加_merge列标识来源)# 新增的行（在new中有，old中没有）added=merged[merged['_merge']=='left_only']# 删除的行（在old中有，new中没有）removed=merged[merged['_merge']=='right_only']# 未变化的行unchanged=merged[merged['_merge']=='both']值变化检测
deffind_value_changes(df_old,df_new,key_col,compare_cols=None):""" 找出值发生变化的行 key_col: 关联键列名 compare_cols: 要比较的列名列表（None则比较所有列） """# 合并两表merged=df_new.merge(df_old,on=key_col,how='inner',suffixes=('_新','_旧'))ifcompare_colsisNone:compare_cols=[cforcindf_new.columnsifc!=key_col]# 找出有变化的行changed_mask=pd.Series(False,index=merged.index)changes=[]forcolincompare_cols:old_col=f'{col}_旧'new_col=f'{col}_新'ifold_colinmerged.columnsandnew_colinmerged.columns:# 比较值（处理NaN）col_changed=((merged[old_col].fillna('NULL')!=merged[new_col].fillna('NULL')))changed_mask=changed_mask|col_changed changed_rows=merged[changed_mask].copy()# 记录具体变化foridx,rowinchanged_rows.iterrows():forcolincompare_cols:old_val=row.get(f'{col}_旧')new_val=row.get(f'{col}_新')ifstr(old_val)!=str(new_val):changes.append({'关联键':row[key_col],'字段':col,'旧值':old_val,'新值':new_val})returnpd.DataFrame(changes),changed_rows# 使用changes,changed_rows=find_value_changes(df_old,df_new,key_col='订单ID',compare_cols=['金额','状态','产品名称'])ifnotchanges.empty:changes.to_excel(r"C:\Data\changes.xlsx",index=False)print(f"发现{len(changes)}处变化")完整对比报告
defgenerate_comparison_report(df_old,df_new,key_col,output_path):"""生成完整对比报告"""withpd.ExcelWriter(output_path,engine='openpyxl')aswriter:# 1. 概览summary=pd.DataFrame({'项目':['旧数据行数','新数据行数','新增行数','删除行数','变化行数','未变化行数'],'数量':[len(df_old),len(df_new),len(df_new[~df_new[key_col].isin(df_old[key_col])]),len(df_old[~df_old[key_col].isin(df_new[key_col])]),0,# 待计算0# 待计算]})# 2. 新增的记录added=df_new[~df_new[key_col].isin(df_old[key_col])]added.to_excel(writer,sheet_name='新增记录',index=False)# 3. 删除的记录removed=df_old[~df_old[key_col].isin(df_new[key_col])]removed.to_excel(writer,sheet_name='删除记录',index=False)# 4. 值变化changes,changed_rows=find_value_changes(df_old,df_new,key_col)changes.to_excel(writer,sheet_name='值变化',index=False)# 更新概览summary.loc[summary['项目']=='变化行数','数量']=len(changed_rows)summary.loc[summary['项目']=='未变化行数','数量']=(len(df_old)-len(removed)-len(changed_rows))summary.to_excel(writer,sheet_name='对比概览',index=False)returnoutput_path# 使用generate_comparison_report(df_old,df_new,key_col='订单ID',output_path=r"C:\Data\comparison_report.xlsx")价格变动监控
defmonitor_price_change(old_file,new_file,output_file):"""监控价格变动"""df_old=pd.read_excel(old_file)df_new=pd.read_excel(new_file)# 合并对比merged=df_new.merge(df_old[['SKU','价格']],on='SKU',how='left',suffixes=('_新','_旧'))# 计算变化merged['价格变化']=merged['价格_新']-merged['价格_旧']merged['变化幅度']=(merged['价格变化']/merged['价格_旧']*100).round(2)merged['变化方向']=merged['价格变化'].apply(lambdax:'涨价'ifx>0else'降价'ifx<0else'不变')# 筛选有变化的changed=merged[merged['价格变化']!=0].copy()changed=changed.sort_values('变化幅度',ascending=False)changed.to_excel(output_file,index=False)# 统计print(f"涨价:{len(changed[changed['变化方向']=='涨价'])}个")print(f"降价:{len(changed[changed['变化方向']=='降价'])}个")print(f"最大涨幅:{changed['变化幅度'].max():.1f}%")print(f"最大降幅:{changed['变化幅度'].min():.1f}%")returnchanged影刀RPA对比流程
【设置变量】 old_file = r"C:\Data\上月数据.xlsx" new_file = r"C:\Data\本月数据.xlsx" 【读取Excel文件】→ old_file → df_old 【读取Excel文件】→ new_file → df_new 【执行Python代码】 # 对比分析 report = generate_comparison_report( df_old, df_new, key_col='订单ID', output_path=r"C:\Data\对比报告.xlsx" ) 【发送邮件】 附件：对比报告.xlsx 内容：本月数据变化{新增X条，删除Y条，修改Z处}有什么坑
temu店群自动化报活动案例
坑1：关联键类型不一致
# 问题：一边是int一边是str，merge匹配不上df_old['订单ID']=df_old['订单ID'].astype(int)df_new['订单ID']=df_new['订单ID'].astype(str)merged=df_old.merge(df_new,on='订单ID')# 结果为空# 解决：统一类型df_old['订单ID']=df_old['订单ID'].astype(str).str.strip()df_new['订单ID']=df_new['订单ID'].astype(str).str.strip()坑2：NaN比较结果意外
# 问题：NaN != NaN 返回True，导致没变化的数据也被标记为变化importnumpyasnp np.nan==np.nan# False!np.nan!=np.nan# True!# 解决：用fillna统一处理old_val=str(row[f'{col}_旧'])ifpd.notna(row[f'{col}_旧'])else'NULL'new_val=str(row[f'{col}_新'])ifpd.notna(row[f'{col}_新'])else'NULL'changed=old_val!=new_val坑3：列名冲突
# 问题：两表有同名列，merge后自动加后缀_x _ymerged=df_new.merge(df_old,on='ID')# 金额列变成 金额_x 和 金额_y# 解决：指定后缀merged=df_new.merge(df_old,on='ID',suffixes=('_新','_旧'))# 金额_新 和 金额_旧，更清晰坑4：重复键导致行数膨胀
# 问题：df_old中某个ID有2条记录，merge后变成2×N条merged=df_new.merge(df_old,on='ID')# 行数暴增# 解决：先去重df_old_unique=df_old.drop_duplicates(subset='ID',keep='last')merged=df_new.merge(df_old_unique,on='ID',how='left')坑5：大数据量对比内存不够
# 解决：分批对比defbatch_compare(df_old,df_new,key_col,batch_size=10000):"""分批对比"""all_changes=[]foriinrange(0,len(df_new),batch_size):batch=df_new.iloc[i:i+batch_size]changes,_=find_value_changes(df_old,batch,key_col)all_changes.append(changes)print(f"已处理{min(i+batch_size,len(df_new))}/{len(df_new)}")returnpd.concat(all_changes,ignore_index=True)