在日常的数据处理任务中，我们经常需要从大量具有相似结构但内容各异的文本文件中提取特定信息。本教程将以一个具体场景为例：从一个包含多层子目录的系统中，查找所有 .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 等）输出不同的信息。
为了高效且健壮地解决此类问题，我们采用以下策略：
首先，我们需要定义文件中每个逻辑部分的行数以及文件包含的逻辑部分数量。然后，使用 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()
# ... 后续处理 ...为了将文件的所有行按固定大小（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) # 将文件内容分块
# ...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() # 移除单位末尾的换行符
)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 单位，此处已在代码中体现。
在 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() # 每处理完一个文件的一部分后打印空行
# ...将上述所有部分整合，形成完整的 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()通过上述方法，我们能够高效、准确地从结构化文本文件中提取所需数据，并根据业务逻辑进行灵活的展示。这种分而治之、模块化设计的思想在处理复杂数据解析任务时尤为重要。
以上就是使用 Python 递归处理结构化文本文件并提取特定数据的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
213
收藏
