7 minutes
文件读写与 IO 操作
文件操作是编程中最常见的任务之一——读取配置文件、写入日志、处理数据文件、导入导出……几乎每个程序都需要和文件打交道。
Python 的文件操作接口设计得既简洁又强大。本章将从基础的 open() 函数开始,逐步深入到上下文管理器、CSV 处理、JSON 读写、现代路径操作以及临时文件的处理。
open() 函数
open() 是 Python 文件操作的入口,返回一个文件对象(file object):
# 打开文件读取
file = open("example.txt", "r", encoding="utf-8")
content = file.read()
file.close() # 务必关闭!
print(content)
文件模式
open() 的第二个参数是模式(mode),决定了你要对文件做什么操作:
| 模式 | 含义 | 文件指针位置 |
|---|---|---|
"r" |
读取(默认) | 开头 |
"w" |
写入(覆盖已有内容) | 开头(清空文件) |
"a" |
追加 | 末尾 |
"x" |
排他创建(文件已存在则报错) | 开头 |
"r+" |
读写 | 开头 |
"w+" |
读写(覆盖) | 开头(清空文件) |
"a+" |
读写(追加) | 末尾 |
加上类型修饰符:
| 修饰符 | 含义 |
|---|---|
"b" |
二进制模式(如 "rb"、"wb") |
"t" |
文本模式(默认,如 "rt" 等价于 "r") |
# 各种模式的示例
# 读取文本文件
with open("data.txt", "r", encoding="utf-8") as f:
data = f.read()
# 写入(覆盖)
with open("output.txt", "w", encoding="utf-8") as f:
f.write("Hello, World!")
# 追加
with open("log.txt", "a", encoding="utf-8") as f:
f.write("新日志行\n")
# 二进制模式(图片、视频等)
with open("image.jpg", "rb") as f:
raw_bytes = f.read()
# 排他创建
try:
with open("new_file.txt", "x", encoding="utf-8") as f:
f.write("这个文件之前不存在")
except FileExistsError:
print("文件已存在!")
with 语句与上下文管理器
永远使用 with 语句打开文件。 这是 Python 文件操作最重要的原则之一。
# 不推荐——手动管理资源
f = open("example.txt", "r")
try:
content = f.read()
finally:
f.close() # 可能会忘记,或者在异常时跳过
# 推荐——自动管理资源
with open("example.txt", "r", encoding="utf-8") as f:
content = f.read()
# 离开 with 块后文件自动关闭
with 语句确保了即使在发生异常时,文件也会被正确关闭。它是 Python 上下文管理器(Context Manager)的具体应用。
同时操作多个文件
# 复制文件——同时打开源文件和目标文件
with open("source.txt", "r", encoding="utf-8") as src, \
open("dest.txt", "w", encoding="utf-8") as dst:
for line in src:
dst.write(line.upper())
读取文件的方式
read() —— 一次读取全部
适合小文件:
with open("small.txt", "r", encoding="utf-8") as f:
content = f.read() # 返回整个文件的字符串
print(content)
也可以指定要读取的字符数:
with open("example.txt", "r", encoding="utf-8") as f:
head = f.read(100) # 前 100 个字符
middle = f.read(50) # 接着读 50 个字符
tail = f.read() # 剩下的全部
readline() —— 逐行读取
with open("poem.txt", "r", encoding="utf-8") as f:
line = f.readline()
while line: # readline() 在文件末尾返回空字符串
print(line, end="")
line = f.readline()
readlines() —— 读取所有行到列表
with open("lines.txt", "r", encoding="utf-8") as f:
lines = f.readlines() # 返回列表,每行是一个元素(包含换行符)
print(f"共 {len(lines)} 行")
推荐方式:直接迭代文件对象
# 最 Pythonic 的方式——内存友好,适合大文件
with open("large_file.txt", "r", encoding="utf-8") as f:
for line in f: # 迭代器——按需逐行读取
print(line.strip()) # strip() 去掉换行符
性能对比:
| 方式 | 内存使用 | 适用场景 |
|---|---|---|
read() |
全部加载 | 小文件(< 100MB) |
readline() |
单行 | 逐行处理 |
readlines() |
全部加载 | 需要随机访问行 |
迭代 for line in f |
单行 | 通用首选 |
写入文件的方式
write() —— 写入字符串
with open("output.txt", "w", encoding="utf-8") as f:
f.write("第一行\n")
f.write("第二行\n")
# write() 不会自动加换行符,需要手动加 \n
writelines() —— 写入多行
lines = ["第一行\n", "第二行\n", "第三行\n"]
with open("output.txt", "w", encoding="utf-8") as f:
f.writelines(lines) # 注意不会自动添加换行符!
文本模式 vs 二进制模式
文本模式("t",默认)
- 处理字符串(
str) - 自动处理行尾转换(Windows:
\r\n↔\n) - 需要指定编码(
encoding="utf-8") - 适合:
.txt、.csv、.json、.xml等文本文件
二进制模式("b")
- 处理字节(
bytes) - 不进行任何转换
- 不需要指定编码
- 适合:图片、视频、音频、压缩包、
.exe等
# 二进制读
with open("photo.jpg", "rb") as f:
header = f.read(4)
print(header) # b'\xff\xd8\xff\xe0' —— JPEG 文件头
# 二进制写
with open("copy.jpg", "wb") as f:
f.write(header) # 写入的是 bytes
文本模式的编码问题
编码问题是文件操作中最常见的陷阱之一:
# 错误——不指定编码
with open("data.txt", "r") as f: # 使用平台默认编码!
content = f.read()
# 正确——显式指定编码
with open("data.txt", "r", encoding="utf-8") as f:
content = f.read()
Windows 系统默认编码是 gbk,Linux/macOS 是 utf-8。不指定编码会导致代码在不同平台上行为不一致。
处理特殊编码:
# 处理带 BOM 的 UTF-8 文件(常见于 Windows 生成的 CSV)
with open("data.csv", "r", encoding="utf-8-sig") as f:
content = f.read()
# GBK 编码的中文文件
with open("old_data.txt", "r", encoding="gbk") as f:
content = f.read()
# 忽略编码错误
with open("messy.txt", "r", encoding="utf-8", errors="ignore") as f:
content = f.read()
# 替换无法解码的字符
with open("messy.txt", "r", encoding="utf-8", errors="replace") as f:
content = f.read()
CSV 文件处理
CSV(Comma-Separated Values)是最常见的数据交换格式之一。
使用内置 csv 模块
import csv
# 读取 CSV
with open("employees.csv", "r", encoding="utf-8") as f:
reader = csv.reader(f)
for row in reader:
print(row) # 每行是一个字符串列表
# 读取为字典(第一行为列名)
with open("employees.csv", "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
print(row["name"], row["age"], row["job"])
写入 CSV
import csv
# 写入标准 CSV
data = [
["name", "age", "job"], # 表头
["Alice", "28", "Engineer"],
["Bob", "35", "Designer"],
["Charlie", "22", "Student"],
]
with open("employees.csv", "w", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerows(data) # 写入多行
# 写入字典格式
with open("employees.csv", "w", encoding="utf-8", newline="") as f:
fieldnames = ["name", "age", "job"]
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerow({"name": "Alice", "age": "28", "job": "Engineer"})
writer.writerow({"name": "Bob", "age": "35", "job": "Designer"})
重要: 写入 CSV 时指定 newline="" 可以避免出现多余的空行。
处理 CSV 的常见陷阱
# 数值被当成字符串——需要手动转换
with open("data.csv", "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
age = int(row["age"]) # 手动转整数
salary = float(row["salary"]) # 手动转浮点数
# 字段中包含逗号或引号——csv 模块会自动处理
# "Smith, John", 30, "他说: ""你好"""
JSON 文件处理
JSON 是现代 Web API 和配置文件的标准格式。
json.dump() —— 写入 JSON
import json
data = {
"name": "Alice",
"age": 28,
"skills": ["Python", "SQL", "Docker"],
"active": True,
"address": None,
}
# 写入为 JSON 文件
with open("profile.json", "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
参数说明:
ensure_ascii=False:保留中文等非 ASCII 字符(默认会转义为\u...)indent=2:格式化缩进(方便人阅读)sort_keys=True:按键排序输出
# 写入为单行(节省空间,适合程序读取)
with open("profile.min.json", "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, separators=(",", ":"))
json.load() —— 读取 JSON
import json
with open("profile.json", "r", encoding="utf-8") as f:
data = json.load(f)
print(data["name"]) # Alice
print(data["skills"]) # ['Python', 'SQL', 'Docker']
json.dumps() / json.loads() —— 字符串操作
import json
# 对象 → JSON 字符串
json_str = json.dumps({"name": "Bob", "age": 30}, ensure_ascii=False, indent=2)
print(json_str)
# JSON 字符串 → 对象
parsed = json.loads('{"name": "Bob", "age": 30}')
print(parsed["name"]) # Bob
处理 JSON 中的特殊类型
from datetime import datetime
import json
# JSON 不支持 datetime,需要自定义处理
def custom_serializer(obj):
"""自定义 JSON 序列化器"""
if isinstance(obj, datetime):
return obj.isoformat()
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")
data = {
"event": "meeting",
"time": datetime.now(),
}
json_str = json.dumps(data, default=custom_serializer, ensure_ascii=False)
print(json_str) # {"event": "meeting", "time": "2026-05-13T14:30:00"}
pathlib.Path —— 现代路径操作
pathlib 是 Python 3.4+ 引入的面向对象路径操作库,比传统的 os.path 更加直观:
from pathlib import Path
# 创建路径
data_dir = Path("data")
config_file = Path("config/settings.json")
# 路径拼接(使用 / 运算符!)
project_dir = Path("/home/user/myproject")
data_path = project_dir / "data" / "raw" / "dataset.csv"
print(data_path) # /home/user/myproject/data/raw/dataset.csv
# 路径属性
print(data_path.name) # dataset.csv
print(data_path.stem) # dataset(不带扩展名)
print(data_path.suffix) # .csv
print(data_path.parent) # /home/user/myproject/data/raw
print(data_path.parents) # 所有父路径的生成器
# 路径存在性检查
if data_path.exists():
print("文件存在!")
# 目录 vs 文件
p = Path("some_path")
p.is_dir()
p.is_file()
Path 的文件操作
from pathlib import Path
# 读取文件
content = Path("example.txt").read_text(encoding="utf-8")
print(content)
# 写入文件
Path("output.txt").write_text("Hello, World!", encoding="utf-8")
# 读取二进制文件
data = Path("image.jpg").read_bytes()
# 写入二进制文件
Path("copy.jpg").write_bytes(data)
# 创建目录
Path("output/data/2026").mkdir(parents=True, exist_ok=True)
用 pathlib 遍历目录:
from pathlib import Path
data_dir = Path("data")
# 列出所有 .csv 文件
for csv_file in data_dir.glob("*.csv"):
print(csv_file.name)
# 递归查找所有 .json 文件
for json_file in data_dir.rglob("*.json"):
print(json_file.relative_to(data_dir))
# 列出当前目录下所有文件(非递归)
for entry in data_dir.iterdir():
if entry.is_file():
print(f"文件: {entry.name}")
elif entry.is_dir():
print(f"目录: {entry.name}")
pathlib vs os.path
# 传统方式
import os
path = os.path.join("data", "raw", "dataset.csv")
name = os.path.basename(path)
dir_name = os.path.dirname(path)
# pathlib 方式
from pathlib import Path
path = Path("data") / "raw" / "dataset.csv"
name = path.name
dir_name = path.parent
推荐: 新代码优先使用 pathlib.Path。
tempfile —— 临时文件
在处理大型数据或测试时,常常需要临时文件。tempfile 模块可以安全地创建临时文件和目录:
import tempfile
from pathlib import Path
# 创建临时文件(自动清理)
with tempfile.TemporaryFile(mode="w+", encoding="utf-8") as f:
f.write("临时数据")
f.seek(0) # 回到文件开头
print(f.read()) # 临时数据
# 离开 with 块后文件自动删除
# 创建临时目录
with tempfile.TemporaryDirectory() as tmpdir:
tmp_path = Path(tmpdir)
work_file = tmp_path / "work.txt"
work_file.write_text("工作中...", encoding="utf-8")
print(f"临时文件: {work_file}")
print(f"内容: {work_file.read_text(encoding='utf-8')}")
# 目录和其中的文件自动删除
# 持久化的临时文件(需要手动删除)
with tempfile.NamedTemporaryFile(delete=False, suffix=".csv") as f:
f.write(b"name,age\nAlice,28")
temp_name = f.name # 保存文件名
print(f"临时文件路径: {temp_name}")
# 使用完后手动删除
Path(temp_name).unlink()
完整示例:数据处理管道
综合运用本章知识,从 CSV 读取数据,处理后再写入 JSON:
import csv
import json
from pathlib import Path
def process_sales_data(input_path: Path, output_path: Path):
"""处理销售数据:从 CSV 读取,清洗,汇总后写入 JSON"""
# 读取原始数据
records = []
with open(input_path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
# 数据清洗
record = {
"date": row["date"],
"product": row["product"].strip(),
"quantity": int(row["quantity"]),
"price": float(row["price"]),
"total": int(row["quantity"]) * float(row["price"]),
}
records.append(record)
# 按产品汇总
summary = {}
for record in records:
product = record["product"]
if product not in summary:
summary[product] = {
"total_quantity": 0,
"total_revenue": 0.0,
"transactions": 0,
}
summary[product]["total_quantity"] += record["quantity"]
summary[product]["total_revenue"] += record["total"]
summary[product]["transactions"] += 1
# 写入 JSON
output = {
"total_records": len(records),
"summary": summary,
"details": records,
}
with open(output_path, "w", encoding="utf-8") as f:
json.dump(output, f, ensure_ascii=False, indent=2)
print(f"处理完成!共 {len(records)} 条记录")
print(f"结果保存至: {output_path}")
# 使用
data_dir = Path(".")
input_file = data_dir / "sales.csv"
output_file = data_dir / "sales_summary.json"
process_sales_data(input_file, output_file)
实战练习
练习 1:日志文件分析器
from pathlib import Path
from collections import Counter
def analyze_log(log_path: Path):
"""分析日志文件:统计各日志级别出现的次数"""
level_counts = Counter()
error_lines = []
with open(log_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
# 假设日志格式:[INFO] 消息 或 [ERROR] 消息
if line.startswith("["):
level = line[1:].split("]")[0]
level_counts[level] += 1
if level == "ERROR":
error_lines.append(line)
print("日志级别统计:")
for level, count in level_counts.most_common():
print(f" {level}: {count} 次")
print(f"\n错误详情(共 {len(error_lines)} 条):")
for line in error_lines[:5]: # 只显示前 5 条
print(f" {line}")
练习 2:文件批量重命名
from pathlib import Path
def batch_rename(directory: Path, prefix: str, extension: str = None):
"""批量重命名文件:添加前缀,可选筛选扩展名"""
for i, file_path in enumerate(directory.iterdir(), start=1):
if not file_path.is_file():
continue
if extension and file_path.suffix != extension:
continue
new_name = f"{prefix}_{i:03d}{file_path.suffix}"
new_path = file_path.with_name(new_name)
file_path.rename(new_path)
print(f" {file_path.name} → {new_name}")
小结
本章我们系统学习了 Python 的文件操作:
open()函数和各种模式(r/w/a/x/r+)with语句确保资源自动释放- 读取方式:
read()、readline()、readlines()、迭代文件对象 - 写入方式:
write()、writelines() - 文本 vs 二进制:编码处理与 BOM
- CSV 模块:读取和写入表格数据
- JSON 模块:
dump/dumps、load/loads,序列化自定义类型 pathlib.Path:面向对象的路径操作tempfile:安全的临时文件和目录管理
下一步: 程序总会有出错的时候。下一章我们将学习如何处理异常、进行调试,编写更健壮的代码。
Summary: 文件读写、CSV/JSON、pathlib 与上下文管理。