1. 问题背景与分析
在日常的数据处理任务中，我们经常需要从大量具有相似结构但内容各异的文本文件中提取特定信息。本教程将以一个具体场景为例：从一个包含多层子目录的系统中，查找所有 .txt 文件。每个 .txt 文件都包含两个逻辑部分，每个部分由固定数量的行组成，并以特定标识符（例如 >）分隔。我们的目标是从每个部分的特定行中，提取“下载速度”和“上传速度”的数据，并根据其数值进行条件判断和格式化输出。
示例文件结构如下，其中每个逻辑部分（由 > 引导）固定包含 8 行：
> this is first output and some another contents these are some test lines to fill the file Testing download speed Download: 0.00 Mbit/s Testing upload speed Upload: 0.00 Mbit/s > this is second output but other texts go here too these are some test lines to fill the file Testing download speed Download: 1200.58 Mbit/s Testing upload speed Upload: 857.25 Kbit/s
我们需要对每个文件的每个部分的“Download:”和“Upload:”行进行解析，提取速度值和单位，并根据预设的条件（例如速度为零、小于 600 Mbit/s 等）输出不同的信息。
2. 解决方案概述
为了高效且健壮地解决此类问题，我们采用以下策略：
- 递归文件查找： 使用 pathlib 库递归地查找所有目标文件。
- 文件内容分块： 由于文件结构一致，每个逻辑部分行数固定，我们可以将文件内容按固定大小分块，将每个逻辑部分视为一个独立的数据块。
- 数据提取与转换： 编写辅助函数，从指定行中解析出速度值（浮点数）和单位（字符串）。
- 结果格式化： 编写辅助函数，根据提取的速度值和单位，生成符合要求的输出字符串。
- 模块化设计： 将不同的功能封装在独立的函数中，提高代码的可读性和可维护性。
3. 具体实现步骤
3.1 定义常量与文件查找
首先，我们需要定义文件中每个逻辑部分的行数以及文件包含的逻辑部分数量。然后，使用 pathlib 库进行递归文件查找。
立即学习“Python免费学习笔记（深入）”；
import sys
from pathlib import Path
# 定义常量：每个逻辑部分的行数，以及文件包含的逻辑部分数量
LINES_PER_PART = 8
PARTS_PER_FILE = 2 # 根据示例文件，每个文件有两部分
def main():
# 递归查找当前目录及其子目录下所有 .txt 文件
result = list(Path(".").rglob("*.txt"))
for filename in result:
with open(filename, 'r') as file:
# 读取文件所有行
lines = file.readlines()
# ... 后续处理 ...3.2 文件内容分块函数
为了将文件的所有行按固定大小（LINES_PER_PART）分割成多个逻辑部分，我们定义一个 chunks 函数。
def chunks(arr, chunk_size):
"""
将列表 arr 分割成大小为 chunk_size 的块。
"""
result = []
for i in range(0, len(arr), chunk_size):
result.append(arr[i:i+chunk_size])
return result在 main 函数中调用此函数：
# ... (在 main 函数内部)
lines = file.readlines()
parts = chunks(lines, LINES_PER_PART) # 将文件内容分块
# ...3.3 速度信息解析函数
parse_speed_info 函数负责从包含速度信息的字符串中提取数值和单位。它假定速度值是字符串的第二个单词，单位是第三个单词。
def parse_speed_info(string):
"""
从速度信息字符串中解析出速度值（浮点数）和单位。
示例输入: "Download: 1200.58 Mbit/s\n"
示例输出: (1200.58, "Mbit/s")
"""
# 忽略 "Download:" 或 "Upload:" 部分，从第二个单词开始解析
speed_info_list = string.split()[1::]
# 返回速度值（转换为浮点数）和单位
return (
float(speed_info_list[0]),
speed_info_list[1].strip() # 移除单位末尾的换行符
)3.4 速度信息格式化函数
stringify_speed_info 函数根据解析出的速度值和单位，生成符合特定条件（如零、小于 600）的报告字符串。
def stringify_speed_info(speed, unit):
"""
根据速度值和单位，生成格式化的输出字符串。
"""
if speed == 0:
return "zero"
elif unit == "Mbit/s" and speed < 600.0: # 仅对 Mbit/s 单位进行小于600的判断
return f"less than 600 {unit}"
else:
return f"{speed} {unit}"注意： 原始问题中对“小于 600”的判断仅针对 Mbit/s 单位，此处已在代码中体现。
3.5 遍历分块并输出结果
在 main 函数中，遍历 parts 列表，对每个逻辑部分提取并处理下载和上传速度信息。根据示例文件结构，下载速度信息在倒数第 3 行 (part[-3])，上传速度信息在最后一行 (part[-1])。
# ... (在 main 函数内部)
parts = chunks(lines, LINES_PER_PART)
for i, part in enumerate(parts, 1): # i 从 1 开始计数，表示第几部分
# 下载速度信息在当前部分的倒数第三行
download_info = parse_speed_info(part[-3])
# 上传速度信息在当前部分的最后一行
upload_info = parse_speed_info(part[-1])
# 打印格式化后的结果
print(f"Download{i} speed of {filename} is {stringify_speed_info(*download_info)}.")
print(f"Upload{i} speed of {filename} is {stringify_speed_info(*upload_info)}.")
print() # 每处理完一个文件的一部分后打印空行
# ...3.6 完整代码
将上述所有部分整合，形成完整的 Python 脚本。
#!/usr/bin/python3
from pathlib import Path
# 定义常量：每个逻辑部分的行数
LINES_PER_PART = 8
def chunks(arr, chunk_size):
"""
将列表 arr 分割成大小为 chunk_size 的块。
"""
result = []
for i in range(0, len(arr), chunk_size):
result.append(arr[i:i+chunk_size])
return result
def parse_speed_info(string):
"""
从速度信息字符串中解析出速度值（浮点数）和单位。
示例输入: "Download: 1200.58 Mbit/s\n"
示例输出: (1200.58, "Mbit/s")
"""
speed_info_list = string.split()[1::] # 忽略 "Download:" 或 "Upload:" 部分
return (
float(speed_info_list[0]),
speed_info_list[1].strip() # 移除单位末尾的换行符
)
def stringify_speed_info(speed, unit):
"""
根据速度值和单位，生成格式化的输出字符串。
"""
if speed == 0:
return "zero"
elif unit == "Mbit/s" and speed < 600.0:
return f"less than 600 {unit}"
else:
return f"{speed} {unit}"
def main():
"""
主函数：执行文件查找、解析和结果输出。
"""
# 递归查找当前目录及其子目录下所有 .txt 文件
result = list(Path(".").rglob("*.txt"))
for filename in result:
try:
with open(filename, 'r') as file:
lines = file.readlines()
# 将文件内容按固定行数分块
parts = chunks(lines, LINES_PER_PART)
# 遍历每个逻辑部分
for i, part in enumerate(parts, 1):
# 检查部分行数是否符合预期，避免索引错误
if len(part) != LINES_PER_PART:
print(f"Warning: File {filename}, Part {i} has unexpected line count ({len(part)} lines). Skipping.")
continue
# 下载速度信息在当前部分的倒数第三行
download_info = parse_speed_info(part[-3])
# 上传速度信息在当前部分的最后一行
upload_info = parse_speed_info(part[-1])
# 打印格式化后的结果
print(f"Download{i} speed of {filename} is {stringify_speed_info(*download_info)}.")
print(f"Upload{i} speed of {filename} is {stringify_speed_info(*upload_info)}.")
print() # 每处理完一个文件的一部分后打印空行
except Exception as e:
print(f"Error processing file {filename}: {e}", file=sys.stderr)
if __name__ == "__main__":
main()4. 注意事项与总结
- 文件结构一致性： 本方案的核心假设是所有 .txt 文件都具有严格一致的内部结构，即每个逻辑部分都包含固定数量的行（LINES_PER_PART），并且速度信息始终位于这些部分的固定相对位置。如果文件结构不一致，例如行数不固定或信息位置变化，则需要更复杂的解析逻辑（如基于正则表达式或关键词搜索）。
- 错误处理： 在完整代码中加入了简单的 try-except 块来捕获文件打开或处理过程中的异常，并增加了对 part 长度的检查，以提高程序的健壮性。在实际应用中，可以根据需求增加更详细的错误日志记录和处理机制。
- 内存效率： 对于非常大的文件，file.readlines() 会一次性将所有内容加载到内存中。如果文件大小达到 GB 级别，这可能会导致内存溢出。对于此类情况，可以考虑逐行读取文件，并使用一个缓冲区来构建每个 part，或者使用生成器（yield）来优化 chunks 函数，实现惰性加载。但对于本例中每个文件只有两部分、每部分 8 行的场景，readlines() 是完全可接受的。
- 可扩展性： 通过将解析和格式化逻辑封装在单独的函数中，代码的可读性和可维护性得到了显著提升。如果未来需要提取其他类型的数据或改变输出格式，只需修改相应的函数即可，而无需改动主处理逻辑。
通过上述方法，我们能够高效、准确地从结构化文本文件中提取所需数据，并根据业务逻辑进行灵活的展示。这种分而治之、模块化设计的思想在处理复杂数据解析任务时尤为重要。
