Python - 文本处理

(图: Python 文件I/O | 菜鸟教程)

一. 前言

文本处理, 简单而不简单.

• Built-in Functions - Python 3.14.2 documentation

open(
    file,
    mode='r',
    buffering=-1,
    encoding=None,
    errors=None,
    newline=None,
    closefd=True,
    opener=None
)

• 文件路径, 必须参数
• 打开文件方式(默认以只读的方式, 避免意外写入损毁文件), 核心参数
• 缓冲设置, 重要参数
• 文件编码格式, 一般使用utf-8, Windows上默认使用使用ansi(即系统页编码, 如时区格式为中国, 对应的则是gbk/gb2312), 核心参数
• 错误处理, 较少使用
• 换行符, 重要参数
• 控制关闭文件时是否关闭底层的「文件描述符」( 系统级的文件标识), 基本不用参数
• 自定义文件打开的底层函数, 替代系统默认的打开逻辑, 基本不用参数

一般常见的参数使用组合:

open(
    file,
    mode='r',
    encoding=None,
)

高频操作方式

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

指定缓冲参数, 需要考虑性能, 一般针对大型文件的处理

open(
    file,
    mode='r',
    encoding=None,
    newline=None
)

需要考虑换行符号, 如\n, \r\n, 系统差异

open(
    file,
    mode='r',
    encoding=None,
    errors=None
)

错误处理, 针对某些系统生成的不规范文件.

二. 参数详解

默认为只读模式, 避免文件损坏, +号表示读写皆可, 默认为文本模式, 二进制需要加b.

注意, 增加了一个x模式, 排他模式, 不允许文件已经存在. (这个参数在很多非官版的文档都未提及)

• 文本模式: 会处理字符编码( 依赖 encoding 参数, 即, 这个参数只针对文本有意义) , 换行符( 依赖 newline 参数) ;
• 二进制模式: 直接操作字节流, encoding 参数无效, 无换行符转换, 适合非文本文件.

2.1 进阶参数

2.1.1 encoding

需要稍微注意的是Windows上的历史遗留问题: 「带 BOM 的 UTF-8」和「无 BOM 的 UTF-8」有什么区别? 网页代码一般使用哪个? - 知乎

(win10之后, 内置的文本编辑器默认保存文件不再带有bom, 同时也默认保存格式为utf-8)

注意这是个坑, 假如没有指定编码格式, 默认使用Windows上的code page(和语言时区设置相关)

with open('check.txt', 'r') as f:
    print(f.encoding)
# cp936, 即gbk

为什么Windows默认使用GBK(CP936)而不是GB18030(CP54936)? - 知乎

PS D:\Code\python_workspace> chcp
Active code page: 936

powershell的默认代码页, 这个特性会导致vscode/pycharm等打印中文字符时出现乱码.

def detect_bom_encoding(file_path):
    """检测文件BOM, 返回编码( 无则返回None) """
    bom_map = {
        b'\xef\xbb\xbf': 'utf-8-sig',
        b'\xfe\xff': 'utf-16-be',
        b'\xff\xfe': 'utf-16-le',
        b'\x00\x00\xfe\xff': 'utf-32-be',
        b'\xff\xfe\x00\x00': 'utf-32-le'
    }

    with open(file_path, 'rb') as f:
        # 读取前4字节( 覆盖所有BOM长度) 
        header = f.read(4)
        for bom, encoding in bom_map.items():
            if header.startswith(bom):
                return encoding
    return None

# 测试
file_path = "test.txt"
bom_encoding = detect_bom_encoding(file_path)
if bom_encoding:
    print(f"检测到BOM, 编码为: {bom_encoding}")
else:
    print("无BOM, 需进一步推断编码")

对于文本的编码检测一般常用的库:

chardet - PyPI

charset-normalizer - PyPI

cchardet - PyPI

import cchardet

def detect_encoding_cchardet(file_path):
    """用cchardet推断编码( 高性能) """
    with open(file_path, 'rb') as f:
        raw_data = f.read(10240)
        result = cchardet.detect(raw_data)
    encoding = result['encoding']
    confidence = result['confidence']
    return encoding, confidence

# 测试
encoding, confidence = detect_encoding_cchardet(file_path)
print(f"推断编码: {encoding}, 置信度: {confidence:.2f}")

假如检测出是gbk/gb2312类似编码, 应该使用现行覆盖面更广的gb18030, 否则很大概率会出现错误.

2.1.2 buffering

buffering( 可选, 默认 -1)

为什么需要缓冲区呢?

答案很简单, 相对于代码运行的CPU/内存, 磁盘IO的速度可以说是非常非常慢的, 假如写入数据时每次更新一部分数据(如, 逐行写入)都要等待写入磁盘那么程序的速度将变得无比的慢, 所以需要一个缓冲区, 等待数据足够多了, 一次性写入磁盘, 避免反复和磁盘IO交互.

从硬件设备上也能窥其中一二, 例如sata协议(机械硬盘)一般走南桥, 而不是通过pcie直连cpu(现在高端主板大多为ssd(nvme)提供足够的pcie直连, 而不再经过南桥). 同时存储/内存和cpu整合设计也正在成为一种趋势.

import time

with open('test.txt', mode='wb', buffering=0) as f:
    f.write(b'test\n')
    time.sleep(10)
    f.write(b'end')

with open('test.txt', encoding='utf-8'm mode='w', buffering=0) as f:
    f.write('test\n')
    time.sleep(10)
    f.write('end')

简单测试, 等待写入和直接写入.

假如需要马上写盘, 可以调用flush().

buffering is an optional integer used to set the buffering policy. Pass 0 to switch buffering off (only allowed in binary mode), 1 to select line buffering (only usable when writing in text mode), and an integer > 1 to indicate the size in bytes of a fixed-size chunk buffer. Note that specifying a buffer size this way applies for binary buffered I/O, but TextIOWrapper (i.e., files opened with mode='r+') would have another buffering. To disable buffering in TextIOWrapper, consider using the write_through flag for io.TextIOWrapper.reconfigure(). When no buffering argument is given, the default buffering policy works as follows:

• Binary files are buffered in fixed-size chunks; the size of the buffer is max(min(blocksize, 8 MiB), DEFAULT_BUFFER_SIZE) when the device block size is available. On most systems, the buffer will typically be 128 kilobytes long.
• " Interactive" text files (files for which isatty() returns True) use line buffering. Other text files use the policy described above for binary files.

(base) lian@jarvis:~$ python
Python 3.12.10 | packaged by conda-forge | (main, Apr 10 2025, 22:21:13) [GCC 13.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import io
>>> io.DEFAULT_BUFFER_SIZE
8192(8K)

• performance: Update io.DEFAULT_BUFFER_SIZE to make python IO faster? - Issue #117151 - python/cpython

• TIL: Python 3.15's new buffer size

2.1.3 errors

errors( 可选, 默认 None)

• 作用: 指定文本模式下编码 / 解码错误的处理方式( 二进制模式无效) .

• 常用取值:

  取值	说明
  'strict'	默认, 遇到编码错误抛 UnicodeError;
  'ignore'	忽略错误字符( 可能丢失数据, 适合非关键文本) ;
  'replace'	用 � 替换错误字符( 保留结构, 丢失具体字符) ;
  'backslashreplace'	用转义序列( 如 \x80) 替换错误字符;

2.1.4 newline

newline( 可选, 默认 None)

• 作用: 文本模式下, 控制换行符的读取 / 写入转换规则( 二进制模式无效) .

• 取值规则:

  取值	读取行为( 将文件中的换行符转成什么)	写入行为( 将 \n 转成什么)
  None	自动识别 \n/\r\n/\r, 统一转 \n	转成系统默认换行符( Windows: \r\n, Linux: \n) ;
  ''	不转换换行符( 保留文件原始换行符)	不转换( \n 原样写入) ;
  '\n'	仅识别 \n 为换行符	\n 原样写入;

import csv

fieldnames = ['Name', 'Age']
with open('output.csv', 'w', newline='') as file:
    dict_writer = csv.DictWriter(file, fieldnames=fieldnames)

    dict_writer.writeheader()  # 写入表头
    dict_writer.writerow({'Name': 'Alice', 'Age': 25})

import csv

fieldnames = ['Name', 'Age']
with open('output1.csv', 'w', newline='\r') as file:
    dict_writer = csv.DictWriter(file, fieldnames=fieldnames)

    dict_writer.writeheader()  # 写入表头
    dict_writer.writerow({'Name': 'Alice', 'Age': 25})

不同符号带来的差异.

2.1.5 closefd

closefd( 可选, 默认 True)

• 作用: 控制关闭文件时是否关闭底层的「文件描述符」( 系统级的文件标识) .
• 取值规则:
  • True( 默认) : 关闭文件对象时, 同时关闭底层文件描述符;
  • False: 仅关闭文件对象, 不关闭文件描述符( 仅当 file 参数是文件描述符( 整数) 时有效) ;
• 注意: 若 file 是文件路径( 99% 的场景) , 设置 closefd=False 会直接报错; 仅底层文件操作( 如 os.open() 返回的文件描述符) 才会用到.

2.1.6 opener

opener( 可选, 默认 None)

• 作用: 自定义文件打开的底层函数, 替代系统默认的打开逻辑.

• 用法: opener 是一个函数, 接收两个参数( file 路径, flags 打开模式标志位) , 返回文件描述符( 整数) .

• 使用场景: 自定义权限校验, 加密文件打开, 访问远程文件等小众场景.

• 示例: 自定义文件打开权限( Linux/Mac 有效)

  import os
  # 自定义opener: 打开文件时设置权限为0o666( 所有用户可读可写) 
  def my_opener(path, flags):
      return os.open(path, flags, 0o666)
  
  with open('test.txt', 'w', opener=my_opener) as f:
      f.write('自定义权限打开')
  

三. 文件对象

一般而言, 对于操作文件, 建议在with语句之下使用, 避免文件打开后没有关闭的问题.

with open('test.txt', 'w') as f:
    f.write('test')

3.1 读取

3.2 写入

writelines()需要稍微注意, 不要被单词带偏, 这个函数只是方便直接写入序列内容, 而不需要转为字符串, 并不是写入后自动换行.

    def writelines(self, lines):
        """
        Write lines to the file from an interable.

        NOTE: writelines() does NOT add line separators.
        """
        self._checkClosed()
        for line in lines:
            self.write(line)

(源码)

3.3 指针

需要注意的是a+模式打开的文件, 需要写入和读取内容.

with open('test.txt', mode='a+', encoding='utf-8') as f:
    # f.seek(0) # 将指针移动到文件头
    for l in f:
        print(l)

由于指针指向文件末尾, 所以打印不出任何内容, 需要手动将指针指向文件头.

四. 大文件处理

对于巨大的文件, 不可能都一次性装进内存.

with open('test.txt', mode='r', encoding='utf-8') as f:
    for l in f:
        yield l

with open(r"test.txt", mode='rb') as f:
    while True:
        chunk = f.read(8192)
        if not chunk:
            break

常见的逐行/逐块按需读取.

4.1 内存映射

Memory Mapping:

内存映射( Memory Mapping) 是操作系统提供的底层机制, 通过 mmap 系统调用将文件的部分 / 全部内容直接映射到进程的虚拟内存空间. 操作这块内存就等同于操作文件本身, 无需显式调用 read()/write(), 由操作系统负责内存与磁盘的同步( 按需加载, 脏页回写) .

对文本处理而言, mmap 最大的价值是突破传统文件读写的性能瓶颈, 尤其适合 GB 级超大文本文件的处理.

Memory mapping is a technique that maps a file or device into a process's virtual address space, allowing the file's content to be accessed as if it were part of the program's memory. This approach eliminates the need for traditional file I/O operations, significantly improving efficiency and simplifying data access.

一文搞懂内存映射(Memory Map)原理 - 知乎

深入理解内存映射: 加速文件访问的神奇原理 - 知乎

找个文本文件测试一下, 号称网络最长文: 宇宙巨校闪级生_百度百科

import time

t = time.time()
s = 'ERROR'
with open(r"D:\Downloads\1951\宇宙巨校闪级生.txt", mode='r', encoding='gb18030') as f:
    for line in f:
        if s in line:
            print(line)

print(time.time() - t)

# 0.6s左右

import mmap
import time

def optimized_mmap_search(file_path, keyword, encoding='gb18030'):
    keyword_bytes = keyword.encode(encoding)  # 这是二进制的匹配, 先将目标关键词编码处理
    keyword_len = len(keyword_bytes)
    line_c = b'\n'  # 换行符, 用于行数据的判断
    s = time.time()

    with open(file_path, 'rb') as f:
        # fileno() 方法返回一个整型的文件描述符(file descriptor FD 整型), 可用于底层操作系统的 I/O 操作. 
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            pos = 0
            # mm.size(), mm.size(), Return the length of the file, which can be larger than the size of the memory-mapped area.
            file_size = mm.size()
            while pos < file_size:
                # mmap.find做二进制子串匹配( C级实现, 比正则快) 
                # find(关键词, 开始位置, 结束位置)
                pos = mm.find(keyword_bytes, pos)
                # 假如没有匹配到内容,遍历整个文件
                if pos == -1:
                    break
                # 查找到关键词
                # 下一步判断出关键所在的一整行
                # 定位行首( 找前一个换行符) 
                line_start = mm.rfind(line_c, 0, pos) + 1  # 假如数据为第一行, 没有换行符, 所以是 -1 + 1 = 0
                # 定位行尾( 找后一个换行符) , pos + keyword_len, 防止关键词词出现在最末端
                line_end = mm.find(line_c, pos + keyword_len)
                # 假如查找到最后
                if line_end == -1:
                    line_end = file_size
                # 解码整行
                line_bytes = mm[line_start:line_end]
                line = line_bytes.decode(encoding, errors='ignore')
                print(line)
                # 跳过当前行, 避免重复匹配
                pos = line_end + 1

    print(f"优化后mmap耗时: {time.time() - s:.2f}秒")

optimized_mmap_search(r"D:\Downloads\1951\宇宙巨校闪级生.txt", 'ERROR', encoding='gb18030')
# 0.07s

在大小约100M, 采用gb18030编码的文本, 判断文本是否包含某个关键词, 内存映射的速度差不多是逐行读取速度的十倍, 理论上, 更大的文件, 例如GB或更大, 这种差异会更明显.

但需要注意, 这种方式的实现代码更为繁琐, 同时也需要考虑处理较小文件时, 冷启动所消耗的时间.

one more thing, 假如需要将匹配到的数据, 更改/删除, 然后保存.

五. 拓展

为了应对当下越来越巨型化的数据处理, 为更快, 更好的读取/存储, 各种格式文件应运而生: pandas 各种存储格式速度对比[CSV, hdf5,SQL, pickle, feather, parquet] - 知乎

列存 vs 行存核心差异

• 列存格式( Parquet/ORC/Feather/HDF5) : 适合「读多写少, 列维度查询」( 如统计某几列的均值 / 求和) , 压缩率高, 大数据分析场景首选;
• 行存格式( Avro/JSONL/CSV) : 适合「写多读少, 整行操作」( 如流式数据写入, 单条数据更新) , 灵活性高, 实时场景首选.