影刀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()![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/3c3a9af1f0df447a8ac4faffc8d12a8a.png#pic_center)

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")![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/eb283a5a4cb8420ab5ece4d80bc0d01e.png#pic_center)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)![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/8d276b553730473788516f2e0ae8e44b.png#pic_center)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}

四、格式选择指南

五、实战场景

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：并发写入冲突

多个流程同时写同一个文件，数据会错乱。用文件锁或者分文件写入。