文件操作
文件操作主要包括:
读取文件
写入文件
追加内容
复制文件
移动文件
删除文件
创建目录
遍历目录
批量处理文件
Python 文件操作常用两套工具:
open()
以及:
pathlib.Path
现代 Python 项目通常推荐优先使用 pathlib,因为路径处理更加清晰、跨平台。
例如有以下目录:
project/
├── data/
│ ├── stocks.txt
│ └── stocks.json
├── output/
└── main.py
其中:
project 目录
data 子目录
stocks.txt 文本文件
stocks.json JSON 文件
main.py Python 文件
文件路径可以分为:
相对路径
绝对路径
相对路径:
data/stocks.txt
绝对路径示例:
/Users/martin/project/data/stocks.txt
Windows 示例:
C:\Users\martin\project\data\stocks.txt
基本格式:
file = open(
"文件路径",
"打开模式",
encoding="utf-8",
)
例如:
file = open(
"data.txt",
"r",
encoding="utf-8",
)
使用完后必须关闭:
file.close()
完整写法:
file = open(
"data.txt",
"r",
encoding="utf-8",
)
content = file.read()
file.close()
print(content)
如果读取过程中发生异常,close() 可能不会执行。
因此更推荐使用 with。
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
content = file.read()
print(content)
离开 with 代码块以后,文件会自动关闭。
推荐结构:
with open(
file_path,
mode,
encoding="utf-8",
) as file:
操作文件
优点:
- 自动关闭文件
- 异常发生时也会释放资源
- 代码更清晰
- 不容易忘记调用
close()
常用模式:
| 模式 | 含义 |
|---|---|
r |
只读,文件必须存在 |
w |
写入,文件存在时覆盖 |
a |
追加,在文件末尾写入 |
x |
新建文件,文件已存在时失败 |
b |
二进制模式 |
t |
文本模式,默认模式 |
+ |
同时支持读写 |
常见组合:
r 读取文本
w 写入文本并覆盖
a 追加文本
rb 读取二进制
wb 写入二进制
r+ 读写文件
w+ 覆盖后读写
a+ 追加并读取
文件 data.txt:
第一行
第二行
第三行
读取全部内容:
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
content = file.read()
print(content)
read() 返回字符串:
print(type(content))
结果:
<class 'str'>
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
content = file.read(5)
print(content)
read(5) 表示读取五个字符。
再次读取时,会从当前位置继续:
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
first = file.read(5)
second = file.read(5)
print(first)
print(second)
readline() 每次读取一行:
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
first_line = file.readline()
second_line = file.readline()
print(first_line)
print(second_line)
读取到的行通常保留结尾换行符:
print(repr(first_line))
结果类似:
'第一行\n'
可以使用:
line.rstrip("\n")
或者:
line.strip()
注意:
strip()
会同时删除两端空格、制表符和换行符。
如果只想删除换行符,推荐:
line.rstrip("\n")
一次读取所有行:
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
lines = file.readlines()
print(lines)
结果类似:
[
"第一行\n",
"第二行\n",
"第三行\n",
]
readlines() 会把整个文件加载到内存。
不适合特别大的文件。
推荐按行读取:
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
for line in file:
print(line.rstrip("\n"))
这种方式:
- 每次只读取一行
- 内存占用较小
- 适合大文件
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
for line_number, line in enumerate(
file,
start=1,
):
print(
line_number,
line.rstrip("\n"),
)
使用 w 模式:
with open(
"output.txt",
"w",
encoding="utf-8",
) as file:
file.write("Hello Python")
如果文件不存在,会创建文件。
如果文件已经存在,会清空原内容后重新写入。
with open(
"output.txt",
"w",
encoding="utf-8",
) as file:
file.write("第一行\n")
file.write("第二行\n")
file.write("第三行\n")
write() 不会自动添加换行符。
需要手动写入:
"\n"
write() 返回写入的字符数量:
with open(
"output.txt",
"w",
encoding="utf-8",
) as file:
count = file.write("Python")
print(count)
结果:
6
lines = [
"第一行\n",
"第二行\n",
"第三行\n",
]
with open(
"output.txt",
"w",
encoding="utf-8",
) as file:
file.writelines(lines)
writelines() 不会自动添加换行符。
错误写法:
lines = [
"第一行",
"第二行",
"第三行",
]
写入后可能变成:
第一行第二行第三行
可以提前添加换行符:
lines = [
f"{line}\n"
for line in [
"第一行",
"第二行",
"第三行",
]
]
或者:
text = "\n".join([
"第一行",
"第二行",
"第三行",
])
with open(
"output.txt",
"w",
encoding="utf-8",
) as file:
file.write(text)
使用 a 模式:
with open(
"log.txt",
"a",
encoding="utf-8",
) as file:
file.write("新增日志\n")
原有内容不会被清空,新内容添加到文件末尾。
例如:
from datetime import datetime
with open(
"app.log",
"a",
encoding="utf-8",
) as file:
now = datetime.now()
file.write(
f"{now:%Y-%m-%d %H:%M:%S} "
"程序启动\n"
)
使用 x 模式:
with open(
"new_file.txt",
"x",
encoding="utf-8",
) as file:
file.write("新文件")
如果文件已经存在,会抛出:
FileExistsError
适合防止意外覆盖文件:
try:
with open(
"config.txt",
"x",
encoding="utf-8",
) as file:
file.write("配置内容")
except FileExistsError:
print("配置文件已经存在")
常见编码:
UTF-8
GBK
GB2312
UTF-16
ASCII
现代项目通常推荐:
encoding="utf-8"
例如:
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
content = file.read()
如果编码不匹配,可能出现:
UnicodeDecodeError
例如文件使用 GBK:
with open(
"data.txt",
"r",
encoding="gbk",
) as file:
content = file.read()
忽略无法解码的字符:
with open(
"data.txt",
"r",
encoding="utf-8",
errors="ignore",
) as file:
content = file.read()
替换无法解码的字符:
errors="replace"
常见选项:
strict 默认,遇到错误抛出异常
ignore 忽略错误字符
replace 使用替代字符
重要数据不建议直接使用:
errors="ignore"
因为可能悄悄丢失内容。
不同系统换行符可能不同:
Linux/macOS \n
Windows \r\n
旧版 macOS \r
Python 文本模式通常会自动处理换行符。
写入时:
with open(
"data.txt",
"w",
encoding="utf-8",
newline="\n",
) as file:
file.write("第一行\n第二行\n")
读取 CSV 时,通常推荐:
newline=""
文件对象会记录当前读写位置。
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
print(file.tell())
file.read(5)
print(file.tell())
tell() 返回当前文件位置。
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
first = file.read(5)
file.seek(0)
second = file.read(5)
print(first)
print(second)
seek(0) 回到文件开头。
在文本模式中,不建议随意使用复杂偏移。
处理字节位置时应使用二进制模式。
file = open(
"data.txt",
"r",
encoding="utf-8",
)
print(file.closed)
file.close()
print(file.closed)
结果:
False
True
使用 with:
with open(
"data.txt",
"r",
encoding="utf-8",
) as file:
print(file.closed)
print(file.closed)
离开 with 后文件已经关闭。
导入:
from pathlib import Path
创建路径对象:
file_path = Path("data.txt")
路径对象不是文件内容,而是文件或目录路径的表示。
from pathlib import Path
current_directory = Path.cwd()
print(current_directory)
home_directory = Path.home()
print(home_directory)
推荐使用 /:
file_path = (
Path("data")
/ "stocks"
/ "600519.json"
)
不推荐手动拼接:
file_path = (
"data/"
+ "stocks/"
+ "600519.json"
)
pathlib 会处理不同操作系统的路径分隔符。
file_path = Path(
"data/stocks/600519.json"
)
文件名:
print(file_path.name)
结果:
600519.json
文件主名:
print(file_path.stem)
结果:
600519
扩展名:
print(file_path.suffix)
结果:
.json
父目录:
print(file_path.parent)
结果:
data/stocks
所有父目录:
print(file_path.parents)
file_path = Path(
"archive.tar.gz"
)
print(file_path.suffix)
结果:
.gz
获取所有扩展名:
print(file_path.suffixes)
结果:
[".tar", ".gz"]
file_path = Path("data.txt")
if file_path.exists():
print("路径存在")
判断是否为文件:
if file_path.is_file():
print("这是文件")
判断是否为目录:
folder_path = Path("data")
if folder_path.is_dir():
print("这是目录")
使用 Path.read_text():
from pathlib import Path
file_path = Path("data.txt")
content = file_path.read_text(
encoding="utf-8",
)
print(content)
它等价于:
with open(
file_path,
"r",
encoding="utf-8",
) as file:
content = file.read()
适合较小文本文件。
大文件仍然推荐逐行读取。
from pathlib import Path
file_path = Path("output.txt")
file_path.write_text(
"Hello Python",
encoding="utf-8",
)
如果文件已经存在,会覆盖原内容。
返回写入的字符数量:
count = file_path.write_text(
"Hello Python",
encoding="utf-8",
)
print(count)
folder = Path("data")
folder.mkdir()
如果目录已存在,会报错:
FileExistsError
允许目录已存在:
folder.mkdir(
exist_ok=True,
)
folder = Path(
"data/stocks/daily"
)
folder.mkdir(
parents=True,
exist_ok=True,
)
参数含义:
parents=True 自动创建上级目录
exist_ok=True 目录已存在时不报错
推荐写法:
output_dir.mkdir(
parents=True,
exist_ok=True,
)
file_path = Path("empty.txt")
file_path.touch()
如果文件不存在,会创建空文件。
如果文件存在,会更新时间。
防止覆盖已有文件:
file_path.touch(
exist_ok=False,
)
文件存在时抛出:
FileExistsError
old_path = Path("old.txt")
new_path = Path("new.txt")
old_path.rename(new_path)
重命名后,旧路径不再存在。
source = Path("data.txt")
target = Path(
"archive/data.txt"
)
target.parent.mkdir(
parents=True,
exist_ok=True,
)
source.replace(target)
replace() 在目标存在时可能覆盖目标。
更复杂的移动操作可以使用:
shutil.move()
file_path = Path("data.txt")
file_path.unlink()
如果文件不存在,会抛出:
FileNotFoundError
Python 3.8 以上可以:
file_path.unlink(
missing_ok=True,
)
文件不存在时不报错。
删除空目录:
folder = Path("empty_folder")
folder.rmdir()
目录必须为空。
删除包含内容的目录,需要使用:
import shutil
shutil.rmtree("folder")
这是危险操作,会递归删除所有内容。
推荐先确认路径:
from pathlib import Path
import shutil
folder = Path("temp_output")
if folder.exists() and folder.is_dir():
shutil.rmtree(folder)
不要对不确定路径调用 rmtree()。
folder = Path("data")
for path in folder.iterdir():
print(path)
只遍历当前目录,不递归进入子目录。
区分文件和目录:
for path in folder.iterdir():
if path.is_file():
print("文件:", path)
elif path.is_dir():
print("目录:", path)
查找当前目录下所有 TXT 文件:
for file_path in folder.glob(
"*.txt"
):
print(file_path)
查找 JSON:
folder.glob("*.json")
查找特定开头:
folder.glob("stock_*.csv")
递归查找所有子目录:
for file_path in folder.rglob(
"*.txt"
):
print(file_path)
等价于类似:
当前目录以及所有子目录中的 TXT 文件
for file_path in folder.glob(
"**/*.json"
):
print(file_path)
通常 rglob() 更直观:
folder.rglob("*.json")
files = sorted(
Path("data").glob("*.txt")
)
for file_path in files:
print(file_path)
按文件名排序:
files = sorted(
Path("data").glob("*.txt"),
key=lambda path: path.name,
)
按修改时间排序:
files = sorted(
Path("data").glob("*.txt"),
key=lambda path: (
path.stat().st_mtime
),
)
最新文件优先:
files = sorted(
Path("data").glob("*.txt"),
key=lambda path: (
path.stat().st_mtime
),
reverse=True,
)
file_path = Path("data.txt")
info = file_path.stat()
文件大小:
print(info.st_size)
单位是字节。
修改时间:
print(info.st_mtime)
创建时间或状态变更时间:
print(info.st_ctime)
不同系统中,st_ctime 含义可能不同。
转换时间:
from datetime import datetime
modified_time = datetime.fromtimestamp(
info.st_mtime
)
print(modified_time)
def format_file_size(
size: int,
) -> str:
units = [
"B",
"KB",
"MB",
"GB",
"TB",
]
value = float(size)
for unit in units:
if value < 1024:
return f"{value:.2f} {unit}"
value /= 1024
return f"{value:.2f} PB"
使用:
size = file_path.stat().st_size
print(format_file_size(size))
图片、音频、视频、压缩包等属于二进制文件。
读取二进制文件:
with open(
"image.jpg",
"rb",
) as file:
data = file.read()
print(type(data))
结果:
<class 'bytes'>
写入二进制文件:
with open(
"copy.jpg",
"wb",
) as file:
file.write(data)
二进制模式不使用:
encoding="utf-8"
file_path = Path("image.jpg")
data = file_path.read_bytes()
写入:
target_path = Path("copy.jpg")
target_path.write_bytes(data)
适合较小文件。
大文件不建议一次性全部加载到内存。
使用 shutil:
import shutil
复制文件内容:
shutil.copyfile(
"source.txt",
"target.txt",
)
复制文件及权限信息:
shutil.copy(
"source.txt",
"target.txt",
)
尽可能复制完整元数据:
shutil.copy2(
"source.txt",
"target.txt",
)
推荐:
from pathlib import Path
import shutil
source = Path("data.txt")
target = Path(
"backup/data.txt"
)
target.parent.mkdir(
parents=True,
exist_ok=True,
)
shutil.copy2(
source,
target,
)
import shutil
shutil.copytree(
"source_folder",
"target_folder",
)
如果目标目录已存在:
shutil.copytree(
"source_folder",
"target_folder",
dirs_exist_ok=True,
)
import shutil
shutil.move(
"source.txt",
"archive/source.txt",
)
也可以移动目录:
shutil.move(
"old_folder",
"new_folder",
)
例如,将目录中的所有 .jpeg 改成 .jpg:
from pathlib import Path
folder = Path("images")
for file_path in folder.glob(
"*.jpeg"
):
new_path = file_path.with_suffix(
".jpg"
)
file_path.rename(new_path)
folder = Path("images")
for file_path in folder.iterdir():
if not file_path.is_file():
continue
new_name = (
f"backup_{file_path.name}"
)
new_path = file_path.with_name(
new_name
)
file_path.rename(new_path)
files = sorted(
Path("images").glob("*.jpg")
)
for index, file_path in enumerate(
files,
start=1,
):
new_name = (
f"image_{index:03d}"
f"{file_path.suffix}"
)
new_path = file_path.with_name(
new_name
)
file_path.rename(new_path)
结果:
image_001.jpg
image_002.jpg
image_003.jpg
批量重命名前,应先检查是否会发生名称冲突。
file_path = Path(
"data/stocks.txt"
)
修改文件名:
new_path = file_path.with_name(
"new_stocks.txt"
)
修改扩展名:
new_path = file_path.with_suffix(
".csv"
)
修改主文件名:
new_path = file_path.with_stem(
"stock_list"
)
结果:
data/stock_list.txt
CSV 是常用表格文本格式。
文件示例:
code,name,price
600519,贵州茅台,1500.0
000001,平安银行,11.5
Python 使用:
import csv
import csv
with open(
"stocks.csv",
"r",
encoding="utf-8",
newline="",
) as file:
reader = csv.reader(file)
for row in reader:
print(row)
每一行是列表:
[
"600519",
"贵州茅台",
"1500.0",
]
CSV 中读取的数据默认都是字符串。
with open(
"stocks.csv",
"r",
encoding="utf-8",
newline="",
) as file:
reader = csv.reader(file)
header = next(reader)
for row in reader:
print(row)
更推荐按字段名读取:
import csv
with open(
"stocks.csv",
"r",
encoding="utf-8",
newline="",
) as file:
reader = csv.DictReader(file)
for row in reader:
print(row["code"])
print(row["name"])
print(float(row["price"]))
每一行类似:
{
"code": "600519",
"name": "贵州茅台",
"price": "1500.0",
}
import csv
rows = [
[
"code",
"name",
"price",
],
[
"600519",
"贵州茅台",
1500.0,
],
[
"000001",
"平安银行",
11.5,
],
]
with open(
"stocks.csv",
"w",
encoding="utf-8",
newline="",
) as file:
writer = csv.writer(file)
writer.writerows(rows)
stocks = [
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.0,
},
{
"code": "000001",
"name": "平安银行",
"price": 11.5,
},
]
import csv
field_names = [
"code",
"name",
"price",
]
with open(
"stocks.csv",
"w",
encoding="utf-8",
newline="",
) as file:
writer = csv.DictWriter(
file,
fieldnames=field_names,
)
writer.writeheader()
writer.writerows(stocks)
某些 Windows Excel 环境中,可以使用:
encoding="utf-8-sig"
例如:
with open(
"stocks.csv",
"w",
encoding="utf-8-sig",
newline="",
) as file:
writer = csv.DictWriter(
file,
fieldnames=field_names,
)
writer.writeheader()
writer.writerows(stocks)
导入:
import json
JSON 示例:
[
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.0
}
]
import json
with open(
"stocks.json",
"r",
encoding="utf-8",
) as file:
data = json.load(file)
print(data)
stocks = [
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.0,
},
{
"code": "000001",
"name": "平安银行",
"price": 11.5,
},
]
import json
with open(
"stocks.json",
"w",
encoding="utf-8",
) as file:
json.dump(
stocks,
file,
ensure_ascii=False,
indent=2,
)
参数:
ensure_ascii=False 中文正常显示
indent=2 使用两个空格缩进
from pathlib import Path
import json
file_path = Path("stocks.json")
text = json.dumps(
stocks,
ensure_ascii=False,
indent=2,
)
file_path.write_text(
text,
encoding="utf-8",
)
读取:
text = file_path.read_text(
encoding="utf-8",
)
stocks = json.loads(text)
import json
from pathlib import Path
def load_json(
file_path: Path,
) -> object:
try:
text = file_path.read_text(
encoding="utf-8",
)
except FileNotFoundError as error:
raise FileNotFoundError(
f"文件不存在:{file_path}"
) from error
except PermissionError as error:
raise PermissionError(
f"没有读取权限:{file_path}"
) from error
try:
return json.loads(text)
except json.JSONDecodeError as error:
raise ValueError(
"JSON 格式错误:"
f"第 {error.lineno} 行,"
f"第 {error.colno} 列"
) from error
大量数据可以使用 JSON Lines 格式:
{"code":"600519","name":"贵州茅台"}
{"code":"000001","name":"平安银行"}
{"code":"300750","name":"宁德时代"}
常见扩展名:
.jsonl
.ndjson
写入:
import json
stocks = [
{
"code": "600519",
"name": "贵州茅台",
},
{
"code": "000001",
"name": "平安银行",
},
]
with open(
"stocks.jsonl",
"w",
encoding="utf-8",
) as file:
for stock in stocks:
line = json.dumps(
stock,
ensure_ascii=False,
)
file.write(line + "\n")
逐行读取:
with open(
"stocks.jsonl",
"r",
encoding="utf-8",
) as file:
for line_number, line in enumerate(
file,
start=1,
):
clean_line = line.strip()
if not clean_line:
continue
stock = json.loads(
clean_line
)
print(line_number, stock)
JSON Lines 适合:
- 大量记录
- 流式处理
- 日志数据
- 每行独立解析
- 单条错误不影响整个文件
使用:
import tempfile
import tempfile
with tempfile.NamedTemporaryFile(
mode="w+",
encoding="utf-8",
suffix=".txt",
) as file:
file.write("临时内容")
file.seek(0)
print(file.read())
离开 with 后,临时文件通常会自动删除。
import tempfile
from pathlib import Path
with tempfile.TemporaryDirectory() as directory:
temp_dir = Path(directory)
file_path = (
temp_dir
/ "data.txt"
)
file_path.write_text(
"临时数据",
encoding="utf-8",
)
print(file_path.read_text(
encoding="utf-8",
))
离开 with 后,临时目录及其内容会自动清理。
直接写文件时,如果程序中途崩溃,目标文件可能只写入一部分。
更安全的方式:
先写临时文件
确认成功
再替换正式文件
示例:
from pathlib import Path
import tempfile
import os
def safe_write_text(
file_path: Path,
content: str,
*,
encoding: str = "utf-8",
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
with tempfile.NamedTemporaryFile(
mode="w",
encoding=encoding,
dir=file_path.parent,
delete=False,
) as temp_file:
temp_file.write(content)
temp_path = Path(
temp_file.name
)
try:
os.replace(
temp_path,
file_path,
)
except Exception:
temp_path.unlink(
missing_ok=True,
)
raise
os.replace() 通常可以原子替换目标文件。
适合:
- 配置文件
- 重要 JSON
- 财务数据
- 数据处理结果
- 不能接受半写入状态的文件
不推荐:
content = file_path.read_text(
encoding="utf-8",
)
如果文件很大,会一次性占用大量内存。
推荐逐行处理:
with file_path.open(
"r",
encoding="utf-8",
) as file:
for line in file:
process(line)
def read_binary_chunks(
file_path: Path,
*,
chunk_size: int = 1024 * 1024,
):
with file_path.open(
"rb"
) as file:
while chunk := file.read(
chunk_size
):
yield chunk
使用:
for chunk in read_binary_chunks(
Path("large_file.bin")
):
print(len(chunk))
from pathlib import Path
def copy_large_file(
source: Path,
target: Path,
*,
chunk_size: int = 1024 * 1024,
) -> None:
target.parent.mkdir(
parents=True,
exist_ok=True,
)
with source.open("rb") as source_file:
with target.open("wb") as target_file:
while chunk := source_file.read(
chunk_size
):
target_file.write(chunk)
实际复制通常可以直接使用:
shutil.copy2()
手动分块主要用于:
- 进度统计
- 限速
- 自定义校验
- 数据转换
from pathlib import Path
def count_lines(
file_path: Path,
) -> int:
with file_path.open(
"r",
encoding="utf-8",
) as file:
return sum(
1
for _ in file
)
使用:
line_count = count_lines(
Path("data.txt")
)
print(line_count)
不会一次性读取全部内容。
from pathlib import Path
def search_text(
file_path: Path,
keyword: str,
) -> list[
tuple[int, str]
]:
results: list[
tuple[int, str]
] = []
with file_path.open(
"r",
encoding="utf-8",
) as file:
for line_number, line in enumerate(
file,
start=1,
):
if keyword in line:
results.append(
(
line_number,
line.rstrip("\n"),
)
)
return results
使用:
results = search_text(
Path("data.txt"),
"贵州茅台",
)
for line_number, line in results:
print(line_number, line)
from pathlib import Path
def search_directory(
folder: Path,
keyword: str,
*,
pattern: str = "*.txt",
) -> list[
tuple[Path, int, str]
]:
results: list[
tuple[Path, int, str]
] = []
for file_path in folder.rglob(
pattern
):
try:
with file_path.open(
"r",
encoding="utf-8",
) as file:
for line_number, line in enumerate(
file,
start=1,
):
if keyword in line:
results.append(
(
file_path,
line_number,
line.rstrip(
"\n"
),
)
)
except UnicodeDecodeError:
continue
return results
读取、替换、写回:
from pathlib import Path
file_path = Path("data.txt")
content = file_path.read_text(
encoding="utf-8",
)
new_content = content.replace(
"旧内容",
"新内容",
)
file_path.write_text(
new_content,
encoding="utf-8",
)
如果文件较大,可以逐行处理并写入新文件。
from pathlib import Path
def replace_large_file(
source: Path,
target: Path,
old: str,
new: str,
) -> None:
target.parent.mkdir(
parents=True,
exist_ok=True,
)
with source.open(
"r",
encoding="utf-8",
) as source_file:
with target.open(
"w",
encoding="utf-8",
) as target_file:
for line in source_file:
target_file.write(
line.replace(
old,
new,
)
)
可以通过哈希判断文件内容是否相同。
from pathlib import Path
import hashlib
def calculate_sha256(
file_path: Path,
*,
chunk_size: int = 1024 * 1024,
) -> str:
digest = hashlib.sha256()
with file_path.open(
"rb"
) as file:
while chunk := file.read(
chunk_size
):
digest.update(chunk)
return digest.hexdigest()
使用:
hash_value = calculate_sha256(
Path("data.zip")
)
print(hash_value)
比较文件:
def files_are_equal(
first: Path,
second: Path,
) -> bool:
if (
first.stat().st_size
!= second.stat().st_size
):
return False
return (
calculate_sha256(first)
== calculate_sha256(second)
)
使用 shutil:
import shutil
压缩目录为 ZIP:
shutil.make_archive(
"backup",
"zip",
"data",
)
会生成:
backup.zip
解压:
shutil.unpack_archive(
"backup.zip",
"output_folder",
)
from zipfile import ZipFile
创建 ZIP:
from pathlib import Path
from zipfile import ZipFile
files = [
Path("a.txt"),
Path("b.txt"),
]
with ZipFile(
"files.zip",
"w",
) as zip_file:
for file_path in files:
zip_file.write(
file_path,
arcname=file_path.name,
)
读取 ZIP 文件列表:
with ZipFile(
"files.zip",
"r",
) as zip_file:
print(zip_file.namelist())
解压全部:
with ZipFile(
"files.zip",
"r",
) as zip_file:
zip_file.extractall(
"output"
)
处理来源不明的压缩包时,要防止路径穿越问题。
判断是否可读、可写,可以使用:
import os
file_path = Path("data.txt")
print(
os.access(
file_path,
os.R_OK,
)
)
print(
os.access(
file_path,
os.W_OK,
)
)
但即使提前判断,真正操作时权限仍可能变化。
因此仍需捕获:
PermissionError
示例:
try:
content = file_path.read_text(
encoding="utf-8",
)
except PermissionError:
print("没有读取权限")
假设代码:
Path("data/stocks.json")
相对路径是相对于:
Path.cwd()
而不一定是相对于 Python 文件所在目录。
查看当前工作目录:
print(Path.cwd())
from pathlib import Path
BASE_DIR = Path(
__file__
).resolve().parent
读取脚本旁边的文件:
data_path = (
BASE_DIR
/ "data"
/ "stocks.json"
)
完整示例:
from pathlib import Path
import json
BASE_DIR = Path(
__file__
).resolve().parent
DATA_FILE = (
BASE_DIR
/ "data"
/ "stocks.json"
)
def load_stocks() -> object:
text = DATA_FILE.read_text(
encoding="utf-8",
)
return json.loads(text)
在交互式环境中,__file__ 可能不存在。
常见异常:
FileNotFoundError
FileExistsError
PermissionError
IsADirectoryError
NotADirectoryError
UnicodeDecodeError
UnicodeEncodeError
OSError
示例:
from pathlib import Path
def read_text_file(
file_path: Path,
) -> str:
try:
return file_path.read_text(
encoding="utf-8",
)
except FileNotFoundError as error:
raise FileNotFoundError(
f"文件不存在:{file_path}"
) from error
except IsADirectoryError as error:
raise ValueError(
f"路径是目录,不是文件:"
f"{file_path}"
) from error
except PermissionError as error:
raise PermissionError(
f"没有读取权限:"
f"{file_path}"
) from error
except UnicodeDecodeError as error:
raise ValueError(
f"文件编码不是 UTF-8:"
f"{file_path}"
) from error
下面的代码存在竞态条件:
if file_path.exists():
content = file_path.read_text(
encoding="utf-8",
)
在检查后、读取前,文件仍可能被删除。
更可靠:
try:
content = file_path.read_text(
encoding="utf-8",
)
except FileNotFoundError:
print("文件不存在")
可以先判断提高可读性,但关键操作仍应处理异常。
推荐让函数接收 Path:
from pathlib import Path
def read_file(
file_path: Path,
) -> str:
return file_path.read_text(
encoding="utf-8",
)
调用:
content = read_file(
Path("data.txt")
)
如果希望同时支持字符串和 Path:
from os import PathLike
def read_file(
file_path: str | PathLike[str],
) -> str:
path = Path(file_path)
return path.read_text(
encoding="utf-8",
)
from collections.abc import Iterator
from pathlib import Path
def read_non_empty_lines(
file_path: Path,
) -> Iterator[str]:
with file_path.open(
"r",
encoding="utf-8",
) as file:
for line in file:
clean_line = line.strip()
if clean_line:
yield clean_line
使用:
for line in read_non_empty_lines(
Path("data.txt")
):
print(line)
适合大文件和流式处理。
文本文件:
600519,贵州茅台,1500.0
000001,平安银行,11.5
300750,宁德时代,300.0
错误数据
定义异常:
class StockFileError(Exception):
"""股票文件处理异常。"""
解析单行:
def parse_stock_line(
line: str,
) -> dict[str, object]:
parts = [
part.strip()
for part in line.split(",")
]
if len(parts) != 3:
raise StockFileError(
f"字段数量错误:{line!r}"
)
code, name, price_text = parts
if (
len(code) != 6
or not code.isdigit()
):
raise StockFileError(
f"股票代码无效:{code!r}"
)
try:
price = float(price_text)
except ValueError as error:
raise StockFileError(
f"股票价格无效:"
f"{price_text!r}"
) from error
if price < 0:
raise StockFileError(
"股票价格不能小于零"
)
return {
"code": code,
"name": name,
"price": price,
}
读取文件:
from pathlib import Path
def load_stocks(
file_path: Path,
) -> tuple[
list[dict[str, object]],
list[tuple[int, str]],
]:
stocks: list[
dict[str, object]
] = []
errors: list[
tuple[int, str]
] = []
try:
file = file_path.open(
"r",
encoding="utf-8",
)
except FileNotFoundError as error:
raise StockFileError(
f"股票文件不存在:"
f"{file_path}"
) from error
with file:
for line_number, line in enumerate(
file,
start=1,
):
clean_line = line.strip()
if not clean_line:
continue
try:
stock = parse_stock_line(
clean_line
)
except StockFileError as error:
errors.append(
(
line_number,
str(error),
)
)
else:
stocks.append(stock)
return stocks, errors
使用:
stocks, errors = load_stocks(
Path("stocks.txt")
)
for stock in stocks:
print(stock)
for line_number, error in errors:
print(
f"第 {line_number} 行:"
f"{error}"
)
需求:
把目录中的文件按扩展名分类
图片放入 images
文档放入 documents
音频放入 audios
其他文件放入 others
from pathlib import Path
import shutil
定义分类:
FILE_CATEGORIES = {
"images": {
".jpg",
".jpeg",
".png",
".gif",
".webp",
},
"documents": {
".txt",
".pdf",
".doc",
".docx",
".xls",
".xlsx",
},
"audios": {
".mp3",
".wav",
".aac",
".flac",
".m4a",
},
"videos": {
".mp4",
".mov",
".mkv",
".avi",
},
}
判断分类:
def get_category(
file_path: Path,
) -> str:
suffix = (
file_path.suffix.lower()
)
for category, suffixes in (
FILE_CATEGORIES.items()
):
if suffix in suffixes:
return category
return "others"
避免重名:
def get_unique_target(
target_path: Path,
) -> Path:
if not target_path.exists():
return target_path
parent = target_path.parent
stem = target_path.stem
suffix = target_path.suffix
index = 1
while True:
candidate = (
parent
/ f"{stem}_{index}{suffix}"
)
if not candidate.exists():
return candidate
index += 1
整理目录:
def organize_folder(
folder: Path,
) -> None:
if not folder.is_dir():
raise NotADirectoryError(
f"目录不存在:{folder}"
)
for file_path in list(
folder.iterdir()
):
if not file_path.is_file():
continue
category = get_category(
file_path
)
target_dir = (
folder
/ category
)
target_dir.mkdir(
parents=True,
exist_ok=True,
)
target_path = (
target_dir
/ file_path.name
)
target_path = get_unique_target(
target_path
)
shutil.move(
str(file_path),
str(target_path),
)
调用:
organize_folder(
Path("downloads")
)
批量移动文件属于破坏性操作,建议先增加预览模式。
def organize_folder(
folder: Path,
*,
dry_run: bool = True,
) -> None:
if not folder.is_dir():
raise NotADirectoryError(
f"目录不存在:{folder}"
)
for file_path in list(
folder.iterdir()
):
if not file_path.is_file():
continue
category = get_category(
file_path
)
target_dir = (
folder
/ category
)
target_path = get_unique_target(
target_dir
/ file_path.name
)
print(
f"{file_path.name} "
f"→ {target_path}"
)
if dry_run:
continue
target_dir.mkdir(
parents=True,
exist_ok=True,
)
shutil.move(
str(file_path),
str(target_path),
)
预览:
organize_folder(
Path("downloads"),
dry_run=True,
)
实际执行:
organize_folder(
Path("downloads"),
dry_run=False,
)
from datetime import datetime
from pathlib import Path
import shutil
def backup_file(
source: Path,
backup_dir: Path,
) -> Path:
if not source.is_file():
raise FileNotFoundError(
f"源文件不存在:{source}"
)
backup_dir.mkdir(
parents=True,
exist_ok=True,
)
timestamp = datetime.now().strftime(
"%Y%m%d_%H%M%S"
)
backup_name = (
f"{source.stem}_"
f"{timestamp}"
f"{source.suffix}"
)
target = (
backup_dir
/ backup_name
)
shutil.copy2(
source,
target,
)
return target
调用:
backup_path = backup_file(
Path("stocks.json"),
Path("backups"),
)
print(
f"备份完成:{backup_path}"
)
删除指定天数以前的日志:
from datetime import (
datetime,
timedelta,
)
from pathlib import Path
def delete_old_files(
folder: Path,
*,
pattern: str = "*.log",
days: int = 30,
dry_run: bool = True,
) -> list[Path]:
if days < 0:
raise ValueError(
"days 不能小于零"
)
cutoff = (
datetime.now()
- timedelta(days=days)
).timestamp()
deleted_files: list[Path] = []
for file_path in folder.rglob(
pattern
):
if not file_path.is_file():
continue
modified_time = (
file_path.stat().st_mtime
)
if modified_time >= cutoff:
continue
print(
f"删除:{file_path}"
)
if not dry_run:
file_path.unlink()
deleted_files.append(
file_path
)
return deleted_files
先预览:
delete_old_files(
Path("logs"),
days=30,
dry_run=True,
)
确认后执行:
delete_old_files(
Path("logs"),
days=30,
dry_run=False,
)
不推荐:
file = open("data.txt")
content = file.read()
推荐:
with open("data.txt") as file:
content = file.read()
open(
"important.txt",
"w",
)
会清空原文件。
如果不希望覆盖,可以使用:
"x"
或者先备份。
不推荐:
open(
"data.txt",
"r",
)
推荐:
open(
"data.txt",
"r",
encoding="utf-8",
)
不推荐:
path = (
folder
+ "/"
+ filename
)
推荐:
path = (
Path(folder)
/ filename
)
不推荐:
content = file.read()
推荐:
for line in file:
process(line)
危险:
shutil.rmtree(folder)
至少应该验证:
folder = folder.resolve()
if folder == Path.home():
raise ValueError(
"禁止删除用户主目录"
)
批量移动、删除、重命名时,建议先输出计划:
dry_run=True
确认无误后再执行。
相对路径实际上通常相对于:
Path.cwd()
需要稳定定位项目文件时,使用:
Path(__file__).resolve().parent
if path.exists():
不能确定路径是文件还是目录。
应按需求判断:
path.is_file()
path.is_dir()
不推荐:
try:
process_file()
except Exception:
pass
至少应记录:
import logging
logger = logging.getLogger(__name__)
try:
process_file()
except OSError:
logger.exception(
"文件处理失败"
)
Path("data") / "stocks.json"
比字符串路径更清晰。
encoding="utf-8"
with file_path.open(...) as file:
for line in file:
写临时文件
→ 写入成功
→ 原子替换
dry_run=True
except FileNotFoundError:
优于:
except Exception:
推荐:
data = load_data(path)
result = process_data(data)
save_data(output_path, result)
而不是把所有逻辑写在一个函数中。
from pathlib import Path
def read_text(
file_path: Path,
*,
encoding: str = "utf-8",
) -> str:
return file_path.read_text(
encoding=encoding,
)
def write_text(
file_path: Path,
content: str,
*,
encoding: str = "utf-8",
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
file_path.write_text(
content,
encoding=encoding,
)
import json
def read_json(
file_path: Path,
) -> object:
text = file_path.read_text(
encoding="utf-8",
)
return json.loads(text)
def write_json(
file_path: Path,
data: object,
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
text = json.dumps(
data,
ensure_ascii=False,
indent=2,
)
file_path.write_text(
text,
encoding="utf-8",
)
读取文本文件并统计字符数。
参考答案:
from pathlib import Path
def count_characters(
file_path: Path,
) -> int:
content = file_path.read_text(
encoding="utf-8",
)
return len(content)
统计文件中非空行数量。
def count_non_empty_lines(
file_path: Path,
) -> int:
count = 0
with file_path.open(
"r",
encoding="utf-8",
) as file:
for line in file:
if line.strip():
count += 1
return count
把列表中的内容写入文本文件,每个元素占一行。
def write_lines(
file_path: Path,
lines: list[str],
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
content = "\n".join(lines)
if lines:
content += "\n"
file_path.write_text(
content,
encoding="utf-8",
)
查找目录中所有 .json 文件。
def find_json_files(
folder: Path,
) -> list[Path]:
return sorted(
folder.rglob("*.json")
)
将一个文件复制到备份目录。
import shutil
def copy_to_backup(
source: Path,
backup_dir: Path,
) -> Path:
backup_dir.mkdir(
parents=True,
exist_ok=True,
)
target = (
backup_dir
/ source.name
)
shutil.copy2(
source,
target,
)
return target
读取 CSV 并转换价格字段。
import csv
def read_stocks_csv(
file_path: Path,
) -> list[dict[str, object]]:
stocks: list[
dict[str, object]
] = []
with file_path.open(
"r",
encoding="utf-8",
newline="",
) as file:
reader = csv.DictReader(file)
for row in reader:
stocks.append({
"code": row["code"],
"name": row["name"],
"price": float(
row["price"]
),
})
return stocks
删除目录中所有空文件。
def delete_empty_files(
folder: Path,
*,
dry_run: bool = True,
) -> list[Path]:
files: list[Path] = []
for file_path in folder.rglob("*"):
if not file_path.is_file():
continue
if file_path.stat().st_size != 0:
continue
print(
f"删除空文件:"
f"{file_path}"
)
if not dry_run:
file_path.unlink()
files.append(file_path)
return files
读取文本:
content = Path(
"data.txt"
).read_text(
encoding="utf-8",
)
写入文本:
Path(
"data.txt"
).write_text(
"内容",
encoding="utf-8",
)
逐行读取:
with Path(
"data.txt"
).open(
"r",
encoding="utf-8",
) as file:
for line in file:
print(line)
创建目录:
Path(
"data/output"
).mkdir(
parents=True,
exist_ok=True,
)
判断文件:
path.exists()
path.is_file()
path.is_dir()
遍历目录:
folder.iterdir()
folder.glob("*.txt")
folder.rglob("*.txt")
删除文件:
path.unlink(
missing_ok=True,
)
复制:
shutil.copy2(
source,
target,
)
移动:
shutil.move(
source,
target,
)
删除目录:
shutil.rmtree(
folder
)
读取 JSON:
data = json.loads(
path.read_text(
encoding="utf-8",
)
)
写入 JSON:
path.write_text(
json.dumps(
data,
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
Python 文件操作最常用的组合是:
from pathlib import Path
配合:
with file_path.open(...)
处理文本文件时:
小文件 → read_text、write_text
大文件 → 逐行读取
二进制文件 → rb、wb
表格文本 → csv
结构化数据 → json
文件复制移动 → shutil
临时文件 → tempfile
实际开发中应重点注意:
明确文件编码
使用 with 自动关闭
避免误用 w 覆盖文件
大文件不要一次性读取
批量删除移动先预览
重要数据使用安全写入
路径优先使用 pathlib
合理处理文件异常
文件操作的本质是:
确定路径
→ 打开资源
→ 读取或写入
→ 校验结果
→ 关闭资源
→ 处理异常
掌握文件操作以后,就可以进一步完成:
批量文件整理
日志分析
JSON 数据处理
CSV 数据转换
Excel 自动化
网页数据保存
股票资料整理
数据库导入导出
自动备份系统