在处理大型文本文件时，我们经常需要将文件内容按固定大小的块或行数进行分组，以便于后续的数据分析或并行处理。例如，可能需要将日志文件中的每三行作为一个逻辑单元进行处理。Python提供了简洁而强大的方式来实现这一需求。
核心分组逻辑
实现文件行内容分组的核心在于利用Python的列表切片（slicing）功能，结合一个以指定步长迭代的循环。首先，我们需要将文件的所有行读取到一个列表中，然后通过步进的方式从这个列表中提取子列表。
假设我们有一个名为 data.txt 的文本文件，其内容如下：
aDB8786793440 bDB8978963432 cDB9898908345 dDB8908908454 eDB9083459089 fDB9082390843 gDB9083490345
我们的目标是将这些行按三行一组进行分组，最终得到一个包含多个子列表的列表，例如： [["a", "b", "c"], ["d", "e", "f"], ["g"]]。
以下是实现这一功能的Python代码：
立即学习“Python免费学习笔记（深入）”；
def group_lines_from_file(file_path: str, group_size: int = 3) -> list[list[str]]:
"""
从文本文件中读取所有行，并按指定大小进行分组。
Args:
file_path (str): 文本文件的路径。
group_size (int): 每组包含的行数。默认为3。
Returns:
list[list[str]]: 包含分组行的列表。每个子列表代表一个组。
"""
groups = []
try:
with open(file_path, 'r', encoding='utf-8') as f:
# 读取所有行，并去除每行末尾的换行符
# 使用strip()确保数据干净，避免换行符影响后续处理
lines = [line.strip() for line in f.readlines()]
# 遍历lines列表，每次跳过group_size个元素
for i in range(0, len(lines), group_size):
# 使用切片操作提取当前组的行
# 切片会自动处理列表末尾不足group_size的情况
groups.append(lines[i : i + group_size])
except FileNotFoundError:
print(f"错误：文件 '{file_path}' 未找到。")
except Exception as e:
print(f"处理文件时发生错误：{e}")
return groups
# 示例用法
file_name = 'data.txt' # 确保此文件存在于脚本同目录下或提供完整路径
# 创建一个示例文件
with open(file_name, 'w', encoding='utf-8') as f:
f.write("aDB8786793440\n")
f.write("bDB8978963432\n")
f.write("cDB9898908345\n")
f.write("dDB8908908454\n")
f.write("eDB9083459089\n")
f.write("fDB9082390843\n")
f.write("gDB9083490345\n")
grouped_data = group_lines_from_file(file_name, group_size=3)
print(grouped_data)
# 预期输出:
# [['aDB8786793440', 'bDB8978963432', 'cDB9898908345'],
# ['dDB8908908454', 'eDB9083459089', 'fDB9082390843'],
# ['gDB9083490345']]代码解析
-
文件读取:
- with open(file_path, 'r', encoding='utf-8') as f:：以只读模式打开指定文件。with语句确保文件在使用完毕后会被正确关闭，即使发生错误。encoding='utf-8'是处理文本文件的良好实践。
- lines = [line.strip() for line in f.readlines()]：f.readlines()会读取文件的所有行并返回一个字符串列表，每个字符串包含末尾的换行符（\n）。line.strip()用于移除这些换行符以及其他空白字符，确保每行内容的纯净性。
-
分组逻辑:
- for i in range(0, len(lines), group_size):：这是实现分组的关键。
- range(start, stop, step) 函数生成一个序列。
- start (0) 表示从列表的第一个元素开始。
- stop (len(lines)) 表示迭代到列表的末尾。
- step (group_size) 表示每次迭代跳过的元素数量。例如，如果 group_size 是 3，i 将依次取 0, 3, 6, ...。
- groups.append(lines[i : i + group_size])：在每次循环中，使用列表切片 lines[i : i + group_size] 提取当前组的行。
- Python的列表切片非常灵活，即使 i + group_size 超出了列表的实际长度，切片操作也会自动截取到列表的末尾，因此无需额外处理最后不足 group_size 行的情况。
- for i in range(0, len(lines), group_size):：这是实现分组的关键。
注意事项与扩展
文件路径: 确保 file_path 是正确的文件路径。如果文件不在脚本的同一目录下，需要提供绝对路径或相对路径。
-
大文件处理: 上述方法一次性将所有行加载到内存中 (f.readlines())。对于非常大的文件（例如，几GB甚至更大），这可能会导致内存不足。在这种情况下，更高效的方法是使用生成器（generator）逐行或逐块读取并处理，而不是一次性加载所有内容。例如：
def group_lines_lazy(file_path: str, group_size: int = 3): buffer = [] with open(file_path, 'r', encoding='utf-8') as f: for line in f: buffer.append(line.strip()) if len(buffer) == group_size: yield buffer buffer = [] if buffer: # 处理剩余的行 yield buffer # for group in group_lines_lazy('data.txt', 3): # print(group) 错误处理: 在 group_lines_from_file 函数中加入了 try-except 块，用于捕获 FileNotFoundError 和其他潜在的 Exception，增强了代码的健壮性。
通用性: group_size 参数使得这个函数可以灵活地按任意行数进行分组，而不仅仅是三行。
总结
通过结合Python的文件I/O操作、列表推导式（用于清理数据）以及 range 函数配合列表切片，我们可以非常高效且优雅地实现文本文件行内容的按指定数量分组。这种方法不仅代码简洁，而且具有良好的可读性和可扩展性，能够满足大多数数据处理场景的需求。对于处理超大文件，应考虑使用生成器模式以优化内存使用。
