30天学会Python编程:9. Python文件与IO操作
9.1 文件操作基础
9.1.1 文件操作流程
9.1.2 文件打开模式
表9-1 Python文件打开模式
模式 | 描述 | 文件存在 | 文件不存在 |
'r' | 只读 | 正常打开 | 报错 |
'w' | 写入 | 清空内容 | 创建新文件 |
'x' | 独占创建 | 报错 | 创建新文件 |
'a' | 追加 | 追加写入 | 创建新文件 |
'b' | 二进制模式 | 配合使用 | - |
't' | 文本模式(默认) | 配合使用 | - |
'+' | 读写模式 | 配合使用 | - |
9.2 文件读写操作
9.2.1 基本读写方法
文件对象主要方法：
file.read(size) # 读取size字节内容
file.readline() # 读取一行
file.readlines() # 读取所有行到列表
file.write(string) # 写入字符串
file.writelines() # 写入字符串序列
file.seek(offset) # 移动文件指针
file.tell() # 返回当前指针位置
9.2.2 文本文件操作
# 传统写法
try:
f = open('data.txt', 'r', encoding='utf-8')
content = f.read()
finally:
f.close()
# 推荐写法(上下文管理器)
with open('data.txt', 'r', encoding='utf-8') as f:
lines = f.readlines()
for line in lines:
print(line.strip())
9.2.3 二进制文件操作
# 图片复制示例
with open('input.jpg', 'rb') as src, open('output.jpg', 'wb') as dst:
while True:
chunk = src.read(1024) # 分块读取
if not chunk:
break
dst.write(chunk)
9.3 上下文管理器
9.3.1 with语句原理
9.3.2 自定义上下文管理器
class DatabaseConnection:
"""数据库连接上下文管理器"""
def __enter__(self):
print("建立数据库连接")
return self # 返回资源对象
def __exit__(self, exc_type, exc_val, exc_tb):
print("关闭数据库连接")
if exc_type: # 处理异常
print(f"发生错误: {exc_val}")
return True # 抑制异常
# 使用示例
with DatabaseConnection() as conn:
print("执行数据库操作")
# raise ValueError("模拟错误") # 测试异常处理
9.4 常见文件格式处理
9.4.1 CSV文件
import csv
# 读取CSV
with open('data.csv', 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
print(row['name'], row['age'])
# 写入CSV
data = [{'name': 'Alice', 'age': 25}, {'name': 'Bob', 'age': 30}]
with open('output.csv', 'w', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=['name', 'age'])
writer.writeheader()
writer.writerows(data)
9.4.2 JSON文件
import json
# 写入JSON
data = {'name': 'Alice', 'scores': [88, 92, 95]}
with open('data.json', 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2)
# 读取JSON
with open('data.json', 'r', encoding='utf-8') as f:
loaded = json.load(f)
print(loaded['name'])
9.4.3 配置文件处理
import configparser
# 写入配置
config = configparser.ConfigParser()
config['DEFAULT'] = {'Server': '192.168.1.1', 'Port': '8080'}
config['USER'] = {'Name': 'Admin'}
with open('config.ini', 'w') as f:
config.write(f)
# 读取配置
config.read('config.ini')
print(config['USER']['Name']) # Admin
9.5 目录操作
9.5.1 os模块操作
import os
# 目录遍历
for root, dirs, files in os.walk('.'):
print(f"当前目录: {root}")
print(f"子目录: {dirs}")
print(f"文件: {files}")
# 创建目录
os.makedirs('data/temp', exist_ok=True)
9.5.2 pathlib模块(推荐)
from pathlib import Path
# 创建目录
data_dir = Path('data') / 'temp'
data_dir.mkdir(parents=True, exist_ok=True)
# 文件操作
file_path = data_dir / 'test.txt'
file_path.write_text('Hello Pathlib!', encoding='utf-8')
print(file_path.read_text(encoding='utf-8'))
9.6 高级IO操作
9.6.1 内存文件
from io import StringIO, BytesIO
# 文本内存文件
text_buffer = StringIO()
text_buffer.write("Hello ")
text_buffer.write("Memory IO!")
print(text_buffer.getvalue()) # Hello Memory IO!
# 二进制内存文件
bytes_buffer = BytesIO()
bytes_buffer.write(b'\x01\x02\x03')
print(bytes_buffer.getvalue()) # b'\x01\x02\x03'
9.6.2 序列化与反序列化
import pickle
# 对象序列化
data = {'name': 'Alice', 'age': 25}
with open('data.pkl', 'wb') as f:
pickle.dump(data, f)
# 反序列化
with open('data.pkl', 'rb') as f:
loaded = pickle.load(f)
print(loaded) # {'name': 'Alice', 'age': 25}
9.7 综合应用举例
案例1：日志分析系统
def analyze_log(log_file):
"""分析日志文件"""
stats = {
'total': 0,
'by_level': {},
'errors': []
}
with open(log_file, 'r', encoding='utf-8') as f:
for line in f:
stats['total'] += 1
if 'ERROR' in line:
level = 'ERROR'
stats['errors'].append(line.strip())
elif 'WARN' in line:
level = 'WARN'
else:
level = 'INFO'
stats['by_level'][level] = stats['by_level'].get(level, 0) + 1
# 生成报告
report = f"""日志分析报告:
总行数: {stats['total']}
级别分布: {stats['by_level']}
错误数量: {len(stats['errors'])}
前3条错误:
"""
for error in stats['errors'][:3]:
report += f"- {error}\n"
# 写入报告
with open('log_report.txt', 'w', encoding='utf-8') as f:
f.write(report)
return stats
# 使用示例
analyze_log('app.log')
案例2：文件加密工具
import hashlib
from pathlib import Path
def file_checksum(file_path, algorithm='sha256'):
"""计算文件校验和"""
h = hashlib.new(algorithm)
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
h.update(chunk)
return h.hexdigest()
def encrypt_file(src, dst, key):
"""简单文件加密"""
key_hash = hashlib.sha256(key.encode()).digest()
with open(src, 'rb') as s, open(dst, 'wb') as d:
for i, byte in enumerate(s.read()):
# 简单异或加密
encrypted = byte ^ key_hash[i % len(key_hash)]
d.write(bytes([encrypted]))
# 使用示例
src_file = 'secret.txt'
enc_file = 'secret.enc'
key = "mysecretpassword"
# 加密文件
encrypt_file(src_file, enc_file, key)
# 验证文件完整性
print(f"原始文件校验和: {file_checksum(src_file)}")
print(f"加密文件校验和: {file_checksum(enc_file)}")
9.8 学习路线图
9.9 学习总结
- 核心要点：
- 掌握文件打开模式的区别
- 理解上下文管理器原理
- 熟悉常见文件格式处理
- 掌握安全文件操作实践
- 实践建议：
- 始终使用with语句管理文件
- 处理文件指定明确编码
- 大文件采用分块读写
- 使用pathlib替代os.path
- 进阶方向：
- 异步文件操作(asyncio)
- 内存映射文件(mmap)
- 文件监控(watchdog)
- 压缩文件处理(zip/gzip)
- 常见陷阱：
- 忘记关闭文件导致资源泄漏
- 编码不一致导致乱码
- 文件路径跨平台问题
- 大文件一次性读取内存溢出
持续更新Python编程学习日志与技巧，敬请关注！