Day34 - 文件读写与JSON处理
详细讲解
1. 文件打开与关闭
1.1 基本语法
# 传统方式 - 需要手动关闭
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 块结束后自动关闭
1.2 文件打开模式
| 模式 | 说明 | 文件指针位置 | 如果文件存在 | 如果文件不存在 |
|---|---|---|---|---|
'r' |
读取(默认) | 开头 | 读取 | 报错 |
'w' |
写入 | 开头 | 覆盖 | 创建 |
'x' |
创建并写入 | 开头 | 报错 | 创建 |
'a' |
追加 | 末尾 | 追加 | 创建 |
'r+' |
读取+写入 | 开头 | 读取 | 报错 |
'w+' |
读取+写入 | 开头 | 覆盖 | 创建 |
'a+' |
读取+追加 | 末尾 | 追加 | 创建 |
1.3 编码问题
# 推荐使用 UTF-8 编码
with open('file.txt', 'r', encoding='utf-8') as f:
content = f.read()
# 处理不同编码
with open('file.txt', 'r', encoding='gbk') as f:
content = f.read()
# 遇到编码错误时的处理
with open('file.txt', 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
with open('file.txt', 'r', encoding='utf-8', errors='replace') as f:
content = f.read() # 替换为 ?
2. 文件读取操作
2.1 读取全部内容
# 读取整个文件
with open('example.txt', 'r', encoding='utf-8') as f:
content = f.read() # 返回字符串
# 读取前 N 个字符
with open('example.txt', 'r', encoding='utf-8') as f:
content = f.read(100) # 读取前100个字符
2.2 按行读取
# 读取单行(包含换行符)
with open('example.txt', 'r', encoding='utf-8') as f:
line1 = f.readline()
line2 = f.readline()
# 读取所有行(返回列表)
with open('example.txt', 'r', encoding='utf-8') as f:
lines = f.readlines()
# 直接迭代文件对象(推荐,内存高效)
with open('example.txt', 'r', encoding='utf-8') as f:
for line in f:
print(line.strip()) # strip() 去除换行符
2.3 逐字符读取
with open('example.txt', 'r', encoding='utf-8') as f:
while True:
char = f.read(1) # 读取单个字符
if not char:
break
print(char, end='')
3. 文件写入操作
3.1 写入字符串
# 写入单行(不自动添加换行符)
with open('output.txt', 'w', encoding='utf-8') as f:
f.write('第一行')
f.write('第二行') # 不会换行
# 写入多行
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('第一行', file=f)
print('第二行', file=f)
3.2 追加模式
# 在文件末尾追加
with open('log.txt', 'a', encoding='utf-8') as f:
f.write('新的日志条目\n')
3.3 二进制文件
# 写入二进制数据
with open('data.bin', 'wb') as f:
f.write(b'\x00\x01\x02\x03')
# 读取二进制数据
with open('data.bin', 'rb') as f:
data = f.read()
print(data) # b'\x00\x01\x02\x03'
4. 文件指针操作
with open('example.txt', 'r', encoding='utf-8') as f:
# 获取文件指针位置
print(f.tell()) # 0
# 读取一些内容
content = f.read(10)
print(f.tell()) # 10
# 移动文件指针
f.seek(0) # 移到开头
print(f.tell()) # 0
f.seek(5) # 移到第5个字节
print(f.tell()) # 5
# 从末尾往前移动(需要使用二进制模式)
# f.seek(-10, 2) # 从文件末尾往前10字节
5. JSON 数据处理
5.1 JSON 模块概述
JSON(JavaScript Object Notation)是一种轻量级数据交换格式。
import json
# Python 对象与 JSON 的对应关系:
# dict ↔ object
# list/tuple ↔ array
# str ↔ string
# int/float ↔ number
# True/None ↔ true/null
# False ↔ false
5.2 序列化(Python → JSON)
# 字典转 JSON 字符串
data = {'name': '张三', 'age': 25, 'city': '北京'}
# 转为字符串
json_str = json.dumps(data)
print(json_str)
# 输出: {"name": "\u5f20\u4e09", "age": 25, "city": "\u5317\u4eac"}
# 美化输出
json_str_pretty = json.dumps(data, indent=4, ensure_ascii=False)
print(json_str_pretty)
# 按键排序
json_str_sorted = json.dumps(data, sort_keys=True, ensure_ascii=False)
print(json_str_sorted)
# 处理特殊类型
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
# 自定义编码器
def person_encoder(obj):
if isinstance(obj, Person):
return {'name': obj.name, 'age': obj.age, 'type': 'Person'}
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")
person = Person('李四', 30)
json_str = json.dumps(person, default=person_encoder, ensure_ascii=False)
print(json_str)
5.3 反序列化(JSON → Python)
# JSON 字符串转字典
json_str = '{"name": "张三", "age": 25}'
data = json.loads(json_str)
print(data) # {'name': '张三', 'age': 25}
# 从文件读取 JSON
with open('data.json', 'r', encoding='utf-8') as f:
data = json.load(f)
# 处理特殊解码
def person_decoder(dct):
if 'type' in dct and dct['type'] == 'Person':
return Person(dct['name'], dct['age'])
return dct
json_str = '{"name": "王五", "age": 28, "type": "Person"}'
person = json.loads(json_str, object_hook=person_decoder)
5.4 JSON 文件操作
# 写入 JSON 文件
data = {
'users': [
{'name': '张三', 'age': 25},
{'name': '李四', 'age': 30}
],
'count': 2
}
with open('users.json', 'w', encoding='utf-8') as f:
json.dump(data, f, indent=4, ensure_ascii=False)
# 读取 JSON 文件
with open('users.json', 'r', encoding='utf-8') as f:
data = json.load(f)
print(data)
# 追加写入 JSON Lines 格式
with open('logs.jsonl', 'a', encoding='utf-8') as f:
log_entry = {'timestamp': '2024-01-01', 'level': 'INFO', 'message': '系统启动'}
f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
6. 文件操作上下文管理器
# 基本上下文管理器
with open('file.txt', 'r') as f:
content = f.read()
# 多个文件
with open('input.txt', 'r') as infile, open('output.txt', 'w') as outfile:
content = infile.read()
outfile.write(content.upper())
# 自定义上下文管理器
class FileManager:
def __init__(self, filename, mode):
self.filename = filename
self.mode = mode
self.file = None
def __enter__(self):
self.file = open(self.filename, self.mode)
return self.file
def __exit__(self, exc_type, exc_val, exc_tb):
if self.file:
self.file.close()
return False # 不抑制异常
# 使用
with FileManager('test.txt', 'w') as f:
f.write('Hello, World!')
7. 实战案例:配置文件读写
# config.json
# {
# "database": {
# "host": "localhost",
# "port": 3306,
# "name": "myapp"
# },
# "app": {
# "debug": true,
# "secret_key": "abc123"
# }
# }
import json
from pathlib import Path
class Config:
def __init__(self, config_path='config.json'):
self.config_path = Path(config_path)
self.data = {}
def load(self):
"""加载配置文件"""
if self.config_path.exists():
with open(self.config_path, 'r', encoding='utf-8') as f:
self.data = json.load(f)
else:
print(f"配置文件 {self.config_path} 不存在,使用默认配置")
self.data = self.get_default_config()
return self
def save(self):
"""保存配置到文件"""
with open(self.config_path, 'w', encoding='utf-8') as f:
json.dump(self.data, f, indent=4, ensure_ascii=False)
@staticmethod
def get_default_config():
"""获取默认配置"""
return {
'database': {
'host': 'localhost',
'port': 5432,
'name': 'default_db',
'user': 'admin',
'password': 'admin'
},
'app': {
'debug': False,
'secret_key': 'change-me-in-production',
'log_level': 'INFO'
}
}
def get(self, key, default=None):
"""获取配置项(支持点号分隔的路径)"""
keys = key.split('.')
value = self.data
for k in keys:
if isinstance(value, dict):
value = value.get(k)
if value is None:
return default
else:
return default
return value
def set(self, key, value):
"""设置配置项"""
keys = key.split('.')
target = self.data
for k in keys[:-1]:
if k not in target:
target[k] = {}
target = target[k]
target[keys[-1]] = value
# 使用示例
if __name__ == '__main__':
config = Config('app_config.json')
config.load()
# 获取配置
db_host = config.get('database.host')
print(f"数据库主机: {db_host}")
# 修改配置
config.set('app.debug', True)
config.save()
背诵版
核心速查表
┌─────────────────────────────────────────────────────────────┐ │ 文件操作模式 │ ├─────────────────────────────────────────────────────────────┤ │ r │ 读取 │ 开头 │ 文件不存在报错 │ │ w │ 写入 │ 开头 │ 文件不存在创建,存在的覆盖 │ │ a │ 追加 │ 末尾 │ 文件不存在创建 │ │ x │ 新建 │ 开头 │ 文件存在报错 │ │ + │ 读写 │ │ 组合使用如 r+, w+, a+ │ ├─────────────────────────────────────────────────────────────┤ │ b │ 二进制 │ 组合使用如 rb, wb, ab │ └─────────────────────────────────────────────────────────────┘JSON 操作速查
| 操作 | 函数 | 说明 |
|---|---|---|
| Python → JSON 字符串 | json.dumps(obj) |
序列化 |
| JSON 字符串 → Python | json.loads(str) |
反序列化 |
| Python → JSON 文件 | json.dump(obj, f) |
写入文件 |
| JSON 文件 → Python | json.load(f) |
读取文件 |
读取方法对比
| 方法 | 返回值 | 特点 |
|---|---|---|
read() |
str | 读取全部 |
readline() |
str | 读取一行 |
readlines() |
list | 读取所有行到列表 |
for line in f |
- | 迭代(推荐内存高效) |
考前记忆
面试重点
-
with语句的优势- 自动关闭文件
- 异常安全
- 避免资源泄漏
-
read()vsreadlines()vs 迭代read()一次性读取,内存占用大readlines()返回列表,占用内存- 迭代方式最节省内存(推荐)
-
json.dump()vsjson.dumps()dump()写入文件dumps()转为字符串
-
ensure_ascii=False的重要性- 允许非 ASCII 字符(如中文)正常显示
- 不设置会转义为
\uXXXX
-
文件指针操作
tell()获取位置seek(offset, whence)移动位置
记忆口诀
文件操作用 with, 自动关闭不失手。 json dumps 序列化, loads 反序列化。 read readline readlines, 迭代读取最推荐。测试题
选择题
1. 下面哪个选项可以正确读取文件所有行?
# A.
with open('file.txt', 'r') as f:
content = f.read()
# B.
with open('file.txt', 'r') as f:
lines = f.readlines()
# C.
with open('file.txt', 'r') as f:
for line in f:
print(line)
# D.
以上全部正确
答案:D
2. 如果要追加内容到文件,应该使用哪个模式?
# A. 'r+'
# B. 'w'
# C. 'a'
# D. 'x'
答案:C
3. json.dumps(data, ensure_ascii=False) 的作用是?
# A. 忽略所有 ASCII 字符
# B. 保留非 ASCII 字符(如中文)
# C. 将所有字符转为 ASCII
# D. 报错
答案:B
4. 读取大文件时,推荐使用哪种方式?
# A. read() 一次性读取
# B. readlines() 读取到列表
# C. for line in file 迭代
# D. readline() 循环
答案:C
5. seek(0) 的作用是?
# A. 移到文件开头
# B. 移到文件末尾
# C. 当前位置
# D. 下一行
答案:A
编程题
1. 文件复制程序:
def copy_file(src, dst):
"""复制文件"""
with open(src, 'rb') as src_file:
with open(dst, 'wb') as dst_file:
# 分块复制(适合大文件)
while chunk := src_file.read(8192):
dst_file.write(chunk)
print(f"已复制: {src} -> {dst}")
# 使用
copy_file('source.txt', 'destination.txt')
2. 日志文件处理:
import json
from datetime import datetime
def append_log(log_file, level, message):
"""追加日志到 JSON Lines 文件"""
log_entry = {
'timestamp': datetime.now().isoformat(),
'level': level,
'message': message
}
with open(log_file, 'a', encoding='utf-8') as f:
f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
def read_logs(log_file):
"""读取所有日志"""
logs = []
with open(log_file, 'r', encoding='utf-8') as f:
for line in f:
if line.strip():
logs.append(json.loads(line))
return logs
def filter_logs(log_file, level):
"""按级别过滤日志"""
with open(log_file, 'r', encoding='utf-8') as f:
for line in f:
if line.strip():
entry = json.loads(line)
if entry['level'] == level:
yield entry
# 使用
append_log('app.log', 'INFO', '应用启动')
append_log('app.log', 'ERROR', '发生错误')
append_log('app.log', 'INFO', '处理完成')
# 读取所有日志
all_logs = read_logs('app.log')
print(f"共 {len(all_logs)} 条日志")
# 过滤错误日志
for log in filter_logs('app.log', 'ERROR'):
print(f"[{log['timestamp']}] {log['message']}")
3. 配置文件管理器:
import json
class ConfigManager:
"""配置文件管理器"""
def __init__(self, config_file='config.json'):
self.config_file = config_file
self.config = {}
self.load()
def load(self):
"""从文件加载配置"""
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
self.config = json.load(f)
except FileNotFoundError:
print(f"配置文件 {self.config_file} 不存在")
self.config = self._default_config()
def save(self):
"""保存配置到文件"""
with open(self.config_file, 'w', encoding='utf-8') as f:
json.dump(self.config, f, indent=4, ensure_ascii=False)
@staticmethod
def _default_config():
"""默认配置"""
return {
'version': '1.0.0',
'database': {
'host': 'localhost',
'port': 3306,
'name': 'testdb',
'charset': 'utf8mb4'
},
'server': {
'host': '0.0.0.0',
'port': 8080,
'debug': True
},
'logging': {
'level': 'INFO',
'file': 'app.log'
}
}
def get(self, key_path):
"""获取配置项(支持路径如 'database.host')"""
keys = key_path.split('.')
value = self.config
for key in keys:
if isinstance(value, dict):
value = value.get(key)
else:
return None
return value
def set(self, key_path, value):
"""设置配置项"""
keys = key_path.split('.')
target = self.config
for key in keys[:-1]:
target = target.setdefault(key, {})
target[keys[-1]] = value
# 使用示例
if __name__ == '__main__':
config = ConfigManager('app_config.json')
# 读取配置
print(f"数据库主机: {config.get('database.host')}")
print(f"服务器端口: {config.get('server.port')}")
# 修改配置
config.set('server.debug', False)
config.set('database.host', '192.168.1.100')
# 保存
config.save()
print("配置已保存")
问答题
Q1: 为什么推荐使用 with 语句操作文件?
- 自动关闭:代码块结束后自动调用
close() - 异常安全:即使发生异常也会关闭文件
- 代码简洁:无需手动 try-finally
- 避免资源泄漏:确保文件描述符被释放
Q2: 如何处理不同编码的文件?
# 常见编码处理
with open('file.txt', 'r', encoding='utf-8') as f: # UTF-8
pass
with open('file.txt', 'r', encoding='gbk') as f: # GBK
pass
# 错误处理
with open('file.txt', 'r', encoding='utf-8', errors='ignore') as f: # 忽略错误
pass
with open('file.txt', 'r', encoding='utf-8', errors='replace') as f: # 替换为 ?
pass
Q3: JSON 和 Python 字典的区别是什么?
| 方面 | JSON | Python dict |
|---|---|---|
| 类型 | 字符串格式 | Python 对象 |
| 键 | 必须是字符串 | 可哈希对象 |
| 值 | 有限类型 | 任意类型 |
| 布尔值 | true/false | True/False |
| 空值 | null | None |
| 注释 | 不支持 | 支持 |