青少年创意编程

Appearance

第9章 文件操作 ​

学习目标 ​

完成本章学习后，读者将能够：

  • 掌握文件读写的基本操作与上下文管理器模式
  • 熟练使用pathlib进行跨平台路径操作
  • 理解文本文件与二进制文件的区别及编码处理
  • 运用JSON、CSV、pickle进行数据序列化与反序列化
  • 自定义上下文管理器管理资源生命周期

9.1 文件基础 ​

9.1.1 打开与关闭文件 ​

python
# 手动管理（不推荐）
file = open("example.txt", "r", encoding="utf-8")
content = file.read()
file.close()

# with语句（推荐）- 自动关闭文件，即使发生异常
with open("example.txt", "r", encoding="utf-8") as file:
    content = file.read()

工程实践：始终使用with语句操作文件。它保证即使发生异常也能正确关闭文件，避免资源泄漏。手动调用close()在异常发生时可能不会执行。

9.1.2 文件模式 ​

模式描述文件存在时文件不存在时
r只读读取FileNotFoundError
w只写清空创建
x独占写FileExistsError创建
a追加追加到末尾创建
r+读写读取FileNotFoundError
b二进制模式--
t文本模式（默认）--
python
# 写入文件
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")
    f.write("第二行\n")

# 追加内容
with open("output.txt", "a", encoding="utf-8") as f:
    f.write("追加内容\n")

# 二进制模式
with open("data.bin", "wb") as f:
    f.write(b"\x00\x01\x02\x03")
with open("data.bin", "rb") as f:
    data = f.read()

9.1.3 文件指针 ​

python
with open("example.txt", "r", encoding="utf-8") as f:
    print(f.tell())          # 0 - 当前位置
    
    content = f.read(10)     # 读取10个字符
    print(f.tell())          # 10
    
    f.seek(0)                # 回到开头
    f.seek(5)                # 移到第5个字符

9.2 读写文件 ​

9.2.1 读取文件 ​

python
# 读取全部内容
with open("example.txt", "r", encoding="utf-8") as f:
    content = f.read()

# 逐行读取（推荐，内存友好）
with open("example.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())

# 读取所有行为列表
with open("example.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()

# 读取指定字节数
with open("example.txt", "r", encoding="utf-8") as f:
    chunk = f.read(1024)     # 读取1KB

工程实践：处理大文件时，始终使用逐行迭代for line in f而非f.read()或f.readlines()。后者将整个文件加载到内存，可能导致内存溢出。

9.2.2 写入文件 ​

python
# 写入字符串
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello, World!\n")

# 写入多行
with open("output.txt", "w", encoding="utf-8") as f:
    lines = ["第一行\n", "第二行\n", "第三行\n"]
    f.writelines(lines)

# 使用print写入
with open("output.txt", "w", encoding="utf-8") as f:
    print("Hello, World!", file=f)
    print("Python文件操作", file=f)

9.2.3 大文件复制 ​

python
# 分块复制（内存友好）
def copy_file(src: str, dst: str, chunk_size: int = 8192) -> None:
    with open(src, "rb") as src_file, open(dst, "wb") as dst_file:
        while chunk := src_file.read(chunk_size):
            dst_file.write(chunk)

# 使用shutil（推荐）
import shutil
shutil.copy2("source.bin", "destination.bin")  # 保留元数据

9.3 路径操作 ​

9.3.1 pathlib（推荐） ​

python
from pathlib import Path

# 创建路径
path = Path("home") / "user" / "documents" / "file.txt"

# 路径组件
print(path.name)       # "file.txt"
print(path.stem)       # "file"
print(path.suffix)     # ".txt"
print(path.parent)     # Path("home/user/documents")

# 绝对路径
print(Path("file.txt").resolve())

# 存在性检测
print(path.exists())
print(path.is_file())
print(path.is_dir())

# 文件信息
print(path.stat().st_size)    # 文件大小

# 创建与删除
Path("new_dir").mkdir(parents=True, exist_ok=True)
path.unlink()                  # 删除文件
path.rename("new_name.txt")    # 重命名

# 便捷读写
content = Path("file.txt").read_text(encoding="utf-8")
Path("output.txt").write_text("Hello!\n", encoding="utf-8")

# 文件搜索
for py_file in Path(".").glob("*.py"):
    print(py_file)
for py_file in Path(".").rglob("*.py"):    # 递归搜索
    print(py_file)

工程实践：优先使用pathlib而非os.path。pathlib面向对象、支持/运算符拼接路径、API更一致，是Python 3.4+的标准做法。

9.3.2 os.path（旧式） ​

python
import os

path = "/home/user/documents/file.txt"
print(os.path.basename(path))     # "file.txt"
print(os.path.dirname(path))      # "/home/user/documents"
print(os.path.join("a", "b", "c"))  # "a/b/c"（跨平台）
print(os.path.splitext(path))     # ("/home/user/documents/file", ".txt")
print(os.path.exists(path))

9.4 目录操作 ​

9.4.1 遍历目录 ​

python
from pathlib import Path

# 列出当前目录
for item in Path(".").iterdir():
    print(f"{'[D]' if item.is_dir() else '[F]'} {item.name}")

# 递归遍历
for root, dirs, files in os.walk("."):
    for name in files:
        print(Path(root) / name)

# 按模式搜索
for py_file in Path("src").rglob("*.py"):
    print(py_file)

9.4.2 创建与删除 ​

python
from pathlib import Path
import shutil

# 创建目录
Path("new_dir").mkdir()
Path("a/b/c").mkdir(parents=True, exist_ok=True)  # 递归创建，已存在不报错

# 删除
Path("file.txt").unlink()        # 删除文件
Path("empty_dir").rmdir()        # 删除空目录
shutil.rmtree("non_empty_dir")   # 删除非空目录

9.5 序列化 ​

9.5.1 JSON ​

python
import json
from dataclasses import dataclass, asdict
from datetime import datetime

data = {"name": "Alice", "age": 25, "skills": ["Python", "SQL"]}

# 序列化
json_str = json.dumps(data, ensure_ascii=False, indent=2)
with open("data.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

# 反序列化
parsed = json.loads(json_str)
with open("data.json", "r", encoding="utf-8") as f:
    data = json.load(f)

# 自定义序列化
def json_serializer(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"Type {type(obj)} not serializable")

json_str = json.dumps({"time": datetime.now()}, default=json_serializer)

# dataclass序列化
@dataclass
class Person:
    name: str
    age: int

person = Person("Alice", 25)
json_str = json.dumps(asdict(person))

9.5.2 CSV ​

python
import csv

# 写入CSV
data = [
    {"name": "Alice", "age": 25, "city": "北京"},
    {"name": "Bob", "age": 30, "city": "上海"},
]
with open("data.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["name", "age", "city"])
    writer.writeheader()
    writer.writerows(data)

# 读取CSV
with open("data.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row["name"], row["age"])

9.5.3 pickle ​

python
import pickle

data = {"name": "Alice", "time": datetime.now()}

# 序列化到文件
with open("data.pkl", "wb") as f:
    pickle.dump(data, f)

# 从文件反序列化
with open("data.pkl", "rb") as f:
    loaded = pickle.load(f)

# 序列化到字节
pickled = pickle.dumps(data)
loaded = pickle.loads(pickled)

安全警告：pickle可以执行任意代码，绝不要加载不受信任来源的pickle数据。在Web应用中应始终使用JSON而非pickle。

9.5.4 格式对比 ​

特性JSONCSVpickle
数据类型基本类型纯文本表格任意Python对象
可读性高高不可读
安全性安全安全不安全
跨语言是是否
适用场景API、配置表格数据Python内部缓存

9.6 上下文管理器 ​

9.6.1 自定义上下文管理器 ​

python
# 基于类的实现
class Timer:
    def __init__(self, name: str = ""):
        self.name = name
    
    def __enter__(self):
        import time
        self.start = time.perf_counter()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        elapsed = time.perf_counter() - self.start
        print(f"{self.name} 执行时间: {elapsed:.4f}秒")
        return False

with Timer("计算"):
    sum(range(1000000))

# 基于生成器的实现
from contextlib import contextmanager

@contextmanager
def timer(name: str = ""):
    import time
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{name} 执行时间: {elapsed:.4f}秒")

with timer("计算"):
    sum(range(1000000))

9.6.2 实用上下文管理器 ​

python
from contextlib import contextmanager
import os

@contextmanager
def change_dir(path: str):
    old_dir = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(old_dir)

@contextmanager
def temp_file(content: str, suffix: str = ".txt"):
    import tempfile
    with tempfile.NamedTemporaryFile(mode="w", suffix=suffix, delete=False, encoding="utf-8") as f:
        f.write(content)
        name = f.name
    try:
        yield name
    finally:
        os.unlink(name)

9.7 IO模型与性能 ​

9.7.1 同步与异步IO ​

python
import time
import asyncio
import aiofiles

def sync_io_example():
    """同步IO示例"""
    start = time.time()
    
    with open("file1.txt", "w") as f:
        f.write("content1")
    with open("file2.txt", "w") as f:
        f.write("content2")
    with open("file3.txt", "w") as f:
        f.write("content3")
    
    print(f"同步IO耗时: {time.time() - start:.4f}秒")

async def async_io_example():
    """异步IO示例（需要aiofiles库）"""
    start = time.time()
    
    async with aiofiles.open("file1.txt", "w") as f:
        await f.write("content1")
    async with aiofiles.open("file2.txt", "w") as f:
        await f.write("content2")
    async with aiofiles.open("file3.txt", "w") as f:
        await f.write("content3")
    
    print(f"异步IO耗时: {time.time() - start:.4f}秒")

学术注记：Python的文件IO默认是阻塞式同步IO。对于高并发场景（如Web服务器处理大量文件请求），应使用异步IO（asyncio + aiofiles）或线程池。异步IO利用操作系统提供的非阻塞接口，单线程即可处理多个IO操作。

9.7.2 缓冲与性能 ​

python
import time

def benchmark_buffer():
    """缓冲区大小对性能的影响"""
    data = b"x" * 10_000_000
    
    with open("test.bin", "wb", buffering=0) as f:
        start = time.time()
        f.write(data)
        print(f"无缓冲: {time.time() - start:.4f}秒")
    
    with open("test.bin", "wb", buffering=8192) as f:
        start = time.time()
        f.write(data)
        print(f"8KB缓冲: {time.time() - start:.4f}秒")
    
    with open("test.bin", "wb", buffering=65536) as f:
        start = time.time()
        f.write(data)
        print(f"64KB缓冲: {time.time() - start:.4f}秒")

benchmark_buffer()

9.7.3 内存映射文件 ​

python
import mmap

def mmap_example():
    """内存映射文件：将文件映射到内存"""
    
    with open("large_file.bin", "wb") as f:
        f.write(b"\x00" * 10_000_000)
    
    with open("large_file.bin", "r+b") as f:
        mm = mmap.mmap(f.fileno(), 0)
        
        print(f"文件大小: {len(mm)} 字节")
        
        mm[0:4] = b"TEST"
        
        mm.seek(100)
        mm.write(b"Hello")
        
        mm.close()

def mmap_search():
    """在大型文件中搜索"""
    with open("large_file.txt", "r", encoding="utf-8") as f:
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            index = mm.find(b"search_term")
            if index != -1:
                print(f"找到于位置: {index}")

学术注记：内存映射文件（mmap）将文件直接映射到进程的虚拟内存空间，操作系统负责在内存和磁盘间交换数据。适用于：1）处理超大文件（超过内存容量）；2）随机访问文件内容；3）多进程共享内存通信。

9.7.4 文件系统监控 ​

python
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class FileChangeHandler(FileSystemEventHandler):
    def on_created(self, event):
        print(f"文件创建: {event.src_path}")
    
    def on_modified(self, event):
        print(f"文件修改: {event.src_path}")
    
    def on_deleted(self, event):
        print(f"文件删除: {event.src_path}")

def watch_directory(path: str):
    """监控目录变化"""
    observer = Observer()
    observer.schedule(FileChangeHandler(), path, recursive=True)
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

9.8 前沿技术动态 ​

9.8.1 异步文件IO ​

python
import asyncio
import aiofiles

async def read_file_async(path: str) -> str:
    async with aiofiles.open(path, mode='r', encoding='utf-8') as f:
        return await f.read()

async def write_file_async(path: str, content: str) -> None:
    async with aiofiles.open(path, mode='w', encoding='utf-8') as f:
        await f.write(content)

async def main():
    content = await read_file_async("data.txt")
    await write_file_async("output.txt", content.upper())

asyncio.run(main())

9.8.2 现代路径操作 ​

python
from pathlib import Path

# Python 3.12+ 新增方法
p = Path("data/output.txt")
p.parent.mkdir(parents=True, exist_ok=True)

# 相对路径计算
base = Path("/home/user/project")
target = Path("/home/user/data/file.txt")
relative = target.relative_to(base)

# 路径匹配
for py_file in Path(".").glob("**/*.py"):
    print(py_file)

9.8.3 高性能序列化 ​

python
import orjson
import msgspec

data = {"name": "Alice", "age": 30, "items": [1, 2, 3]}

json_bytes = orjson.dumps(data)
loaded = orjson.loads(json_bytes)

encoder = msgspec.json.Encoder()
decoder = msgspec.json.Decoder()
encoded = encoder.encode(data)
decoded = decoder.decode(encoded)

9.8.4 文件系统监控 ​

python
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        print(f"Modified: {event.src_path}")
    
    def on_created(self, event):
        print(f"Created: {event.src_path}")

observer = Observer()
observer.schedule(MyHandler(), path=".", recursive=True)
observer.start()

9.9 本章小结 ​

本章系统介绍了Python文件操作的完整体系：

  1. 文件读写：with语句、文本/二进制模式、逐行读取
  2. 路径操作：pathlib面向对象路径处理（推荐）
  3. 目录操作：遍历、创建、删除、复制
  4. 序列化：JSON（跨语言安全）、CSV（表格数据）、pickle（Python内部）
  5. 上下文管理器：资源管理的标准模式，支持类与生成器两种实现
  6. IO模型：同步/异步IO、缓冲策略、内存映射文件

9.9.1 文件操作最佳实践 ​

场景推荐方案原因
普通文件读写with + open()自动资源管理
路径操作pathlib.Path面向对象、跨平台
配置文件JSON/YAML可读性好、跨语言
大文件处理逐行迭代/mmap内存友好
高并发IOasyncio + aiofiles非阻塞、高效
临时文件tempfile模块安全、自动清理

9.9.2 常见陷阱与规避 ​

python
# 陷阱1：忘记关闭文件
f = open("file.txt", "r")
content = f.read()
# 忘记f.close()！

# 正确做法
with open("file.txt", "r") as f:
    content = f.read()

# 陷阱2：编码错误
with open("file.txt", "r") as f:  # 可能UnicodeDecodeError
    content = f.read()

# 正确做法
with open("file.txt", "r", encoding="utf-8") as f:
    content = f.read()

# 陷阱3：大文件内存溢出
content = open("large.txt").read()  # 全部加载到内存！

# 正确做法
with open("large.txt") as f:
    for line in f:
        process(line)

# 陷阱4：pickle安全风险
data = pickle.loads(untrusted_data)  # 可能执行恶意代码！

# 正确做法：使用JSON
data = json.loads(untrusted_data)

9.10 练习题 ​

基础题 ​

  1. 编写程序，统计文本文件中的行数、单词数和字符数。

  2. 使用JSON存储和读取学生信息列表。

  3. 实现一个简单的日志记录器，支持按日期分割日志文件。

进阶题 ​

  1. 实现文件搜索工具，在指定目录中递归搜索包含特定内容的文件。

  2. 编写程序，合并多个CSV文件并去重。

  3. 实现INI格式配置文件管理器，支持读写和类型转换。

项目实践 ​

  1. 文件同步工具：编写一个程序，要求：
    • 比较两个目录的文件差异
    • 支持增量同步（仅复制修改过的文件）
    • 使用文件哈希（MD5/SHA256）判断文件是否相同
    • 支持排除规则（如忽略.git目录）
    • 生成同步报告
    • 使用pathlib处理路径

思考题 ​

  1. 为什么with语句比手动close()更安全？__exit__方法的返回值有什么作用？

  2. JSON和pickle的核心区别是什么？为什么pickle数据不安全？

  3. pathlib相比os.path有哪些优势？在什么场景下仍需使用os.path？

9.11 延伸阅读 ​

9.11.1 文件系统与IO ​

  • 《操作系统概念》 (Silberschatz等) — 文件系统与IO原理
  • 《UNIX环境高级编程》 (W. Richard Stevens) — 文件IO系统调用
  • Python IO文档 (https://docs.python.org/3/library/io.html) — Python IO层次结构

9.11.2 路径与文件操作 ​

9.11.3 序列化与数据格式 ​

9.11.4 异步IO ​

下一章：第10章 类与对象

青少年创意编程 - 高中Python组 - 江苏省宿城中等专业学校

Copyright © 2026 陈瑾瑜