OnCompleted( Action continuation, object?
keys()) writer.writeheader() writer.writerow(transformed) total_rows += 1 print(f"流式处理完成: {total_rows} 行") return total_rows def compose_pipeline(*fns): """组合多个变换函数为管道 用法: pipeline = compose_pipeline(filter_invalid, normalize_text, extract_features) 每个函数接收一个 dict，返回变换后的 dict 或 None（过滤） """ def pipeline(row: dict): result = row for fn in fns: result = fn(result) if result is None: return None return result return pipeline 四、数据处理优化的代价：精度损失、生态割裂与调试复杂度 Pandas 类型优化存在精度风险。
0.0.1", "DB_PORT": "54321", "DB_NAME": "业务库", "DB_USER": "ai_mcp", "DB_PASSWORD": "数据库密码" } } } } 保存文件，完全关闭Cursor后重新打开，编辑器自动拉起KES MCP方案并加载全部数据库工具。
表结构变更会自动执行；如果数据库账号没有建表权限，参考 migrate.sql 手动处理。