Skip to content

第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 知识图谱

9.9.1 Python IO体系

Python IO类型层次结构
┌─────────────────────────────────────────────────────────────┐
│                    IOBase (抽象基类)                         │
├─────────────────────────────────────────────────────────────┤
│  close(), closed, fileno(), flush(), readable(),            │
│  seekable(), writable()                                     │
└─────────────────────────────────────────────────────────────┘

        ┌───────────┼───────────┐
        ▼           ▼           ▼
┌───────────┐ ┌───────────┐ ┌───────────┐
│ RawIOBase │ │BufferedIOBase│ │ TextIOBase│
│ (原始IO)  │ │ (缓冲IO)    │ │ (文本IO)  │
├───────────┤ ├─────────────┤ ├───────────┤
│ read()    │ │ read()      │ │ read()    │
│ write()   │ │ write()     │ │ write()   │
│ readinto()│ │ read1()     │ │ readline()│
│           │ │ readinto()  │ │ encoding  │
└───────────┘ └─────────────┘ └───────────┘
        │           │           │
        ▼           ▼           ▼
   FileIO      BufferedReader  TextIOWrapper
   (系统调用)   BufferedWriter  (编码转换)
               BufferedRandom

文件打开模式:
┌─────────────────────────────────────────┐
│ r   只读(默认)                        │
│ w   只写(清空)                        │
│ x   独占创建                            │
│ a   追加                                │
│ b   二进制模式                          │
│ +   读写模式                            │
│ t   文本模式(默认)                    │
└─────────────────────────────────────────┘

9.9.2 pathlib路径操作

pathlib.Path 核心操作

路径构建:
┌─────────────────────────────────────────┐
│ Path("file.txt")                        │
│ Path("dir") / "subdir" / "file.txt"     │
│ Path.home() / "documents"               │
│ Path.cwd()                              │
└─────────────────────────────────────────┘

路径属性:
┌─────────────────────────────────────────┐
│ .name        文件名(含扩展名)          │
│ .stem        文件名(不含扩展名)        │
│ .suffix      扩展名                     │
│ .suffixes    所有扩展名列表             │
│ .parent      父目录                     │
│ .parents     所有父目录序列             │
│ .parts       路径组件元组               │
│ .anchor      根路径(/或C:\)           │
└─────────────────────────────────────────┘

路径方法:
┌─────────────────────────────────────────┐
│ .exists()        是否存在               │
│ .is_file()       是否文件               │
│ .is_dir()        是否目录               │
│ .resolve()       解析为绝对路径         │
│ .absolute()      绝对路径               │
│ .relative_to()   相对路径               │
│ .with_suffix()   更改扩展名             │
│ .with_name()     更改文件名             │
└─────────────────────────────────────────┘

文件操作:
┌─────────────────────────────────────────┐
│ .read_text()     读取文本文件           │
│ .write_text()    写入文本文件           │
│ .read_bytes()    读取二进制文件         │
│ .write_bytes()   写入二进制文件         │
│ .mkdir()         创建目录               │
│ .rmdir()         删除空目录             │
│ .unlink()        删除文件               │
│ .rename()        重命名                 │
│ .iterdir()       遍历目录               │
│ .glob()          模式匹配               │
│ .rglob()         递归模式匹配           │
└─────────────────────────────────────────┘

9.9.3 上下文管理器协议

上下文管理器协议

with语句执行流程:
┌─────────────────────────────────────────┐
│ with ContextManager() as target:        │
│     # 代码块                            │
└─────────────────────────────────────────┘


┌─────────────────────────────────────────┐
│ 1. 调用 __enter__()                     │
│    返回值绑定到 target                   │
└─────────────────────────────────────────┘


┌─────────────────────────────────────────┐
│ 2. 执行代码块                           │
└─────────────────────────────────────────┘


┌─────────────────────────────────────────┐
│ 3. 调用 __exit__(exc_type, exc_val, tb) │
│    - 正常结束: 参数全为None             │
│    - 异常发生: 参数包含异常信息         │
│    - 返回True: 抑制异常                 │
│    - 返回False: 传播异常                │
└─────────────────────────────────────────┘

实现方式:
┌─────────────────────────────────────────┐
│ 方式1: 类实现                           │
│ class MyContext:                        │
│     def __enter__(self): ...            │
│     def __exit__(self, *args): ...      │
├─────────────────────────────────────────┤
│ 方式2: 生成器 + contextlib              │
│ @contextmanager                         │
│ def my_context():                       │
│     # setup                             │
│     try:                                │
│         yield value                     │
│     finally:                            │
│         # cleanup                       │
└─────────────────────────────────────────┘

9.10 技术选型指南

9.10.1 文件读写选型

场景推荐方案原因
普通文本文件with + open()自动资源管理
大文件处理逐行迭代内存友好
二进制文件open(..., "rb")避免编码问题
配置文件json/yaml可读性、跨语言
日志文件logging模块功能完整
临时文件tempfile模块安全、自动清理

9.10.2 路径操作选型

场景推荐方案原因
新代码pathlib.Path面向对象、跨平台
遗留代码os.path兼容性
路径拼接Path / operator简洁直观
批量操作Path.glob()内置模式匹配
系统调用os模块底层控制

9.10.3 序列化选型

场景推荐方案原因
跨语言数据交换JSON标准、安全
配置文件YAML/JSON可读性
Python内部序列化pickle功能强大
安全敏感数据JSONpickle不安全
大数据存储msgpack/protobuf高效二进制

9.10.4 文件操作方法选择

需求推荐方法替代方案
读取全部Path.read_text()open().read()
写入全部Path.write_text()open().write()
逐行读取for line in freadlines()
追加内容open(..., "a")seek + write
文件存在检查Path.exists()os.path.exists()
创建目录Path.mkdir(parents=True)os.makedirs()

9.11 常见问题与解决方案

9.11.1 文件编码问题

python
# 问题:读取文件编码错误
with open("file.txt", "r") as f:  # UnicodeDecodeError
    content = f.read()

# 解决方案1:指定编码
with open("file.txt", "r", encoding="utf-8") as f:
    content = f.read()

# 解决方案2:错误处理策略
with open("file.txt", "r", encoding="utf-8", errors="replace") as f:
    content = f.read()

# 解决方案3:自动检测编码
import chardet
with open("file.txt", "rb") as f:
    raw = f.read()
    result = chardet.detect(raw)
    content = raw.decode(result["encoding"])

9.11.2 大文件内存溢出

python
# 问题:一次性读取大文件
content = open("large.txt").read()  # 内存溢出!

# 解决方案1:逐行处理
with open("large.txt") as f:
    for line in f:
        process(line)

# 解决方案2:分块读取
def read_chunks(f, chunk_size=8192):
    while chunk := f.read(chunk_size):
        yield chunk

with open("large.txt") as f:
    for chunk in read_chunks(f):
        process(chunk)

# 解决方案3:内存映射
import mmap
with open("large.txt", "r+") as f:
    mm = mmap.mmap(f.fileno(), 0)
    # 可以像操作内存一样操作文件
    mm.close()

9.11.3 文件路径跨平台问题

python
import os
from pathlib import Path

# 问题:硬编码路径分隔符
path = "dir\\subdir\\file.txt"  # Windows only!

# 解决方案1:使用pathlib
path = Path("dir") / "subdir" / "file.txt"

# 解决方案2:使用os.path.join
path = os.path.join("dir", "subdir", "file.txt")

# 问题:路径中的反斜杠转义
path = "C:\new\test.txt"  # \n被转义!

# 解决方案:使用原始字符串或pathlib
path = r"C:\new\test.txt"
path = Path("C:/new/test.txt")  # pathlib自动处理

9.11.4 文件锁与并发问题

python
import fcntl
import portalocker

# 问题:多进程同时写入文件导致数据混乱

# 解决方案1:使用fcntl(Unix)
def write_with_lock(filepath, content):
    with open(filepath, "a") as f:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        try:
            f.write(content)
        finally:
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)

# 解决方案2:使用portalocker(跨平台)
def write_with_portalocker(filepath, content):
    with portalocker.Lock(filepath, "a") as f:
        f.write(content)

# 解决方案3:使用临时文件+原子重命名
import os
def atomic_write(filepath, content):
    temp_path = filepath + ".tmp"
    with open(temp_path, "w") as f:
        f.write(content)
    os.replace(temp_path, filepath)  # 原子操作

9.11.5 pickle安全问题

python
import pickle
import json

# 问题:pickle反序列化可能执行恶意代码
# data = pickle.loads(untrusted_data)  # 危险!

# 解决方案1:使用JSON(推荐)
data = json.loads(untrusted_data)

# 解决方案2:限制pickle的类
import pickle
class RestrictedUnpickler(pickle.Unpickler):
    def find_class(self, module, name):
        if module == "builtins" and name in {"str", "int", "list", "dict"}:
            return getattr(builtins, name)
        raise pickle.UnpicklingError(f"unsafe: {module}.{name}")

def safe_loads(data):
    return RestrictedUnpickler(io.BytesIO(data)).load()

# 解决方案3:使用更安全的序列化格式
import msgpack
data = msgpack.unpackb(untrusted_data, raw=False)

9.12 本章小结

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

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

9.12.1 文件操作最佳实践

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

9.12.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.13 练习题

基础题

  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.14 延伸阅读

9.14.1 文件系统与IO

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

9.14.2 路径与文件操作

9.14.3 序列化与数据格式

9.14.4 异步IO


下一章:第10章 类与对象

Python技术丛书 - 江苏省宿城中等专业学校