第5篇：评论数据的工程化：CSV 解析、字段归一化、脱敏与指纹幂等去重

本系列博客基于一个真实可运行的电商评论舆情分析项目。上一篇《第4篇LangGraph 实战把 5 个 Agent 阶段编排成一个可观测 Pipeline》一、痛点开场模型分析之前数据质量决定一切。电商评论来自不同平台每家的导出格式都不一样列名评论内容、内容、comment、content……叫什么的都有编码有的 UTF-8 带 BOM有的是 Windows 的 GBK评分数字、5分、好评、★★★★★重复同一批数据被运营导了三次每次导入都会重复入库如果不处理这些问题模型再强也是垃圾进垃圾出。这篇讲数据接入层的工程细节解析 → 归一化 → 脱敏 → 指纹去重 → 幂等落库。二、阶段一解析 CSV/Excel编码自动探测CSV 依次尝试utf-8-sig带 BOM、gbkWindows 中文、utf-8def_read_csv(path)-tuple[list[dict],str]:encodings[utf-8-sig,gbk,utf-8]last_errNoneforencinencodings:try:withopen(path,r,encodingenc,newline)asf:readercsv.DictReader(f)ifnotreader.fieldnames:raiseValueError(CSV 无表头)returnlist(reader),encexcept(UnicodeDecodeError,ValueError)ase:last_erreraiseValueError(fCSV 解析失败尝试 utf-8/gbk 均失败:{last_err})Excel 读取用openpyxl只读模式首行当表头跳过全空行def_read_excel(path)-list[dict]:wbopenpyxl.load_workbook(path,read_onlyTrue,data_onlyTrue)wswb.active headersnext(ws.iter_rows(values_onlyTrue))rows[dict(zip(headers,r))forrinws.iter_rows(values_onlyTrue)ifrisnotNoneandany(cisnotNoneandstr(c).strip()forcinr)]wb.close()returnrows三、阶段二字段归一化列别名映射_COLUMN_ALIASES把五花八门的表头映射到 canonical 字段_COLUMN_ALIASES{评论内容:content,内容:content,评论:content,正文:content,评价内容:content,comment:content,content:content,商品:product_name,产品:product_name,product_name:product_name,平台:platform_code,platform:platform_code,...用户名:user_name,昵称:user_name,username:user_name,...评分:user_rating,星级:user_rating,rating:user_rating,...评论时间:review_time,时间:review_time,time:review_time,...}匹配前先归一化表头去空白、转小写def_normalize_header(h)-str:return_WS_RE.sub(,str(hor)).strip().lower()def_match_column(header)-Optional[str]:norm_normalize_header(header)return_COLUMN_ALIASES.get(norm)or_COLUMN_ALIASES.get(norm.lower())评分容错支持数字、5分、好评/中评/差评最后 clamp 到 0~5def_parse_rating(value)-Optional[float]:# 5 → 5.0 | 5分 → 5.0 | 好评 → 5.0 | 中评 → 3.0 | 差评 → 1.0# 解析失败返回 None行计入错误统计...returnmax(0.0,min(5.0,rating))时间容错支持十几种格式 时间戳_TIME_FORMATS[%Y-%m-%d %H:%M:%S,%Y-%m-%d %H:%M,%Y-%m-%d,%Y/%m/%d %H:%M:%S,%Y/%m/%d,%Y年%m月%d日,%Y.%m.%d,...]四、阶段三用户名脱敏隐私是硬约束_mask_name只保留首字符def_mask_name(name)-str:ifnotname:returnreturnstr(name)[0]***生产还需要对手机号、邮箱、地址做 PII 检测与脱敏原始文件设访问权限和保留期限。五、阶段四指纹去重核心指纹怎么算def_build_fingerprint(tenant_id,platform_code,product_name,content)-str:key|.join([str(tenant_id),str(platform_codeor),str(product_nameor),_normalize_text(content),# 去空白保证格式差异不影响幂等])returnhashlib.sha256(key.encode(utf-8)).hexdigest()为什么要包含这四个维度租户不同品牌之间不冲突平台 商品同文案跨商品不误判正文内容本体两层去重# 第一层内存 set去当前批次内重复减少无效写入seenset()forrowinvalid_rows:fp_build_fingerprint(...)iffpinseen:duplicate_count1continueseen.add(fp)# 第二层数据库唯一约束跨批次/并发/重启后的最终保障# 建表# UNIQUE (tenant_id, fingerprint)# 插入# INSERT ... ON CONFLICT (tenant_id, fingerprint) DO NOTHING真正的幂等不能只靠 Python set必须依赖数据库唯一约束。因为跨批次重复第一批导过第二批再来并发请求两个请求同时导入进程重启内存 set 清零六、阶段五幂等落库asyncdefsave_batch_node(state)-dict:# 1. 解析/创建 platform、product 维度# 2. 创建 import_batches 批次记录# 3. 批量 INSERT ... ON CONFLICT DO NOTHING# 4. 回写批次统计inserted / duplicate / failed批次统计从落库事实汇总而不是只信内存计数。七、坏行处理行级隔离一行数据错误不应该让整批失败# 缺正文 → 该行计入 failed_rows记录原因# 评分解析失败 → 评分置空行照常入库# 文件损坏 / 无表头 → 批次级错误整批失败当前实现只保留前 20 条解析错误生产应提供错误报告文件供用户下载修正。八、踩坑与边界评分/时间解析失败要保留统计不能静默丢弃否则运营不知道数据质量有问题指纹误判同一评论被用户修改、同文案不同人发都可能误判重复有外部review_id时应优先用业务唯一键大文件当前整文件读入内存生产要流式解析 分片落库ON CONFLICT DO NOTHING的统计口径要区分当前批次内重复和历史已存在分开统计更清晰九、总结解析兼容编码、格式、表头全兜住归一化别名映射 评分/时间容错脱敏隐私前置处理幂等内存预去重 数据库唯一约束双保险可观测批次统计反映真实落库结果下一篇预告《FastAPI 长任务异步化实践从HTTP 阻塞到进程内 TaskManager》