在网络运维和数据采集场景中，网络速度数据往往会按照时间、区域等维度拆分存储到不同层级的目录和结构化文本文件中，要完整提取这些数据，就需要用到递归遍历和文本解析的相关技术。
场景说明与文件结构
假设我们的数据存储结构如下，根目录下有多个按日期命名的文件夹，每个日期文件夹下又有按区域命名的子文件夹，子文件夹中包含网络速度的结构化文本文件：
network_data/
├── 2024-05-01/
│ ├── east_region/
│ │ ├── speed_001.txt
│ │ └── speed_002.txt
│ └── west_region/
│ └── speed_001.txt
└── 2024-05-02/
└── north_region/
└── speed_001.txt
每个speed_xxx.txt文件的内容为固定格式的结构化文本，示例如下：
timestamp: 1714521600 upload_speed: 12.5Mbps download_speed: 85.3Mbps latency: 23ms packet_loss: 0.2%
递归遍历目录实现
我们首先编写递归函数遍历所有子目录，找到所有的.txt目标文件。递归的核心逻辑是判断当前路径是文件还是目录，如果是目录则继续递归调用自身，如果是目标文件则加入待处理列表。
import os
def recursive_find_txt(root_dir):
"""递归遍历目录，返回所有txt文件的路径列表"""
txt_file_list = []
# 遍历当前目录下的所有条目
for entry in os.listdir(root_dir):
entry_path = os.path.join(root_dir, entry)
# 如果是目录，递归处理
if os.path.isdir(entry_path):
sub_files = recursive_find_txt(entry_path)
txt_file_list.extend(sub_files)
# 如果是txt文件，加入列表
elif entry.endswith(".txt"):
txt_file_list.append(entry_path)
return txt_file_list
# 测试遍历逻辑
if __name__ == "__main__":
root_path = "./network_data"
all_txt_files = recursive_find_txt(root_path)
print(f"共找到{len(all_txt_files)}个txt文件")
for file_path in all_txt_files:
print(file_path)
结构化文本解析逻辑
找到所有目标文件后，我们需要解析每个文件的内容，提取网络速度相关的字段。由于文件结构是固定的键值对形式，我们可以逐行读取文件，按冒号分割键值对，再存储到字典中。
def parse_speed_file(file_path):
"""解析单个网络速度文件，返回数据字典"""
speed_data = {}
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
# 按冒号分割键值对
if ":" in line:
key, value = line.split(":", 1)
speed_data[key.strip()] = value.strip()
return speed_data
# 测试解析逻辑
if __name__ == "__main__":
test_file = "./network_data/2024-05-01/east_region/speed_001.txt"
data = parse_speed_file(test_file)
print("解析结果：")
for k, v in data.items():
print(f"{k}: {v}")
完整流程整合
我们将递归遍历和文本解析的逻辑整合起来，实现批量处理所有文件并汇总数据的功能，同时可以添加简单的数据校验，过滤掉格式异常的文件。
import os
def recursive_find_txt(root_dir):
"""递归遍历目录，返回所有txt文件的路径列表"""
txt_file_list = []
for entry in os.listdir(root_dir):
entry_path = os.path.join(root_dir, entry)
if os.path.isdir(entry_path):
sub_files = recursive_find_txt(entry_path)
txt_file_list.extend(sub_files)
elif entry.endswith(".txt"):
txt_file_list.append(entry_path)
return txt_file_list
def parse_speed_file(file_path):
"""解析单个网络速度文件，返回数据字典"""
speed_data = {}
required_keys = ["timestamp", "upload_speed", "download_speed", "latency"]
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
if ":" in line:
key, value = line.split(":", 1)
speed_data[key.strip()] = value.strip()
# 校验必要字段是否齐全
for key in required_keys:
if key not in speed_data:
return None
return speed_data
def batch_process_speed_data(root_dir):
"""批量处理所有速度文件，返回汇总数据列表"""
all_files = recursive_find_txt(root_dir)
result_list = []
for file_path in all_files:
file_data = parse_speed_file(file_path)
if file_data:
# 添加文件路径信息方便溯源
file_data["file_path"] = file_path
result_list.append(file_data)
else:
print(f"文件{file_path}格式异常，已跳过")
return result_list
if __name__ == "__main__":
root_path = "./network_data"
all_speed_data = batch_process_speed_data(root_path)
print(f"成功解析{len(all_speed_data)}条有效数据")
# 打印前两条数据查看效果
for item in all_speed_data[:2]:
print(item)
注意事项与优化方向
在实际使用中，需要注意几个问题：首先是递归深度，如果目录层级过深可能会触发Python的递归深度限制，此时可以改用栈或队列实现迭代式遍历；其次是文件编码问题，如果文本文件不是utf-8编码，需要对应调整open函数的encoding参数；最后是性能问题，如果文件数量极多，可以考虑使用多线程并行解析提升处理效率。
这种递归遍历加结构化文本解析的思路不仅适用于网络速度数据，也可以推广到其他层级存储的键值对格式文本文件处理场景中，只需要调整解析部分的字段匹配逻辑即可。