6  文件路径与数据持久化

点击下载本章节代码

Note本章学习目标

• 掌握现代Python文件路径操作（pathlib）
• 学会高效的文件读写方法
• 了解Python对象序列化与反序列化
• 掌握大规模数据文件处理技巧

6.1 环境准备

%pip install joblib pandas openpyxl pyarrow fastparquet tqdm --upgrade

Tip为什么使用 pathlib？

Python 3.4+ 引入的 pathlib.Path 相比传统的 os.path 有诸多优势：

• 面向对象的API，更直观易用
• 跨平台兼容性更好（自动处理Windows/Linux/Mac路径差异）
• 支持链式调用，代码更简洁
• 集成了常用文件操作方法（读写、复制、删除等）

6.2 路径操作基础

from pathlib import Path
import pandas as pd
from tqdm import tqdm

Path.cwd()

PosixPath('/Users/xinlu/_project/AI4MPAcc-Course')

file = Path("./data/example.txt")
type(file)

pathlib.PosixPath

file.exists()

True

file.absolute()

PosixPath('/Users/xinlu/_project/AI4MPAcc-Course/data/example.txt')

6.3 路径拆分

file = file.absolute()
file

PosixPath('/Users/xinlu/_project/AI4MPAcc-Course/data/example.txt')

file.parts

('/', 'Users', 'xinlu', '_project', 'AI4MPAcc-Course', 'data', 'example.txt')

file.is_dir()

False

file.is_file()

True

file.name

'example.txt'

file.stem

'example'

file.suffix

'.txt'

file.parent

PosixPath('/Users/xinlu/_project/AI4MPAcc-Course/data')

list(file.parents)

[PosixPath('/Users/xinlu/_project/AI4MPAcc-Course/data'),
 PosixPath('/Users/xinlu/_project/AI4MPAcc-Course'),
 PosixPath('/Users/xinlu/_project'),
 PosixPath('/Users/xinlu'),
 PosixPath('/Users'),
 PosixPath('/')]

6.4 文本文件读写

TipPath 的便捷读写方法

pathlib.Path 提供了简洁的文本读写方法：

• Path.read_text() - 读取整个文件为字符串
• Path.write_text() - 将字符串写入文件
• Path.read_bytes() - 读取二进制文件
• Path.write_bytes() - 写入二进制数据

这些方法自动处理文件的打开和关闭，比传统的 open() 更简洁。

# 创建目录（如果不存在）
Path("./data/").mkdir(parents=True, exist_ok=True)

# 写入文本文件（推荐始终指定 encoding="utf-8"）
file.write_text('暨南', encoding="utf-8")

2

# 读取文本文件
file.read_text(encoding="utf-8")

'暨南'

# 编码错误演示：用错误的编码读取
try:
    file.read_text(encoding="gbk")
except Exception as e:
    print(f"编码错误: {e}")

# 删除文件
file.unlink()

Warning字符编码陷阱

在处理中文文本时，务必：

1. 写入时明确指定 encoding="utf-8"
2. 读取时使用相同编码
3. 避免使用 gbk 或系统默认编码（容易产生乱码）

推荐统一使用 UTF-8 编码。

try:
    file.write_text('暨南大学管理学院', encoding="utf-8")
    # 尝试用错误的编码读取会报错
    print(file.read_text(encoding="gbk"))
except Exception as e:
    print(f"错误: {e}")
finally:
    # 正确读取
    print(file.read_text(encoding="utf-8"))

错误: 'gbk' codec can't decode byte 0xad in position 10: illegal multibyte sequence
暨南大学管理学院

6.5 文件重命名

file.with_suffix(".json")

PosixPath('/Users/xinlu/_project/AI4MPAcc-Course/data/example.json')

file.rename("./data/example_new_name.txt")

PosixPath('data/example_new_name.txt')

print(file)

file.exists()

/Users/xinlu/_project/AI4MPAcc-Course/data/example.txt

False

Path("./data/example_new_name.txt").rename("./data/example.txt")

PosixPath('data/example.txt')

6.6 COPY/MOVE/DELETE 文件/文件夹

Path("./make/new/folder/").mkdir(parents=True, exist_ok=True)

import shutil
shutil.copy(file, Path("./make/new/folder/"))

'make/new/folder/example.txt'

new_file = file.with_suffix(".json")
new_file

PosixPath('/Users/xinlu/_project/AI4MPAcc-Course/data/example.json')

shutil.copyfile(file, new_file)

PosixPath('/Users/xinlu/_project/AI4MPAcc-Course/data/example.json')

shutil.move(new_file, Path("./make/new/folder/"))

'make/new/folder/example.json'

shutil.copytree("./make", "./data/make_copy")

'./data/make_copy'

shutil.rmtree("./make", ignore_errors=True)
shutil.rmtree("./data/make_copy", ignore_errors=True)

6.7 文件搜索与批量处理

Noteglob vs rglob

• glob(pattern) - 只搜索当前目录
• rglob(pattern) - 递归搜索所有子目录（r = recursive）

常用通配符： - * - 匹配任意字符 - ** - 匹配任意级别的目录 - ? - 匹配单个字符 - [abc] - 匹配括号内的任一字符

import os

# 获取当前Python环境路径
env_path = os.environ.get('CONDA_PREFIX', '.')
print(f"环境路径: {env_path}")

环境路径: /Users/xinlu/miniconda3

# 搜索当前目录下的所有 .py 文件（不含子目录）
py_files = list(Path(env_path).glob('*.py'))
print(f"找到 {len(py_files)} 个 .py 文件")
py_files[:3]  # 显示前3个

找到 0 个 .py 文件

[]

# 递归搜索所有子目录中的 .py 文件
all_py_files = list(Path(env_path).rglob('*.py'))
print(f"递归搜索找到 {len(all_py_files)} 个 .py 文件")
all_py_files[:3]

递归搜索找到 48916 个 .py 文件

[PosixPath('/Users/xinlu/miniconda3/bin/pdf2txt.py'),
 PosixPath('/Users/xinlu/miniconda3/bin/dumppdf.py'),
 PosixPath('/Users/xinlu/miniconda3/bin/pwiz.py')]

# 批量复制图片文件示例
png_files = Path(env_path).rglob('*.png')

# 创建目标文件夹
output_dir = Path("./data/collected_images/")
output_dir.mkdir(parents=True, exist_ok=True)

# 使用 tqdm 显示进度条
from shutil import copy2

for i, img_path in enumerate(tqdm(list(png_files)[:10], desc="复制图片")):
    if img_path.is_file():
        # 为避免文件名冲突，添加序号
        new_name = f"{i:03d}_{img_path.name}"
        copy2(img_path, output_dir / new_name)

print(f"已复制到: {output_dir.absolute()}")

已复制到: /Users/xinlu/_project/AI4MPAcc-Course/data/collected_images

# 筛选文件和目录
items = list(Path(env_path).glob('*'))

files = [x for x in items if x.is_file()]
dirs = [x for x in items if x.is_dir()]

print(f"文件数量: {len(files)}")
print(f"目录数量: {len(dirs)}")
print(f"目录示例: {dirs[:3]}")

文件数量: 6
目录数量: 15
目录示例: [PosixPath('/Users/xinlu/miniconda3/man'), PosixPath('/Users/xinlu/miniconda3/conda-meta'), PosixPath('/Users/xinlu/miniconda3/condabin')]

6.8 保存和加载 Python 单个对象

import joblib
# 创建一个示例对象，例如一个简单的字典
data = {'name': 'Alice', 'age': 25, 'city': 'New York'}

# 使用 joblib 保存对象到文件
joblib.dump(data, 'data.joblib')

['data.joblib']

# 加载保存的对象
loaded_data = joblib.load('data.joblib')
loaded_data

{'name': 'Alice', 'age': 25, 'city': 'New York'}

Path('data.joblib').unlink()

6.9 保存和加载 Python 多个对象

import joblib

# 创建多个示例对象
model = {'type': 'LinearRegression', 'parameters': {'intercept': 0.5, 'slope': 2.3}}
config = {'batch_size': 32, 'epochs': 100}
data = [1, 2, 3, 4, 5]

# 使用 joblib 保存多个对象到一个文件
joblib.dump((model, config, data), 'multiple_objects.joblib')

# 加载保存的多个对象
loaded_model, loaded_config, loaded_data = joblib.load('multiple_objects.joblib')

print(loaded_model)
print(loaded_config)
print(loaded_data)

{'type': 'LinearRegression', 'parameters': {'intercept': 0.5, 'slope': 2.3}}
{'batch_size': 32, 'epochs': 100}
[1, 2, 3, 4, 5]

Path('multiple_objects.joblib').unlink()

. . .

• 把数据转化为DataFrame，然后利用pandas的强大IO接口存为 xlsx, csv 等数据

6.10 DataFrame拆分例子

from dfply import diamonds

Path('./temp/').mkdir()

clarity_set = set(diamonds.clarity.tolist())
clarity_set

{'I1', 'IF', 'SI1', 'SI2', 'VS1', 'VS2', 'VVS1', 'VVS2'}

for c in clarity_set:
  df_subset = diamonds[diamonds.clarity == c]
  df_subset.to_excel(f'./temp/clarity_{c}.xlsx', index = False)
  print(f'clarity = {c} save completed!')

clarity = SI1 save completed!
clarity = VVS2 save completed!
clarity = I1 save completed!
clarity = VS2 save completed!
clarity = IF save completed!
clarity = VVS1 save completed!
clarity = VS1 save completed!
clarity = SI2 save completed!

6.11 DataFrame读取例子

file_list = [x for x in Path('./temp/').glob("*.xlsx")]
file_list

[PosixPath('temp/clarity_VVS1.xlsx'),
 PosixPath('temp/clarity_SI2.xlsx'),
 PosixPath('temp/clarity_VS1.xlsx'),
 PosixPath('temp/clarity_I1.xlsx'),
 PosixPath('temp/clarity_IF.xlsx'),
 PosixPath('temp/clarity_VS2.xlsx'),
 PosixPath('temp/clarity_SI1.xlsx'),
 PosixPath('temp/clarity_VVS2.xlsx')]

import pandas as pd
df_list = [pd.read_excel(x) for x in file_list]
df_list[0].head(2)

	carat	cut	color	clarity	depth	table	price	x	y	z
0	0.24	Very Good	I	VVS1	62.3	57.0	336	3.95	3.98	2.47
1	0.32	Ideal	I	VVS1	62.0	55.3	553	4.39	4.42	2.73

df_all = pd.concat(df_list)
df_all.shape

(53940, 10)

import shutil
shutil.rmtree("./temp/", ignore_errors=True)