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 - 迭代(推荐内存高效)

考前记忆

面试重点

  1. with 语句的优势

    • 自动关闭文件
    • 异常安全
    • 避免资源泄漏
  2. read() vs readlines() vs 迭代

    • read() 一次性读取,内存占用大
    • readlines() 返回列表,占用内存
    • 迭代方式最节省内存(推荐)
  3. json.dump() vs json.dumps()

    • dump() 写入文件
    • dumps() 转为字符串
  4. ensure_ascii=False 的重要性

    • 允许非 ASCII 字符(如中文)正常显示
    • 不设置会转义为 \uXXXX
  5. 文件指针操作

    • 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 语句操作文件?

  1. 自动关闭:代码块结束后自动调用 close()
  2. 异常安全:即使发生异常也会关闭文件
  3. 代码简洁:无需手动 try-finally
  4. 避免资源泄漏:确保文件描述符被释放

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
注释 不支持 支持

参考资料