影刀RPA 数据采集的完整性校验：怎么知道数据没丢

影刀RPA 数据采集的完整性校验怎么知道数据没丢作者林焱什么情况用这个你写了一个电商平台商品数据采集流程显示共采集5000条。看起来很美好——直到运营同学跑来问“怎么少了200个商品竞品有5200个SKU。”你不知道是采集漏了、还是网站本身就只有5000条、还是翻页中间丢了几页。没有完整性校验的数据采集采集量的数字毫无意义——你不知道它到底是全量还是丢了一半。这篇文章教你为采集流程加上完整性校验机制让每一批数据都有证据链可以回溯。怎么做第一步采集前记录预期采集之前先搞清楚应该采到多少拼多多店群自动化报活动上架defrecord_expectation(driver):在采集开始前记录页面显示的预期总量expectations{page_url:driver.current_url,page_title:driver.title,timestamp:datetime.now().isoformat(),}# 尝试获取页面上的总数信息try:total_textdriver.find_element(By.CSS_SELECTOR,.total-count).text expectations[displayed_total]int(re.findall(r\d,total_text)[0])except:expectations[displayed_total]Nonetry:total_pagesdriver.find_element(By.CSS_SELECTOR,.total-pages).text expectations[displayed_pages]int(re.findall(r\d,total_pages)[0])except:expectations[displayed_pages]Nonereturnexpectations# 在采集流程开始时调用expectedrecord_expectation(driver)print(f页面显示总量:{expected.get(displayed_total,未知)})第二步采集过程中逐页校验classCollectionValidator:采集完整性校验器def__init__(self,expected_totalNone):self.expected_totalexpected_total self.page_records[]# 每页的采集记录self.collected_items[]deflog_page(self,page_num,items_on_page,url):记录每页采集的数据record{page:page_num,count:len(items_on_page),url:url,timestamp:datetime.now().isoformat(),first_item:items_on_page[0]ifitems_on_pageelseNone,last_item:items_on_page[-1]ifitems_on_pageelseNone,}self.page_records.append(record)self.collected_items.extend(items_on_page)defcheck_anomalies(self):检查异常anomalies[]# 1. 检查是否有空页forrecordinself.page_records:ifrecord[count]0:anomalies.append(f第{record[page]}页采集为空)# 2. 检查是否有页数量异常少iflen(self.page_records)2:avg_countsum(r[count]forrinself.page_records)/len(self.page_records)forrecordinself.page_records:ifrecord[count]avg_count*0.5:# 少于平均的50%anomalies.append(f第{record[page]}页数据偏少:{record[count]}条 (平均{avg_count:.0f}条))# 3. 检查总数与页面显示是否一致ifself.expected_total:actual_totallen(self.collected_items)ifactual_total!self.expected_total:diffself.expected_total-actual_total anomalies.append(f总数不匹配: 预期{self.expected_total}条实际{actual_total}条相差{diff}条)# 4. 检查是否有重复数据dup_countself._check_duplicates()ifdup_count0:anomalies.append(f发现{dup_count}条重复数据)returnanomaliesdef_check_duplicates(self,key_fieldid):检查重复基于关键字段seenset()duplicates0foriteminself.collected_items:keyitem.get(key_field)ifkeyinseen:duplicates1seen.add(key)returnduplicatesdefgenerate_report(self):生成校验报告anomaliesself.check_anomalies()report{采集时间:datetime.now().isoformat(),总页数:len(self.page_records),总条数:len(self.collected_items),预期条数:self.expected_total,每页详情:self.page_records,异常:anomalies,状态:通过ifnotanomalieselse异常}# 打印报告print(f\n{*40})print(f采集校验报告)print(f{*40})print(f总条数:{report[总条数]})ifreport[预期条数]:print(f预期条数:{report[预期条数]})diffreport[预期条数]-report[总条数]ifdiff!0:print(f差异:{diff}条 (采集了{report[总条数]/report[预期条数]*100:.1f}%))print(f页数:{report[总页数]})ifanomalies:print(f\n⚠️ 发现{len(anomalies)}个异常:)forainanomalies:print(f -{a})else:print(\n✓ 校验通过)returnreport第三步API采集的完整性校验API采集比网页采集容易做校验——API通常返回总数defcollect_with_integrity_check(api_url,headers):带完整性校验的API采集all_data[]page1expected_totalNonewhileTrue:resprequests.get(api_url,params{page:page,size:50},headersheaders,timeout10)ifresp.status_code!200:print(f第{page}页请求失败:{resp.status_code})breakdataresp.json()# 第一页获取预期总数ifpage1:expected_totaldata.get(total)ordata.get(totalCount)print(f预期总数:{expected_total})itemsdata.get(data,{}).get(list,[])ordata.get(list,[])all_data.extend(items)# 检查是否超出预期ifexpected_totalandlen(all_data)expected_total:print(f⚠️ 采集量({len(all_data)})超出预期({expected_total})可能重复)# 翻页iflen(items)50:breakpage1# 最终校验actuallen(all_data)ifexpected_totalandactual!expected_total:print(f⚠️ 数量不匹配: 预期{expected_total}, 实际{actual}, 差异{expected_total-actual})else:print(f✓ 校验通过:{actual}条)returnall_data第四步关键字段的完整性检查数量对了不代表数据质量对了。逐字段检查defcheck_field_completeness(data,required_fields):检查关键字段的完整性totallen(data)field_stats{}forfieldinrequired_fields:empty_countsum(1foritemindataifnotitem.get(field))empty_rateempty_count/totaliftotal0else0field_stats[field]{总数:total,为空数:empty_count,完整率:f{(1-empty_rate)*100:.1f}%,状态:✓ifempty_rate0.05else⚠️ifempty_rate0.2else✗}# 输出字段完整性报告print(f\n字段完整性检查 (共{total}条):)forfield,statsinfield_stats.items():print(f{stats[状态]}{field}:{stats[完整率]}({stats[为空数]}/{stats[总数]}为空))returnfield_stats# 使用required[商品名称,价格,库存,SKU编码]field_statscheck_field_completeness(collected_data,required)第五步采集日志——采集过程的完整记录defgenerate_collection_log(validator):生成完整的采集日志——出问题时回溯的证据log{开始时间:expected.get(timestamp),页面显示总数:expected.get(displayed_total),页面显示页数:expected.get(displayed_pages),实际采集页数:len(validator.page_records),实际采集条数:len(validator.collected_items),每页详情:[],数据完整性:{},异常:[]}forpage_recordinvalidator.page_records:log[每页详情].append({页码:page_record[page],条数:page_record[count],时间:page_record[timestamp],首条:str(page_record[first_item])[:50],末条:str(page_record[last_item])[:50],})anomaliesvalidator.check_anomalies()field_completenesscheck_field_completeness(validator.collected_items,[商品名称,价格])log[异常]anomalies log[字段完整性]field_completeness# 保存日志log_pathfD:/rpa_logs/collection_{datetime.now().strftime(%Y%m%d_%H%M%S)}.jsonwithopen(log_path,w,encodingutf-8)asf:json.dump(log,f,ensure_asciiFalse,indent2)returnlog有什么坑坑一页面显示的总数和实际数据量不一致现象页面显示共1000条但实际只采集到980条。原因有的是页面统计包含了下线/删除的条目有的是翻页时数据变化了。解决页面显示的总数只能作为参考值不能作为绝对值。一致性差距超过5%需要人工确认。坑二重复采集了相同数据TEMU店群矩阵自动化运营核价报活动现象翻页时第3页和第4页有重复数据分页边界问题。解决采集后按唯一键去重去重前后数量差异超过1%要标记异常。坑三采集过程中数据变化现象采集到第50页时前几页的数据已经变了商品被下架、价格被修改。解决这种情况无法完全避免。但可以通过首尾数据一致性检查来发现采集完最后一页后回到第一页确认第一页的第一条数据是否和开始时一致。坑四只检查数量不检查内容现象5000条数据数量是对的但其中500条的商品名称字段全是空的。解决校验要包含数量校验字段完整性校验两者缺一不可。坑五校验太慢拖累采集速度现象采集10秒校验跑了30秒。解决校验和采集分开——采集时做基本校验数量采集完后做深度校验去重、字段完整性。总结数据完整性校验不需要写得完美才上线。从最基本的做起采集完成后对比页面显示的总数和你实际采集到数量差距超过5%就告警。这一行代码就能帮你发现80%的数据丢失问题。