5.1 文件读取与解析

数据处理的起点往往是文件操作。实际工作中，90%的报错来自三个问题：编码错误、路径混淆和内存溢出。本节针对日常高频场景给出可直接复用的代码模板。

5.1.1 文本文件读取

最基础的日志或配置文件读取，推荐始终使用 with 语句并显式指定编码：

# 正确做法：指定编码，自动关闭句柄
def read_text_safe(filepath):
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            return f.read()
    except UnicodeDecodeError:
        # 常见坑：Windows生成的文件可能是gbk编码
        with open(filepath, 'r', encoding='gbk', errors='ignore') as f:
            return f.read()
    except FileNotFoundError:
        print(f"文件不存在: {filepath}")
        return None

# 逐行读取（节省内存）
def process_line_by_line(filepath, callback):
    with open(filepath, 'r', encoding='utf-8') as f:
        for line_num, line in enumerate(f, 1):
            if line.strip():  # 跳过空行
                callback(line.strip(), line_num)

注意：Python 3.11+ 推荐使用 encoding='utf-8' 作为默认，但处理国内 Windows 环境生成的文件时，建议先尝试 UTF-8，失败再回退到 GBK。

5.1.2 CSV 数据处理

对于结构化数据，Pandas 是事实标准，但需注意类型推断和缺失值：

import pandas as pd

def load_csv_robust(filepath):
    """带异常处理的 CSV 读取"""
    try:
        # engine='python' 对不规则分隔符更宽容
        df = pd.read_csv(
            filepath, 
            encoding='utf-8',
            dtype={'user_id': str},  # 防止 ID 被识别为数字（前导零丢失）
            na_values=['NA', 'NULL', ''],  # 统一定义缺失值
            engine='python'
        )
        print(f"成功加载 {len(df)} 行，{len(df.columns)} 列")
        return df
    except pd.errors.EmptyDataError:
        print("警告：空文件")
        return pd.DataFrame()
    except Exception as e:
        print(f"解析错误: {e}")
        raise

# 处理大 CSV（分块读取）
def process_large_csv(filepath, chunksize=10000):
    chunk_iter = pd.read_csv(filepath, chunksize=chunksize)
    for chunk in chunk_iter:
        # 在这里处理每个 chunk，例如数据清洗
        processed = chunk.dropna(subset=['关键字段'])
        yield processed

5.1.3 JSON 解析

API 返回或配置文件常用 JSON，重点关注嵌套结构提取：

import json
from pathlib import Path

def parse_json_file(filepath):
    path = Path(filepath)
    if not path.exists():
        return {}
    
    try:
        with open(path, 'r', encoding='utf-8') as f:
            data = json.load(f)
            
        # 实际场景：提取嵌套数据（如 API 返回的 data 字段）
        if isinstance(data, dict) and 'data' in data:
            return data['data']
        return data
        
    except json.JSONDecodeError as e:
        print(f"JSON 格式错误: {e}")
        # 常见修复：去除 BOM 头或非法字符后重试
        content = path.read_bytes().decode('utf-8-sig')
        return json.loads(content)

# 处理 JSON Lines 格式（每行一个 JSON 对象，常见于日志）
def read_jsonl(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        for line in f:
            if line.strip():
                yield json.loads(line)

5.1.4 大文件与内存管理

当文件超过内存容量（如 10GB 日志），使用生成器和分块：

# 方法 1：逐块读取二进制文件
def read_in_chunks(filepath, chunk_size=8192):
    with open(filepath, 'rb') as f:
        while chunk := f.read(chunk_size):
            yield chunk

# 方法 2：使用内存映射（适合随机访问大文件）
import mmap

def search_in_large_file(filepath, keyword):
    with open(filepath, 'r+b') as f:
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            if mm.find(keyword.encode()) != -1:
                return True
    return False

5.1.5 路径与编码的实用建议

1. 路径处理：停止使用字符串拼接，改用 pathlib：

   from pathlib import Path
   data_dir = Path("data") / "raw" / "2024"
   file_path = data_dir / "report.csv"
   

1. 编码检测：遇到未知编码时，使用 chardet 库自动检测：

   import chardet
   raw = open('unknown.txt', 'rb').read()
   result = chardet.detect(raw)
   print(result['encoding'])  # 输出如 'GB2312'
   

1. 生产环境 checklist：

• 始终验证文件存在性（path.exists()）
• 限制读取大小（防止 DoS 攻击或误操作读取超大文件）
• 敏感数据读取后及时删除变量（del data 并调用 gc.collect()）

小结：文件读取看似简单，但健壮的生产代码必须处理编码异常、文件不存在、内存限制三种情况。建议将上述函数封装为项目内的 file_utils.py，统一处理所有 IO 操作。