第三十六篇：文件读写——open 与 with 语句详解

前言

文件操作是程序员的基本功。Python 提供了简洁而强大的文件 I/O API，通过 open() 函数和 with 语句可以安全、高效地读写各种格式的文件。本文将系统讲解 Python 文件读写的核心知识。

open() 函数基础

基本语法

# open(file, mode='r', buffering=-1, encoding=None)

# 最简单的打开方式
f = open("hello.txt", "r")
content = f.read()
f.close()

文件打开模式

模式	名称	说明
'r'	read	只读（默认），文件不存在则报错
'w'	write	只写，文件存在则清空，不存在则创建
'a'	append	追加写入，文件不存在则创建
'x'	create	独占创建，文件已存在则报错
'b'	binary	二进制模式，与其他模式组合使用（如 'rb'）
't'	text	文本模式（默认）
'+'	update	更新模式，可读可写（如 'r+', 'w+'）

常见组合示例

# 文本模式
open("file.txt", "r")     # 读取文本文件（默认）
open("file.txt", "w")     # 写入文本文件（覆盖）
open("file.txt", "a")     # 追加到文本文件末尾
open("file.txt", "r+")    # 读取和写入文本文件

# 二进制模式
open("image.png", "rb")   # 读取二进制文件
open("image.png", "wb")   # 写入二进制文件
open("data.bin", "ab")    # 追加到二进制文件

# 文本 + 更新模式
open("file.txt", "w+")    # 清空后读写
open("file.txt", "a+")    # 追加并允许读取

读取文件

多种方式读取内容

# 方法一：read() — 一次性读取全部内容
with open("sample.txt", "r", encoding="utf-8") as f:
    content = f.read()
print(content)  # 整个文件内容作为字符串返回

# 方法二：read(size) — 读取指定字节数
with open("sample.txt", "r", encoding="utf-8") as f:
    chunk = f.read(100)  # 最多读取 100 个字符
print(chunk)

# 方法三：readline() — 逐行读取（包含换行符）
with open("sample.txt", "r", encoding="utf-8") as f:
    line = f.readline()  # 读取第一行
    print(line.rstrip())
    line = f.readline()  # 读取第二行
    print(line.rstrip())

# 方法四：readlines() — 读取所有行为列表
with open("sample.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()
print(lines)  # ['第一行\n', '第二行\n', '第三行']

# 方法五：迭代器（最推荐，节省内存）
with open("sample.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.rstrip())

实战：统计文件信息

def analyze_file(filepath):
    """分析文件的基本信息"""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            content = f.read()
        
        return {
            "filename": filepath,
            "total_chars": len(content),
            "total_lines": content.count("\n") + (1 if content and not content.endswith("\n") else 0),
            "word_count": len(content.split()),
            "longest_line_len": max((len(line) for line in content.split("\n")), default=0),
            "first_word": content.split()[0] if content.split() else None,
        }
    except FileNotFoundError:
        return {"error": f"文件不存在: {filepath}"}
    except UnicodeDecodeError:
        return {"error": f"编码错误: {filepath}（尝试用二进制模式读取）"}

info = analyze_file("sample.txt")
for key, value in info.items():
    print(f"{key}: {value}")
# filename: sample.txt
# total_chars: 1024
# total_lines: 50
# word_count: 256
# longest_line_len: 42
# first_word: Hello

处理大文件

import os

def process_large_file(filepath, chunk_size=8192):
    """分块处理大文件，避免占用过多内存"""
    file_size = os.path.getsize(filepath)
    bytes_read = 0
    
    with open(filepath, "r", encoding="utf-8") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            
            bytes_read += len(chunk.encode("utf-8"))
            progress = (bytes_read / file_size) * 100
            print(f"\r处理进度: {progress:.1f}%", end="", flush=True)
            
            # 处理每个数据块...
            for line in chunk.split("\n"):
                pass  # 你的业务逻辑
    
    print("\n完成!")

# 对于超大文件，可以使用生成器逐行处理
def iter_lines(filepath, encoding="utf-8"):
    """生成器：逐行读取大文件"""
    with open(filepath, "r", encoding=encoding) as f:
        for line in f:
            yield line

# 用法
for line in iter_lines("huge_file.csv"):
    process(line)  # 内存友好！

写入文件

不同写入方式

# 方法一：write() — 写入字符串
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")
    f.write("第二行\n")

# 方法二：writelines() — 写入多行（不自动添加换行）
lines = ["苹果\n", "香蕉\n", "橙子\n"]
with open("fruits.txt", "w", encoding="utf-8") as f:
    f.writelines(lines)

# ⚠️ writelines 不会自动添加换行符！
with open("no_newline.txt", "w") as f:
    f.writelines(["a", "b", "c"])  # 结果是 "abc" 而非 "a\nb\nc\n"

# 方法三：格式化写入
temperature = [23.5, 25.1, 22.8, 26.3, 24.7]
city = "北京"

with open(f"{city}_weather.txt", "w", encoding="utf-8") as f:
    f.write(f"=== {city} 天气预报 ===\n\n")
    f.write(f"{'日期':<10} {'温度(°C)':>10} {'天气':>8}\n")
    f.write("-" * 30 + "\n")
    
    dates = ["周一", "周二", "周三", "周四", "周五"]
    weathers = ["晴", "多云", "阴", "小雨", "晴"]
    
    for date, temp, weather in zip(dates, temperature, weathers):
        f.write(f"{date:<10} {temp:>9.1f} {weather:>8}\n")
    
    f.write("\n平均气温: {:.1f}°C\n".format(sum(temperature) / len(temperature)))

CSV 数据导出

def export_to_csv(data, filepath):
    """导出数据为 CSV 文件"""
    if not data:
        raise ValueError("数据不能为空")
    
    headers = list(data[0].keys())
    
    with open(filepath, "w", encoding="utf-8-sig", newline="") as f:
        # 写入表头
        f.write(",".join(headers) + "\n")
        
        # 写入数据行
        for row in data:
            values = []
            for header in headers:
                value = str(row.get(header, ""))
                # 如果值中包含逗号或引号，需要用引号包裹
                if "," in value or '"' in value:
                    value = '"' + value.replace('"', '""') + '"'
                values.append(value)
            f.write(",".join(values) + "\n")

users = [
    {"name": "张三", "age": 25, "email": "zhang@example.com"},
    {"name": "李四", "age": 30, "email": "li@example.com"},
    {"name": "王五, \\"经理\\"", "age": 35, "email": "wang@example.com"},
]

export_to_csv(users, "users.csv")

异常处理与安全性

完整的文件操作模板

import os

def safe_read_file(filepath, encoding="utf-8"):
    """安全地读取文件"""
    try:
        if not os.path.exists(filepath):
            raise FileNotFoundError(f"文件不存在: {filepath}")
        
        file_size = os.path.getsize(filepath)
        if file_size > 100 * 1024 * 1024:  # 限制 100MB
            raise FileSizeError(f"文件过大: {filepath} ({file_size} bytes)")
        
        with open(filepath, "r", encoding=encoding) as f:
            content = f.read()
        
        return content
    
    except FileNotFoundError:
        print(f"❌ 文件未找到: {filepath}")
        return None
    except PermissionError:
        print(f"❌ 没有读取权限: {filepath}")
        return None
    except UnicodeDecodeError as e:
        print(f"❌ 编码错误: {e}\n请使用正确的编码格式读取此文件")
        return None
    except OSError as e:
        print(f"❌ 系统错误: {e}")
        return None

检查文件状态

import os

def check_file(filepath):
    """全面检查文件状态"""
    checks = {}
    
    checks["exists"] = os.path.exists(filepath)
    
    if checks["exists"]:
        stat = os.stat(filepath)
        checks["is_file"] = os.path.isfile(filepath)
        checks["is_dir"] = os.path.isdir(filepath)
        checks["size"] = stat.st_size
        checks["created"] = stat.st_ctime
        checks["modified"] = stat.st_mtime
        checks["permissions"] = oct(stat.st_mode)[-3:]
        
        # 判断是否可读写
        checks["readable"] = os.access(filepath, os.R_OK)
        checks["writable"] = os.access(filepath, os.W_OK)
    
    else:
        checks["is_file"] = False
        checks["is_dir"] = False
        checks["size"] = 0
        checks["created"] = None
        checks["modified"] = None
        checks["permissions"] = None
        checks["readable"] = False
        checks["writable"] = False
    
    return checks

info = check_file("sample.txt")
for key, value in info.items():
    print(f"{key}: {value}")

with 语句 —— 上下文管理器

为什么需要 with？

# ❌ 传统方式容易忘记关闭文件
f = open("file.txt", "r")
data = f.read()
# 如果中间抛出异常，f.close() 不会执行！
return data
f.close()

# ✅ with 语句自动关闭文件
with open("file.txt", "r") as f:
    data = f.read()
# 离开 with 块后文件自动关闭，即使发生异常也会关闭
return data

with 的内部机制

class MyFile:
    """自定义上下文管理器"""
    
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None
    
    def __enter__(self):
        """进入 with 块时调用"""
        self.file = open(self.filename, self.mode, encoding="utf-8")
        return self.file  # 这个返回值会绑定到 as 变量
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """离开 with 块时调用"""
        if exc_type:
            print(f"发生了异常: {exc_val}")
        # 无论是否有异常，都关闭文件
        if self.file:
            self.file.close()
        # 返回 False（或 None）表示不抑制异常
        # 返回 True 则会吞掉异常
        return False

# 使用自定义上下文管理器
with MyFile("test.txt", "w") as f:
    f.write("Hello from context manager!\n")

print(f.closed)  # True ← 文件已自动关闭

多个文件同时管理

# 可以同时管理多个资源
with open("source.txt", "r", encoding="utf-8") as src, \
     open("destination.txt", "w", encoding="utf-8") as dst:
    
    content = src.read()
    dst.write(content.upper())
    print("文件内容已复制到目标文件并转为大写")

# 注意：只要有一个资源打开失败，后面的都不会继续
# with open("nonexistent.txt", "r") as f1, \
#      open("other.txt", "w") as f2:
#     # 这里只有第一个成功的话第二个才会打开

二进制文件操作

图片复制

def copy_image(src, dest):
    """复制图片文件"""
    with open(src, "rb") as fin, open(dest, "wb") as fout:
        while True:
            chunk = fin.read(8192)  # 每次读取 8KB
            if not chunk:
                break
            fout.write(chunk)
    
    import os
    src_size = os.path.getsize(src)
    dest_size = os.path.getsize(dest)
    print(f"复制完成: {src_size} → {dest_size} bytes")

copy_image("original.png", "copy.png")

JSON 序列化存储

import json

def save_config(config_dict, filepath="config.json"):
    """保存配置到 JSON 文件"""
    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(config_dict, f, ensure_ascii=False, indent=2)
    print(f"配置已保存到 {filepath}")

def load_config(filepath="config.json"):
    """从 JSON 文件加载配置"""
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            config = json.load(f)
        return config
    except FileNotFoundError:
        print(f"配置文件不存在，使用默认配置")
        return get_default_config()
    except json.JSONDecodeError as e:
        print(f"JSON 解析错误: {e}")
        return get_default_config()

def get_default_config():
    return {
        "app_name": "MyApp",
        "version": "1.0.0",
        "debug": False,
        "database": {
            "host": "localhost",
            "port": 5432,
            "name": "mydb"
        },
        "features": ["auth", "logging", "cache"]
    }

# 保存配置
save_config({
    "app_name": "MyApp Pro",
    "version": "2.0.0",
    "debug": True,
    "database": {"host": "prod-db.example.com", "port": 5432, "name": "production"},
    "features": ["auth", "logging", "cache", "monitoring"],
})

# 加载配置
config = load_config()
print(f"应用: {config['app_name']} v{config['version']}")

序列化的完整示例

import pickle

def serialize_data(data, filepath):
    """使用 pickle 序列化数据"""
    with open(filepath, "wb") as f:
        pickle.dump(data, f, protocol=pickle.HIGHEST_PROTOCOL)

def deserialize_data(filepath):
    """反序列化数据"""
    with open(filepath, "rb") as f:
        return pickle.load(f)

# 保存复杂数据结构
data = {
    "users": [("小明", 25, "admin"), ("小红", 28, "editor")],
    "scores": [95, 88, 92, 76, 100],
    "metadata": {"created": "2026-09-09", "count": 5},
}

serialize_data(data, "data.pkl")
loaded = deserialize_data("data.pkl")
print(loaded)

路径操作

os.path vs pathlib

from pathlib import Path

# pathlib 是现代推荐的路径处理方式
base = Path(".")

# 创建路径
script_path = Path(__file__).resolve()
parent_dir = script_path.parent
current_dir = Path.cwd()

# 检查文件/目录
target = Path("data/reports/report_2026.txt")

if target.exists():
    print(f"文件大小: {target.stat().st_size} bytes")
    print(f"修改时间: {target.stat().st_mtime}")
elif target.parent.exists():
    print("父目录存在，准备创建文件")
    target.parent.mkdir(parents=True, exist_ok=True)

# 路径拼接（推荐使用 / 运算符）
path = Path("data") / "reports" / "report_2026.txt"
print(path)  # data\reports\report_2026.txt

# 遍历目录
for item in Path(".").iterdir():
    print(f"{item.name} - {'文件夹' if item.is_dir() else '文件'}")

# 获取文件后缀和名称
print(target.suffix)     # .txt
print(target.stem)       # report_2026
print(target.name)       # report_2026.txt

# glob 通配符匹配
for csv_file in Path("data").glob("*.csv"):
    print(f"找到 CSV 文件: {csv_file.name}")

# recursively 递归查找
for all_files in Path(".").rglob("*.py"):
    print(all_files)

完整的文件工具类

from pathlib import Path
import shutil

class FileTools:
    """常用文件操作工具"""
    
    @staticmethod
    def read_text(filepath, encoding="utf-8"):
        """读取文本文件"""
        path = Path(filepath)
        if not path.exists():
            raise FileNotFoundError(f"文件不存在: {filepath}")
        return path.read_text(encoding=encoding)
    
    @staticmethod
    def write_text(filepath, content, encoding="utf-8"):
        """写入文本文件"""
        path = Path(filepath)
        # 确保父目录存在
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding=encoding)
    
    @staticmethod
    def append_text(filepath, content, encoding="utf-8"):
        """追加文本到文件"""
        path = Path(filepath)
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(path.read_text(encoding=encoding) + content, encoding=encoding)
    
    @staticmethod
    def copy_file(src, dest):
        """复制文件"""
        src_path = Path(src)
        dest_path = Path(dest)
        dest_path.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(src_path, dest_path)
        print(f"已复制: {src} → {dest}")
    
    @staticmethod
    def count_lines(filepath):
        """统计文件行数"""
        path = Path(filepath)
        with path.open("r", encoding="utf-8") as f:
            return sum(1 for _ in f)
    
    @staticmethod
    def find_files(directory, pattern="*"):
        """查找目录下匹配的文件"""
        return list(Path(directory).glob(pattern))

# 使用示例
FileTools.write_text("output/hello.txt", "Hello World!")
content = FileTools.read_text("output/hello.txt")
print(content)  # Hello World!

lines = FileTools.count_lines("output/hello.txt")
print(f"行数: {lines}")

最佳实践总结

✅ 推荐做法

1. 始终使用 with 语句 — 确保文件正确关闭
2. 显式指定 encoding — 始终使用 encoding="utf-8" 避免平台差异
3. 大文件使用迭代 — for line in f: 比 readlines() 节省内存
4. 正确处理异常 — 区分 FileNotFoundError、PermissionError 等具体异常
5. 优先使用 pathlib — 比 os.path 更直观现代
6. 二进制操作指定 'b' 模式 — 图片、音频等非文本文件必须用二进制模式

❌ 避免做法

# ❌ 不使用 with
f = open("file.txt")
data = f.read()
# 如果抛出异常，f.close() 永远不会执行
return data

# ❌ 不指定编码
f = open("file.txt")  # Windows 下可能是 gbk 编码
# ✅ 始终指定 UTF-8
f = open("file.txt", encoding="utf-8")

# ❌ 用 open 写大文件
with open("big.log") as f:
    all_lines = f.readlines()  # 可能耗尽内存

# ✅ 用迭代器逐行处理
with open("big.log") as f:
    for line in f:
        process(line)

总结

本文系统讲解了 Python 文件读写的核心知识：

1. open() 函数 — 参数、模式选择和文件句柄操作
2. 读取文件 — read/readline/readlines/迭代器四种方式
3. 写入文件 — write/writelines/CSV 导出
4. 异常处理 — 安全的文件操作模板
5. with 语句 — 上下文管理器自动释放资源
6. 二进制文件 — 图片复制、JSON/pickle 序列化
7. pathlib — 现代路径操作和文件系统工具
8. 最佳实践 — 编码声明、with 语句、内存优化

下一篇文章将深入讲解 StringIO 与 BytesIO。

---

本文是 Python 教程系列的第三十六篇。