影刀RPA 文本文件的读写:TXT、CSV、JSON的读取与写入
影刀RPA 文本文件的读写:TXT、CSV、JSON的读取与写入
作者:林焱
Excel不是唯一的数据存储格式。很多场景下你需要处理文本文件——日志文件是TXT、导出的数据是CSV、API返回的数据是JSON。影刀RPA在处理这些文本文件时,主要依赖Python节点的文件操作能力。
这篇文章把TXT、CSV、JSON三种最常见文本格式的读写方法讲全。
一、TXT文件读写
1.1 读取TXT文件
# 读取整个文件withopen("D:\\data\\log.txt","r",encoding="utf-8")asf:content=f.read()set_variable("file_content",content)# 逐行读取withopen("D:\\data\\log.txt","r",encoding="utf-8")asf:lines=f.readlines()set_variable("file_lines",lines)# 逐行处理(大文件推荐)result=[]withopen("D:\\data\\log.txt","r",encoding="utf-8")asf:forlineinf:line=line.strip()# 去掉行尾换行符ifline:result.append(line)set_variable("processed_lines",result)坑1:编码问题
文件编码可能是UTF-8、GBK、GB2312等。用错误的编码打开会乱码或报错。
# 自动检测编码importchardetwithopen("D:\\data\\log.txt","rb")asf:raw=f.read()encoding=chardet.detect(raw)["encoding"]withopen("D:\\data\\log.txt","r",encoding=encoding)asf:content=f.read()坑2:BOM头
UTF-8 with BOM的文件开头有一个\ufeff字符,读取后内容开头会多一个不可见字符。用utf-8-sig编码打开自动去掉BOM:
withopen("D:\\data\\log.txt","r",encoding="utf-8-sig")asf:content=f.read()1.2 写入TXT文件
# 覆盖写入withopen("D:\\data\\output.txt","w",encoding="utf-8")asf:f.write("第一行内容\n")f.write("第二行内容\n")# 追加写入withopen("D:\\data\\log.txt","a",encoding="utf-8")asf:f.write("新追加的内容\n")# 写入多行lines=["第一行","第二行","第三行"]withopen("D:\\data\\output.txt","w",encoding="utf-8")asf:f.write("\n".join(lines))坑:写入后没换行
f.write("内容")不会自动加换行符,需要手动加\n。如果用print函数写文件会自动加换行:
withopen("D:\\data\\output.txt","w",encoding="utf-8")asf:print("第一行",file=f)print("第二行",file=f)1.3 TXT文件处理实战
拼多多店群自动化上架方案
# 日志文件解析importrewithopen("D:\\logs\\app.log","r",encoding="utf-8")asf:logs=f.readlines()# 提取ERROR级别的日志errors=[]forlineinlogs:if"ERROR"inline:# 提取时间和错误信息match=re.match(r'\[(.*?)\] ERROR (.*)',line)ifmatch:errors.append({"time":match.group(1),"message":match.group(2)})set_variable("error_logs",errors)二、CSV文件读写
2.1 用csv模块读取
importcsv# 读取CSV文件withopen("D:\\data\\products.csv","r",encoding="utf-8-sig")asf:reader=csv.reader(f)headers=next(reader)# 第一行是表头data=[rowforrowinreader]# data是一个二维列表# data[0] = ["商品A", "99.9", "100"]set_variable("csv_headers",headers)set_variable("csv_data",data)2.2 用DictReader读取(推荐)
importcsvwithopen("D:\\data\\products.csv","r",encoding="utf-8-sig")asf:reader=csv.DictReader(f)data=[dict(row)forrowinreader]# data是一个字典列表# data[0] = {"商品名": "商品A", "价格": "99.9", "库存": "100"}set_variable("csv_data",data)2.3 用pandas读取CSV(最推荐)
importpandasaspd# 基本读取df=pd.read_csv("D:\\data\\products.csv",encoding="utf-8-sig")# 指定分隔符(TSV文件用制表符)df=pd.read_csv("D:\\data\\products.tsv",sep="\t")# 跳过前几行df=pd.read_csv("D:\\data\\products.csv",skiprows=2)# 指定列名df=pd.read_csv("D:\\data\\products.csv",names=["名称","价格","库存"],header=None)# 指定某列为索引df=pd.read_csv("D:\\data\\products.csv",index_col="商品ID")# 数据类型转换df=pd.read_csv("D:\\data\\products.csv",dtype={"价格":float,"库存":int})set_variable("csv_data",df.to_dict("records"))2.4 写入CSV
importcsv data=get_variable("output_data")headers=["商品名","价格","库存"]withopen("D:\\data\\output.csv","w",encoding="utf-8-sig",newline="")asf:writer=csv.DictWriter(f,fieldnames=headers)writer.writeheader()writer.writerows(data)坑1:newline参数
Windows下写CSV必须加newline="",否则每行之间会多一个空行。
坑2:逗号在数据中
CSV用逗号分隔字段,如果数据本身包含逗号(如"商品A,蓝色"),需要用引号包裹。csv模块自动处理:
# csv.writer会自动给包含逗号的数据加引号writer.writerow(["商品A,蓝色","99.9","100"])# 输出: "商品A,蓝色",99.9,100坑3:CSV编码问题
Excel打开UTF-8编码的CSV文件会乱码。用utf-8-sig编码写入(带BOM头),Excel就能正确识别:
withopen("D:\\data\\output.csv","w",encoding="utf-8-sig",newline="")asf:2.5 CSV与Excel互转
importpandasaspd# CSV转Exceldf=pd.read_csv("D:\\data\\products.csv",encoding="utf-8-sig")df.to_excel("D:\\data\\products.xlsx",index=False)# Excel转CSVdf=pd.read_excel("D:\\data\\products.xlsx")df.to_csv("D:\\data\\products.csv",index=False,encoding="utf-8-sig")三、JSON文件读写
3.1 读取JSON文件
importjsonwithopen("D:\\data\\config.json","r",encoding="utf-8")asf:config=json.load(f)# config是一个字典# {"server": "192.168.1.1", "port": 8080, "debug": true}set_variable("config",config)3.2 写入JSON文件
importjson data=get_variable("output_data")withopen("D:\\data\\output.json","w",encoding="utf-8")asf:json.dump(data,f,ensure_ascii=False,indent=2)坑1:ensure_ascii参数
json.dump默认ensure_ascii=True,中文字符会被转义成\uXXXX。设置ensure_ascii=False保留中文:
# ensure_ascii=True(默认): {"name": "\u5546\u54c1A"}# ensure_ascii=False: {"name": "商品A"}json.dump(data,f,ensure_ascii=False,indent=2)坑2:indent参数
indent=2让JSON输出有缩进,方便人阅读。不设置indent则输出紧凑格式(文件更小但不可读)。
3.3 JSON与Python对象的转换
importjson# Python字典 → JSON字符串data={"name":"商品A","price":99.9,"tags":["热销","推荐"]}json_str=json.dumps(data,ensure_ascii=False)# JSON字符串 → Python字典parsed=json.loads(json_str)# 嵌套JSON处理nested={"product":{"name":"商品A","variants":[{"color":"红","stock":10},{"color":"蓝","stock":20}]}}# 访问嵌套数据forvariantinnested["product"]["variants"]:print(f"{variant['color']}:{variant['stock']}")3.4 JSON作为配置文件
importjson# 读取配置withopen("D:\\config\\settings.json","r",encoding="utf-8")asf:config=json.load(f)# 使用配置server=config.get("server","localhost")# 默认值localhostport=config.get("port",8080)timeout=config.get("timeout",30)retry_count=config.get("retry",3)# 修改配置并保存config["last_run"]="2026-07-01"withopen("D:\\config\\settings.json","w",encoding="utf-8")asf:json.dump(config,f,ensure_ascii=False,indent=2)坑:JSON不支持注释
标准JSON格式不支持注释。如果配置文件需要注释,用JSON5格式或YAML,或者在配置字段里加_comment字段:
{"_comment":"这是服务器配置","server":"192.168.1.1","port":8080}四、格式选择指南
TEMU店群如何管理运营?
| 格式 | 适合场景 | 优势 | 劣势 |
|---|---|---|---|
| TXT | 日志、简单文本 | 最简单 | 无结构 |
| CSV | 表格数据 | 通用、Excel兼容 | 无类型、无嵌套 |
| JSON | 配置、API数据 | 有结构、支持嵌套 | 比CSV冗余 |
| Excel | 报表、复杂表格 | 有格式、有公式 | 二进制、依赖库 |
五、实战场景
5.1 API响应保存为JSON
importjsonimportrequests response=requests.get("https://api.example.com/products")data=response.json()# 保存到文件withopen("D:\\data\\api_response.json","w",encoding="utf-8")asf:json.dump(data,f,ensure_ascii=False,indent=2)# 提取需要的数据转成CSVimportpandasaspd df=pd.DataFrame(data["products"])df.to_csv("D:\\data\\products.csv",index=False,encoding="utf-8-sig")5.2 多格式数据合并
importjsonimportpandasaspd# 从JSON读取withopen("D:\\data\\api_data.json","r",encoding="utf-8")asf:json_data=json.load(f)df_api=pd.DataFrame(json_data["products"])# 从CSV读取df_csv=pd.read_csv("D:\\data\\local_data.csv",encoding="utf-8-sig")# 从Excel读取df_excel=pd.read_excel("D:\\data\\manual.xlsx")# 合并所有数据df_all=pd.concat([df_api,df_csv,df_excel],ignore_index=True)# 输出为Exceldf_all.to_excel("D:\\data\\merged.xlsx",index=False)5.3 日志写入
fromdatetimeimportdatetimedefwrite_log(message,level="INFO"):"""写入日志文件"""timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S")log_line=f"[{timestamp}] [{level}]{message}\n"withopen("D:\\logs\\rpa.log","a",encoding="utf-8")asf:f.write(log_line)# 使用write_log("流程开始执行")write_log("采集到50条数据","INFO")write_log("网站连接超时","ERROR")write_log("流程执行完毕")六、避坑清单
坑1:文件操作后忘记关闭
不使用with语句时,文件需要手动f.close()。如果忘记关闭,文件锁不释放,其他操作会报错。永远用with open(...) as f自动管理。
坑2:大文件一次性读取
几百MB的日志文件一次性f.read()会占用大量内存。逐行读取更安全:
withopen("big_file.log","r")asf:forlineinf:process(line)坑3:CSV中的数字变成文本
CSV没有数据类型,所有值都是文本。用pandas读取时指定数据类型:
df=pd.read_csv("data.csv",dtype={"价格":float,"库存":int})坑4:JSON序列化失败
Python对象中包含日期、自定义类等无法直接序列化为JSON。需要先转换:
fromdatetimeimportdatetime data={"date":datetime.now().isoformat()}# 日期转字符串json.dumps(data)坑5:并发写入冲突
多个流程同时写同一个文件,数据会错乱。用文件锁或者分文件写入。
