影刀RPA新手教程CSV文件处理完全指南——读取写入编码问题与大文件分块处理本文作者林焱 | 转载请注明出处开篇案例一次CSV乱码引发的线上事故去年帮一个电商客户做订单自动化处理。对方每天扔过来一个300MB的CSV文件包含全国2000家门店的销售数据。我写好流程第一次跑就炸了——中文全变成问号。当时我以为是文件本身的问题换了好几种打开方式Excel打开乱码Notepad打开正常。折腾两个小时才发现是编码问题。这个故事告诉我们CSV文件的水比你想的深得多。本文所有案例都围绕电商订单CSV每日处理这条真实业务线展开。模块一安装与准备工作处理CSV前先确认你的影刀RPA版本。建议使用3.0以上版本因为旧版的读取CSV指令对UTF-8 BOM的处理有bug。我当时用的是2.8版本每次读带BOM的UTF-8文件第一列的列名前面会多出一个不可见字符。这个字符在影刀的变量面板里看不到但用来做列匹配时永远对不上。升级到3.2之后这个问题就消失了。如果你还没安装影刀RPA可以去 home.linyan.cloud 找最新版的下载地址和安装视频。安装完成后新建一个空白流程命名为CSV订单处理Demo。模块二元素定位XPath CSS 正则你可能会问处理CSV文件为什么要学元素定位因为真实场景里CSV文件往往不是本地文件而是从网页后台导出的。你需要先用网页自动化登录后台点击导出按钮等待文件下载完成再进行CSV处理。这里就用到了元素定位。XPath实战定位导出按钮//button[contains(class,export) and text()导出订单]这个XPath的含义是找任意一个button标签它的class属性包含export且文字内容等于导出订单。CSS选择器实战定位下载链接a[href*download][href$.csv]{/* 匹配href包含download且以.csv结尾的a标签 */}在影刀里用CSS选择器定位元素时只写选择器部分不用写花括号里的内容。正则提取从下载链接里提取文件名importre urlhttps://admin.example.com/download/order_20240601.csv?tokenabc123patternrorder_\d{8}\.csvmatchre.search(pattern,url)ifmatch:filenamematch.group(0)print(f提取到文件名{filename})模块三变量与数据类型CSV处理的核心变量类型有三个字符串、列表、字典。字符串存文件路径、列名、单元格值。列表存CSV的每一行影刀里叫表格类型底层是列表的列表。字典存表头映射关系比如把订单号映射到列索引0。# 在影刀的Python代码指令里可以这样写headers[订单号,商品名称,数量,单价,买家留言]row_dict{订单号:DD20240601001,商品名称:无线耳机,数量:2,单价:199.00,买家留言:}# 把字典按表头顺序转成列表row_list[row_dict.get(h,)forhinheaders]print(row_list)# [DD20240601001, 无线耳机, 2, 199.00, ]我当时踩过一个坑直接用list(dict.values())把字典转列表。Python 3.7以上虽然保证插入顺序但如果你改动过字典顺序就可能乱掉。所以一定要按表头顺序显式取值。拼多多店群自动化上架方案模块四流程控制CSV处理最常用的流程控制是循环和条件判断。场景跳过空行和标题行# 伪代码逻辑对应影刀的循环指令forindex,rowinenumerate(csv_rows):ifindex0:continue# 跳过表头ifnotroworall(cellforcellinrow):continue# 跳过空行# 处理有效数据行场景根据某一列的值决定处理逻辑比如数量列小于1的行直接跳过并写一条日志quantityint(row[2])ifrow[2].isdigit()else0ifquantity1:print(f警告订单{row[0]}的数量为{quantity}已跳过)continue在影刀里这些逻辑用如果/否则指令配合循环指令来实现不需要写代码。但当你发现影刀的指令拼起来很复杂时直接用Python代码指令代替往往更清晰。模块五网页自动化CSV文件的来源很多CSV不是手动下载的是从网页后台导出的。以常见的电商后台为例完整的自动化流程是打开浏览器导航到后台登录页输入账号密码点击登录导航到订单管理页面设置日期筛选条件点击导出按钮等待文件下载完成等待文件下载完成的关键代码importosimporttime download_dirC:/Users/Administrator/Downloadstarget_fileos.path.join(download_dir,order_export.csv)# 等待文件出现最多等60秒foriinrange(60):ifos.path.exists(target_file):print(f文件下载完成耗时{i}秒)breaktime.sleep(1)else:raiseException(文件下载超时)我当时踩过这个坑没等文件写完就读取。CSV文件在下载过程中系统已经创建了文件但内容是空的或者只有一半。正确做法是检查文件大小是否稳定defwait_file_ready(filepath,timeout60):last_size-1foriinrange(timeout):ifnotos.path.exists(filepath):time.sleep(1)continuecurrent_sizeos.path.getsize(filepath)ifcurrent_sizelast_sizeandcurrent_size0:returnTruelast_sizecurrent_size time.sleep(1)returnFalse模块六数据处理——CSV读取影刀RPA自带读取CSV指令但我建议你掌握两种读取方式。方式一用影刀自带的指令在指令面板搜索读取CSV配置以下参数文件路径填CSV文件的完整路径编码默认是UTF-8如果乱码就换成GBK或GB18030分隔符默认是逗号Tab分隔的文件填\t是否包含表头勾选方式二用Python代码读取更灵活importcsvdefread_csv_safe(filepath,encodingNone): 自动检测编码并读取CSV ifencodingisNone:# 尝试常见编码forencin[utf-8-sig,utf-8,gbk,gb18030,big5]:try:withopen(filepath,r,encodingenc)asf:f.read(1024)# 读一小段测试encodingencbreakexceptUnicodeDecodeError:continueelse:raiseException(无法识别文件编码请手动指定)print(f使用编码{encoding})withopen(filepath,r,encodingencoding)asf:readercsv.reader(f)rowslist(reader)returnrows# 用法rowsread_csv_safe(C:/Downloads/order.csv)print(f共读取{len(rows)}行数据)编码问题的根本原理CSV文件本身不记录编码信息。UTF-8 BOM是指在文件开头加了三个字节EF BB BF。Windows的Excel打开无BOM的UTF-8 CSV时会默认用系统编码简体中文是GBK解释导致中文乱码。解决方案有两个保存时选UTF-8 with BOM用Excel的数据-从文本导入功能手动选UTF-8编码模块七数据处理——CSV写入写CSV比读CSV更容易出编码问题。用影刀指令写CSV在指令面板搜索写入CSV配置文件路径建议用.csv扩展名编码如果文件要给Excel打开选GBK如果给自己后续处理选UTF-8写入模式覆盖或追加用Python写CSV推荐importcsvdefwrite_csv_utf8_bom(filepath,headers,rows): 写入UTF-8 BOM格式的CSV兼容Excel直接打开 withopen(filepath,w,encodingutf-8-sig,newline)asf:writercsv.writer(f)writer.writerow(headers)forrowinrows:writer.writerow(row)print(f已写入{len(rows)}行到{filepath})# 用法headers[订单号,商品名称,数量,处理状态]rows[[DD001,无线耳机,2,已处理],[DD002,手机壳,1,待处理],]write_csv_utf8_bom(C:/Output/result.csv,headers,rows)注意那个newline参数这是Python csv模块的坑。如果不加这个参数在Windows上写的CSV每行之间会多一个空行。我当时踩过这个坑生成的CSV在Excel里打开每隔一行就有一个空行客户以为程序有bug。模块八大文件分块处理300MB的CSV用上面的方法一次性读入内存没问题。但如果是3GB的CSV呢影刀的读取CSV指令会把整个文件加载到内存文件太大时会导致内存溢出。解决方案是分块读取逐块处理。importcsvdefprocess_large_csv(input_path,output_path,chunk_size5000): 分块处理大CSV文件 chunk_size每块处理的行数根据内存调整 withopen(output_path,w,encodingutf-8-sig,newline)asout_f:writerNonewithopen(input_path,r,encodingutf-8-sig)asin_f:readercsv.reader(in_f)headersnext(reader)# 读取表头# 写入输出文件的表头writercsv.writer(out_f)writer.writerow(headers)chunk[]fori,rowinenumerate(reader):chunk.append(row)iflen(chunk)chunk_size:# 处理这一块processedprocess_chunk(chunk)writer.writerows(processed)print(f已处理{i1}行)chunk[]# 处理最后一块ifchunk:processedprocess_chunk(chunk)writer.writerows(processed)print(大文件处理完成)defprocess_chunk(chunk): 对每一块数据进行处理 这里以过滤掉数量为0的行为例 result[]forrowinchunk:try:quantityint(row[2])# 假设第3列是数量ifquantity0:result.append(row)except(ValueError,IndexError):pass# 跳过格式错误的行returnresult这篇文章的完整代码可以在 home.linyan.cloud 下载到。模块九鼠标键盘与图像操作CSV处理过程中有时会遇到网页导出CSV后没有直接下载链接而是打开了新标签页显示CSV内容的情况。这时你需要用鼠标全选内容CtrlC复制再粘贴到本地文件。全选复制的实现在影刀里用以下指令组合发送快捷键 CtrlA全选发送快捷键 CtrlC复制用获取剪贴板文本指令拿到内容用Python代码把文本内容按CSV格式解析并保存importpyperclip# 获取剪贴板内容textpyperclip.paste()# 按行分割linestext.split(\n)rows[line.split(,)forlineinlinesifline.strip()]print(f从剪贴板读取到{len(rows)}行)如果网页里的CSV内容是用Tab分隔的把split(,)改成split(\t)。模块十进阶技能技能一自动检测CSV的分隔符有些CSV用逗号有些用分号有些用Tab。用Python的csv.Sniffer可以自动检测importcsvdefdetect_delimiter(filepath):withopen(filepath,r,encodingutf-8-sig)asf:samplef.read(4096)dialectcsv.Sniffer().sniff(sample)returndialect.delimiter# 用法delimdetect_delimiter(unknown.csv)print(f检测到的分隔符{repr(delim)})# 逗号返回 ,Tab返回 \t分号返回 ;技能二处理带引号的CSV字段CSV标准规定字段内容包含逗号时用双引号把整个字段包起来。比如张三,经理,技术部,5000影刀的读取CSV指令默认能处理这种情况。但如果你自己解析CSV千万不要用split(,),那样会把张三,经理拆成两个字段。一定要用csv.reader它会正确处理引号和转义。技能三合并多个CSV文件importcsvimportglobdefmerge_csv_files(file_pattern,output_path): 合并多个CSV文件假设表头相同 filesglob.glob(file_pattern)print(f找到{len(files)}个CSV文件)withopen(output_path,w,encodingutf-8-sig,newline)asout_f:writercsv.writer(out_f)wrote_headerFalseforfilepathinfiles:withopen(filepath,r,encodingutf-8-sig)asin_f:readercsv.reader(in_f)headersnext(reader)ifnotwrote_header:writer.writerow(headers)wrote_headerTrueforrowinreader:writer.writerow(row)print(f已合并{filepath})print(f合并完成结果保存至{output_path})模块十一平台实战把CSV处理流程部署到影刀控制台调度中心时有几个实战要点。要点一文件路径用相对路径或配置变量不要把文件路径写死在流程里。在影刀的变量面板创建一个字符串变量csv_input_dir存输入目录。流程里所有涉及文件路径的地方都用这个变量拼接。这样换环境时只需要改一个地方。TEMU店群如何管理运营要点二处理完的文件自动归档importshutilimportosfromdatetimeimportdatetimedefarchive_processed_file(filepath): 把处理完的CSV移到archive目录用日期重命名 archive_dirC:/CSV_Archiveos.makedirs(archive_dir,exist_okTrue)todaydatetime.now().strftime(%Y%m%d)filenameos.path.basename(filepath)name,extos.path.splitext(filename)new_namef{name}_{today}{ext}new_pathos.path.join(archive_dir,new_name)shutil.move(filepath,new_path)print(f文件已归档{new_path})要点三用任务监控查看执行状态在影刀控制台的任务监控页面可以看到每个机器人执行CSV处理任务的状态。如果某次执行异常控制台会记录错误日志包括哪一行数据出了问题。我当时有一次处理300万行数据跑了40分钟最后报了一个索引超出范围的错误。去控制台看日志发现是第2,847,391行缺少了单价字段。如果没有控制台的日志我根本不知道是哪个文件哪一行出了问题。模块十二系统联动与工程化规范工程化规范一CSV处理流程的标准结构流程开始 ├─ 初始化定义变量、创建输出目录 ├─ 输入校验检查CSV文件是否存在、编码是否正确 ├─ 数据处理读取、清洗、转换、写入 ├─ 输出校验检查输出文件行数是否与输入一致 ├─ 归档把处理完的文件移到archive目录 └─ 流程结束成功/失败工程化规范二用配置表管理多个CSV任务如果你的机器人需要处理多种类型的CSV订单、库存、物流……不要写多个流程。写一个通用流程用配置表决定每次怎么处理# config.csv 内容示例# task_name,input_file,output_file,encoding,delimiter,chunk_size# 订单处理,order.csv,order_result.csv,utf-8-sig,,,5000# 库存同步,stock.txt,stock_result.csv,gbk,\t,10000importcsvdefload_task_config(config_path,task_name):withopen(config_path,r,encodingutf-8-sig)asf:readercsv.DictReader(f)forrowinreader:ifrow[task_name]task_name:returnrowraiseException(f找不到任务配置{task_name})工程化规范三错误行单独导出处理CSV时总会遇到几行格式不对的数据。不要直接跳过把它们写到另一个文件里方便后续人工核查error_rows[]# 在处理循环中try:processedprocess_row(row)result.append(processed)exceptExceptionase:error_rows.append(row[str(e)])# 把错误信息附加到行尾# 处理完成后导出错误行iferror_rows:withopen(errors.csv,w,encodingutf-8-sig,newline)asf:writercsv.writer(f)writer.writerow(headers[错误原因])writer.writerows(error_rows)print(f有{len(error_rows)}行处理失败已导出到errors.csv)速查表CSV处理常见问题问题原因解决方案Excel打开中文乱码缺少BOM用utf-8-sig编码写入读出来第一列名不对UTF-8 BOM被当成了列名的一部分用utf-8-sig编码读取写入的CSV有空行没加newline“”open时加newline参数包含逗号的字段被拆开用split分割而不是csv.reader改用csv.reader大文件内存溢出一次性读入内存用分块读取方案下载的文件内容不完整没等文件写完就读取用wait_file_ready函数报错排查指南报错UnicodeDecodeError: ‘utf-8’ codec can’t decode byte原因文件不是UTF-8编码。解决用GBK或GB18030编码重新打开。报错csv.Error: line contains NULL byte原因CSV文件里包含了NULL字符\x00。解决读取前先清洗文件把NULL字符替换掉withopen(filepath,rb)asf:contentf.read().replace(b\x00,b)withopen(filepath.clean,wb)asf:f.write(content)报错IndexError: list index out of range原因某一行的列数不够。解决处理每一行前先检查列数不够的补空字符串expected_columns5whilelen(row)expected_columns:row.append()总结CSV文件处理是影刀RPA最基础也最常用的技能之一。核心要点就三个编码要对、大文件要分块、错误数据要单独导出。把这三个做好了你的CSV处理流程就能稳定跑在生产环境。更多影刀RPA实战案例欢迎访问 home.linyan.cloud 交流讨论。#影刀RPA #RPA教程 #CSV处理 #数据自动化 #影刀新手教程作者林焱