Python 文件 IO 编程指南

第一阶段：基础操作

1.1 文件打开与关闭

f = open('example.txt', 'w', encoding='utf-8')
f.write('Hello, Python 文件操作！\n')
f.write('第二行内容')
f.close()

关键说明：

• 模式 'w' 会覆盖原有内容，请谨慎使用

• encoding='utf-8' 防止中文乱码（Windows 默认编码为 GBK）

• 忘记调用 close() 可能导致数据未写入磁盘或资源泄漏

1.2 读取文件

# read()：一次性读取全部内容（适合小文件）
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())

# readline()：手动逐行读取
with open('example.txt', 'r', encoding='utf-8') as f:
    line = f.readline()
    while line:
        print(line.strip())
        line = f.readline()

# readlines()：读取为列表（适合中等大小文件）
with open('example.txt', 'r', encoding='utf-8') as f:
    lines = f.readlines()
    for idx, line in enumerate(lines, 1):
        print(f"{idx}: {line.strip()}")

原则：大文件用迭代器逐行读取；小文件才用 read() 一次性读入。

1.3 写入操作

# 覆盖写入
with open('output.txt', 'w', encoding='utf-8') as f:
    f.write('单行内容\n')
    f.writelines(['第一行\n', '第二行\n', '第三行\n'])

# 追加写入（不覆盖原有内容）
with open('log.txt', 'a', encoding='utf-8') as f:
    f.write('新日志\n')

1.4 异常处理

文件操作是与外部系统的交互，必须处理 I/O 异常：

try:
    with open('data.txt', 'r', encoding='utf-8') as f:
        content = f.read()
except FileNotFoundError:
    print('文件不存在')
except PermissionError:
    print('没有读取权限')
except IsADirectoryError:
    print('路径指向目录而非文件')
except UnicodeDecodeError:
    print('编码不匹配，尝试指定正确的 encoding 参数')

常见异常：

异常	触发条件
FileNotFoundError	文件不存在（'r' 或 'r+' 模式）
FileExistsError	文件已存在（'x' 模式）
PermissionError	无读写权限
IsADirectoryError	路径是目录而非文件
UnicodeDecodeError	文件编码与指定的 encoding 不符

阶段 1 练习

• 创建 data.txt，写入 5 行内容，逐行读取并打印行号

• 追加 3 行新内容，再次读取验证

• 故意读取不存在的文件，捕获并处理 FileNotFoundError

第二阶段：模式、上下文管理与文件管理

2.1 with 语句的本质

with 语句保证文件在正常退出和异常退出时都能被关闭：

# 不用 with（危险：异常时可能跳过 close）
f = open('test.txt', 'r')
try:
    data = f.read()
finally:
    f.close()   # 必须手动保证关闭

# 用 with（等价于上面的 try-finally，更简洁）
with open('test.txt', 'r') as f:
    data = f.read()
# 离开 with 块自动关闭，即使有异常

with 语句的相关原理请见：python-context-manager

2.2 打开模式

模式	含义	文件不存在时	是否覆盖
'r'	只读	报错	—
'w'	只写	创建	覆盖
'a'	追加	创建	追加
'x'	独占创建	—	文件已存在时报错
'r+'	读写	报错	从头覆盖写入
'w+'	写读	创建	覆盖
'a+'	追加读	创建	追加

# x 模式：安全创建，防止覆盖已有文件
try:
    with open('important.txt', 'x', encoding='utf-8') as f:
        f.write('新建的安全文件')
except FileExistsError:
    print('文件已存在，未覆盖')

2.3 二进制文件操作

# 复制图片（二进制模式）
with open('source.jpg', 'rb') as src:
    data = src.read()
with open('copy.jpg', 'wb') as dst:
    dst.write(data)

# 大二进制文件：分块复制，避免内存溢出
def copy_large_file(src, dst, chunk_size=8192):
    with open(src, 'rb') as f_src, open(dst, 'wb') as f_dst:
        while True:
            chunk = f_src.read(chunk_size)
            if not chunk:
                break
            f_dst.write(chunk)

2.4 文件指针

文件对象内部维护一个指针，记录当前读写位置：

with open('data.txt', 'r+', encoding='utf-8') as f:
    print(f.tell())    # 0（文件开头）
    f.read(5)
    print(f.tell())    # 5（向后移动了 5 字节）
    f.seek(0)          # 回到开头
    f.seek(0, 2)       # 跳到末尾
    size = f.tell()    # 获取文件大小（字节数）

文件指针机制的详细说明见 python-file-pointer-chunked-reading。

2.5 路径处理

使用 pathlib.Path 构建跨平台路径：

from pathlib import Path

# / 运算符拼接，自动适配平台分隔符
config = Path('data') / '2024' / 'config.json'

config.parent.mkdir(parents=True, exist_ok=True)

with open(config, 'r', encoding='utf-8') as f:
    content = f.read()

路径处理的完整参考见 python-path-handling。

2.6 文件管理：shutil 与 os

shutil 和 os 模块提供文件系统操作：

import shutil
import os

# 复制文件（连同权限元数据）
shutil.copy('src.txt', 'dst.txt')
shutil.copy2('src.txt', 'dst.txt')    # 额外保留时间戳等元数据

# 复制整个目录树
shutil.copytree('src_dir', 'dst_dir')

# 移动文件或目录
shutil.move('old_path', 'new_path')

# 删除目录树（不可恢复，谨慎使用）
shutil.rmtree('dir_to_delete')

# 删除单个文件
os.remove('file.txt')

# 重命名
os.rename('old.txt', 'new.txt')

# 递归遍历目录树
for root, dirs, files in os.walk('project'):
    for filename in files:
        filepath = os.path.join(root, filename)
        print(filepath)

更多内容请见标准库 Day 1：文件与路径处理。

阶段 2 练习

• 用 'a+' 模式打开文件，既追加新内容又读取全部内容

• 用 'x' 模式实现安全创建函数：文件已存在则提示，不存在才创建

• 用 shutil.copytree 备份一个目录，再用 os.walk 统计其中的文件总数

第三阶段：结构化数据

3.1 CSV 文件处理

import csv

# 写入 CSV
with open('data.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f)
    writer.writerow(['姓名', '年龄', '城市'])
    writer.writerows([
        ['张三', 25, '北京'],
        ['李四', 30, '上海'],
        ['王五', 28, '广州'],
    ])

# 读取 CSV（DictReader 按列名访问，推荐）
with open('data.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['姓名']}住在{row['城市']}，今年{row['年龄']}岁")

3.2 JSON 文件处理

import json

data = {
    'name': '产品A',
    'price': 99.9,
    'tags': ['电子', '热销'],
    'stock': None
}

# 写入（ensure_ascii=False 保留中文，indent=2 美化输出）
with open('config.json', 'w', encoding='utf-8') as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

# 读取
with open('config.json', 'r', encoding='utf-8') as f:
    config = json.load(f)
    print(config['name'])   # 产品A

3.3 大文件处理

# 逐行迭代：内存占用最低
def process_large_text_file(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        for line_num, line in enumerate(f, 1):
            process(line_num, line.strip())

# 分块读取二进制：适合 GB 级文件
def read_in_chunks(filepath, chunk_size=1024 * 1024):  # 每次 1 MB
    with open(filepath, 'rb') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

for chunk in read_in_chunks('large_video.mp4'):
    process_chunk(chunk)

阶段 3 练习

• 用 pathlib 批量将目录下所有 .txt 重命名为 .bak

• 读取 CSV，筛选出年龄 > 25 的行，写入新 CSV

• 创建配置类，用 JSON 持久化保存用户设置，支持读取和更新

第四阶段：高级与实战

4.1 临时文件与内存文件

import tempfile
from io import StringIO, BytesIO

# 临时文件（退出 with 块后自动删除）
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=True) as tmp:
    tmp.write('临时数据')
    tmp.flush()
    print(f'临时文件路径：{tmp.name}')

# StringIO：在内存中像操作文件一样操作字符串
output = StringIO()
output.write('第一行\n')
output.write('第二行\n')
content = output.getvalue()
output.close()

# BytesIO：内存中的二进制缓冲区
buf = BytesIO(b'\x00\x01\x02\x03')
print(buf.read())

4.2 文件锁（多进程安全写入）

# pip install portalocker
import portalocker

# 防止多个进程同时写入同一文件导致数据混乱
with open('shared.log', 'a') as f:
    portalocker.lock(f, portalocker.LOCK_EX)   # 独占锁
    f.write('关键日志\n')
    portalocker.unlock(f)

4.3 内存映射（超大文件随机访问）

import mmap

# 将文件映射到内存地址空间，不将整个文件载入 Python 堆
with open('huge_file.bin', 'r+b') as f:
    with mmap.mmap(f.fileno(), 0) as mm:   # 0 表示映射整个文件
        print(mm[:100])        # 读取前 100 字节
        mm[0:5] = b'HELLO'    # 修改后直接同步到磁盘

4.4 异步文件 IO

# pip install aiofiles
import aiofiles
import asyncio

async def async_copy(src, dst):
    async with aiofiles.open(src, 'rb') as f_src:
        async with aiofiles.open(dst, 'wb') as f_dst:
            while True:
                chunk = await f_src.read(8192)
                if not chunk:
                    break
                await f_dst.write(chunk)

asyncio.run(async_copy('source.bin', 'dest.bin'))

4.5 实战：按日期轮转的日志处理器

from datetime import datetime
from pathlib import Path

class DailyRotatingLogger:
    """按日期轮转日志，自动创建日期子目录。"""
    def __init__(self, base_dir='logs'):
        self.base_path = Path(base_dir)
        self.base_path.mkdir(exist_ok=True)
        self.current_file = None
        self.current_date = None

    def _get_file(self):
        today = datetime.now().strftime('%Y-%m-%d')
        if today != self.current_date:
            if self.current_file:
                self.current_file.close()
            self.current_date = today
            daily_dir = self.base_path / today
            daily_dir.mkdir(exist_ok=True)
            self.current_file = open(daily_dir / 'app.log', 'a', encoding='utf-8')
        return self.current_file

    def log(self, message, level='INFO'):
        timestamp = datetime.now().strftime('%H:%M:%S')
        self._get_file().write(f'[{timestamp}] {level}: {message}\n')
        self._get_file().flush()   # 立即刷盘，防止崩溃丢失日志

    def close(self):
        if self.current_file:
            self.current_file.close()

logger = DailyRotatingLogger()
logger.log('系统启动')
logger.log('用户登录', level='DEBUG')
logger.close()

掌握清单

• 能用 with open 正确读写文本和二进制文件

• 理解 r / w / a / x / r+ 各模式的行为差异

• 能捕获并处理常见文件 I/O 异常

• 使用 pathlib.Path 构建跨平台路径

• 用 shutil 完成文件和目录的复制、移动、删除

• 熟练处理 CSV 和 JSON 格式文件

• 掌握大文件分块读取，控制内存占用

• 理解文件指针，能用 seek() / tell() 进行定位