Skip to main content
Septvean's Documents
Toggle Dark/Light/Auto mode Toggle Dark/Light/Auto mode Toggle Dark/Light/Auto mode Back to homepage

文件操作

一、文件操作是什么

文件操作主要包括:

读取文件
写入文件
追加内容
复制文件
移动文件
删除文件
创建目录
遍历目录
批量处理文件

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

三、使用 open 打开文件

基本格式:

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 自动关闭文件

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)

八、按行读取文件

1. readline

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")

2. readlines

一次读取所有行:

with open(
    "data.txt",
    "r",
    encoding="utf-8",
) as file:
    lines = file.readlines()

print(lines)

结果类似:

[
    "第一行\n",
    "第二行\n",
    "第三行\n",
]

readlines() 会把整个文件加载到内存。

不适合特别大的文件。


3. 直接遍历文件

推荐按行读取:

with open(
    "data.txt",
    "r",
    encoding="utf-8",
) as file:
    for line in file:
        print(line.rstrip("\n"))

这种方式:

  • 每次只读取一行
  • 内存占用较小
  • 适合大文件

4. 获取行号

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")

如果文件不存在,会创建文件。

如果文件已经存在,会清空原内容后重新写入。


1. 写入多行

with open(
    "output.txt",
    "w",
    encoding="utf-8",
) as file:
    file.write("第一行\n")
    file.write("第二行\n")
    file.write("第三行\n")

write() 不会自动添加换行符。

需要手动写入:

"\n"

2. write 的返回值

write() 返回写入的字符数量:

with open(
    "output.txt",
    "w",
    encoding="utf-8",
) as file:
    count = file.write("Python")

print(count)

结果:

6

十、使用 writelines 写入多行

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()

1. errors 参数

忽略无法解码的字符:

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() 返回当前文件位置。


1. seek 移动文件指针

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 后文件已经关闭。


十七、使用 pathlib

导入:

from pathlib import Path

创建路径对象:

file_path = Path("data.txt")

路径对象不是文件内容,而是文件或目录路径的表示。


1. 当前工作目录

from pathlib import Path

current_directory = Path.cwd()

print(current_directory)

2. 用户主目录

home_directory = Path.home()

print(home_directory)

3. 拼接路径

推荐使用 /

file_path = (
    Path("data")
    / "stocks"
    / "600519.json"
)

不推荐手动拼接:

file_path = (
    "data/"
    + "stocks/"
    + "600519.json"
)

pathlib 会处理不同操作系统的路径分隔符。


4. 查看路径信息

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)

5. 多重扩展名

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)

二十一、创建目录

1. 创建单层目录

folder = Path("data")

folder.mkdir()

如果目录已存在,会报错:

FileExistsError

允许目录已存在:

folder.mkdir(
    exist_ok=True,
)

2. 创建多层目录

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()


二十七、遍历目录

1. iterdir

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)

2. glob

查找当前目录下所有 TXT 文件:

for file_path in folder.glob(
    "*.txt"
):
    print(file_path)

查找 JSON:

folder.glob("*.json")

查找特定开头:

folder.glob("stock_*.csv")

3. rglob

递归查找所有子目录:

for file_path in folder.rglob(
    "*.txt"
):
    print(file_path)

等价于类似:

当前目录以及所有子目录中的 TXT 文件

4. 使用 glob 递归

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,
)

二十九、文件信息 stat

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)

1. 格式化文件大小

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"

1. pathlib 读取二进制

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)

1. 添加统一前缀

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)

2. 按顺序编号

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 文件操作

CSV 是常用表格文本格式。

文件示例:

code,name,price
600519,贵州茅台,1500.0
000001,平安银行,11.5

Python 使用:

import csv

1. 读取 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 中读取的数据默认都是字符串。


2. 跳过表头

with open(
    "stocks.csv",
    "r",
    encoding="utf-8",
    newline="",
) as file:
    reader = csv.reader(file)

    header = next(reader)

    for row in reader:
        print(row)

3. 使用 DictReader

更推荐按字段名读取:

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",
}

4. 写入 CSV

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)

5. 使用 DictWriter

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)

6. Excel 打开中文 CSV 乱码

某些 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)

三十七、JSON 文件操作

导入:

import json

JSON 示例:

[
  {
    "code": "600519",
    "name": "贵州茅台",
    "price": 1500.0
  }
]

1. 读取 JSON 文件

import json


with open(
    "stocks.json",
    "r",
    encoding="utf-8",
) as file:
    data = json.load(file)

print(data)

2. 写入 JSON 文件

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             使用两个空格缩进

3. pathlib 写 JSON

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)

4. JSON 异常处理

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

大量数据可以使用 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

1. 临时文件

import tempfile


with tempfile.NamedTemporaryFile(
    mode="w+",
    encoding="utf-8",
    suffix=".txt",
) as file:
    file.write("临时内容")
    file.seek(0)

    print(file.read())

离开 with 后,临时文件通常会自动删除。


2. 临时目录

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)

1. 分块读取二进制文件

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))

2. 分块复制文件

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",
)

如果文件较大,可以逐行处理并写入新文件。


1. 逐行替换大文件

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",
)

1. 使用 zipfile

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())

1. 获取当前脚本目录

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

五十一、不要使用 exists 后再操作代替异常处理

下面的代码存在竞态条件:

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,
)

五十九、文件操作常见错误

1. 忘记关闭文件

不推荐:

file = open("data.txt")
content = file.read()

推荐:

with open("data.txt") as file:
    content = file.read()

2. 使用 w 模式误覆盖文件

open(
    "important.txt",
    "w",
)

会清空原文件。

如果不希望覆盖,可以使用:

"x"

或者先备份。


3. 忘记指定编码

不推荐:

open(
    "data.txt",
    "r",
)

推荐:

open(
    "data.txt",
    "r",
    encoding="utf-8",
)

4. 拼接路径使用字符串

不推荐:

path = (
    folder
    + "/"
    + filename
)

推荐:

path = (
    Path(folder)
    / filename
)

5. 大文件一次性读入内存

不推荐:

content = file.read()

推荐:

for line in file:
    process(line)

6. 删除前不检查路径

危险:

shutil.rmtree(folder)

至少应该验证:

folder = folder.resolve()

if folder == Path.home():
    raise ValueError(
        "禁止删除用户主目录"
    )

7. 批量操作没有预览模式

批量移动、删除、重命名时,建议先输出计划:

dry_run=True

确认无误后再执行。


8. 假设相对路径相对于脚本

相对路径实际上通常相对于:

Path.cwd()

需要稳定定位项目文件时,使用:

Path(__file__).resolve().parent

9. 只检查 exists

if path.exists():

不能确定路径是文件还是目录。

应按需求判断:

path.is_file()
path.is_dir()

10. 隐藏所有异常

不推荐:

try:
    process_file()
except Exception:
    pass

至少应记录:

import logging

logger = logging.getLogger(__name__)

try:
    process_file()
except OSError:
    logger.exception(
        "文件处理失败"
    )

六十、文件操作设计原则

原则一:优先使用 Path

Path("data") / "stocks.json"

比字符串路径更清晰。

原则二:文本文件明确指定编码

encoding="utf-8"

原则三:使用 with 管理文件

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,
    )

读取 JSON

import json


def read_json(
    file_path: Path,
) -> object:
    text = file_path.read_text(
        encoding="utf-8",
    )

    return json.loads(text)

写入 JSON

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 自动化
网页数据保存
股票资料整理
数据库导入导出
自动备份系统