本文详细介绍了如何从多层嵌套目录中的python文件里提取包含特定格式的字典数据，并利用`os.walk`遍历文件系统、`ast.literal_eval`安全解析字典字符串，最终将其高效整合并构建成一个统一的pandas dataframe。教程涵盖了文件路径获取、内容读取、字典解析及dataframe构建与合并的完整流程。
在数据处理和自动化任务中，我们经常会遇到需要从散布在文件系统各处的配置文件或数据文件中提取结构化信息的情况。特别是在复杂的项目结构中，如果数据以Python字典的形式存储在多个.py文件中，并需要将其汇总到一个统一的Pandas DataFrame中进行分析，这就需要一套系统性的处理方法。本教程将指导您完成这一过程，包括文件系统遍历、文件内容读取、字典字符串的安全解析以及最终的DataFrame构建与合并。
首先，我们需要一种机制来扫描指定根目录下的所有子目录，找到包含目标字典的Python文件。Python的os模块提供了强大的文件系统交互功能，其中os.walk()函数是实现这一目标的核心工具。它会递归地遍历目录树，为每个目录生成一个三元组：(root, dirs, files)，分别代表当前目录路径、子目录列表和文件列表。
假设我们的目标文件是form.py，并且它们可能位于任意深度的子目录中。我们可以使用os.walk()结合文件后缀名过滤来找到这些文件。
import os
import pandas as pd
import ast # 用于安全地评估字符串中的Python字面量
# 定义起始搜索路径，可以替换为您的实际根目录
# 例如：base_path = os.environ["JUPYTER_ROOT"] + "/charts/"
base_path = "./charts_data/" # 示例路径，请根据实际情况修改
target_files = []
for root, dirs, files in os.walk(base_path):
for file in files:
if file.endswith("form.py"):
file_path = os.path.join(root, file)
target_files.append(file_path)
print(f"找到 {len(target_files)} 个目标文件。")
# print(target_files) # 打印找到的文件路径列表找到所有目标文件后，下一步是打开每个文件，读取其内容，并从中识别出我们需要的字典字符串。在许多场景中，字典可能以变量赋值的形式存在，例如 def_options = {'key1': 'value1', 'key2': 'value2'}。我们需要解析这些行，提取出字典的纯字符串表示。
立即学习“Python免费学习笔记（深入）”；
为了确保提取的准确性，我们可以结合字符串查找和分割操作。例如，如果已知字典包含特定的键（如"name"和"age"），我们可以利用这些键来定位包含字典的行。
extracted_dictionaries = []
for file_path in target_files:
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
stripped_line = line.strip()
# 假设字典行包含 'name' 和 'age' 键，并且以 'def_options =' 开头
# 您需要根据实际的字典定义格式调整此处的判断逻辑
if "name" in stripped_line and "age" in stripped_line and "def_options =" in stripped_line:
try:
# 分割字符串，获取等号右侧的字典部分
dictionary_str = stripped_line.split("=", 1)[1].strip()
extracted_dictionaries.append(dictionary_str)
break # 假设每个文件只包含一个目标字典，找到后即可跳出当前文件循环
except IndexError:
print(f"警告: 无法从文件 {file_path} 的行中正确分割字典字符串: {stripped_line}")
except Exception as e:
print(f"警告: 处理文件 {file_path} 的行时发生错误: {e} - 行内容: {stripped_line}")
# print(f"提取到 {len(extracted_dictionaries)} 个字典字符串。")
# for d_str in extracted_dictionaries:
# print(d_str)直接使用Python的eval()函数来解析从文件中读取的字符串存在安全风险，因为它会执行任意代码。为了安全地将字典字符串转换为实际的Python字典对象，我们应该使用ast.literal_eval()。这个函数只能评估包含Python字面量（字符串、数字、元组、列表、字典、布尔值和None）的字符串，而不会执行任意代码。
parsed_dictionaries = []
for dict_str in extracted_dictionaries:
try:
# 使用 ast.literal_eval 安全地解析字典字符串
dictionary_obj = ast.literal_eval(dict_str)
if isinstance(dictionary_obj, dict): # 确保解析结果确实是字典
parsed_dictionaries.append(dictionary_obj)
else:
print(f"警告: 解析结果不是字典类型: {dict_str}")
except (ValueError, SyntaxError) as e:
print(f"错误: 无法解析字典字符串 '{dict_str}': {e}")
print(f"成功解析 {len(parsed_dictionaries)} 个字典对象。")有了所有解析好的字典对象列表，最后一步是将其转换为Pandas DataFrame。由于所有字典具有相同的键，我们可以直接将字典列表传递给pd.DataFrame构造函数，或者逐个字典转换为DataFrame行并进行合并。对于大量字典，通常建议先收集所有字典到一个列表，然后一次性构建DataFrame，这比反复连接DataFrame更高效。
# 确保所有字典具有相同的键，这是构建统一DataFrame的前提
if not parsed_dictionaries:
print("没有可用的字典来构建DataFrame。")
else:
# 方式一：直接从字典列表构建DataFrame（推荐）
final_dataframe = pd.DataFrame(parsed_dictionaries)
print("\n最终构建的Pandas DataFrame (方式一):")
print(final_dataframe.head())
# 方式二：逐个字典构建DataFrame并合并（适用于特殊情况，效率可能略低）
# all_dfs = []
# for d in parsed_dictionaries:
# # 将单个字典转换为DataFrame的一行
# df_row = pd.DataFrame([d])
# all_dfs.append(df_row)
#
# if all_dfs:
# final_dataframe_concat = pd.concat(all_dfs, ignore_index=True)
# print("\n最终构建的Pandas DataFrame (方式二):")
# print(final_dataframe_concat.head())
# else:
# print("没有可用的字典来构建DataFrame。")
import os
import pandas as pd
import ast
def import_dictionaries_to_dataframe(base_path: str, filename_pattern: str = "form.py", dict_key_indicators: tuple = ("name", "age"), dict_var_name: str = "def_options =") -> pd.DataFrame:
"""
从多层嵌套目录的Python文件中提取字典，并构建Pandas DataFrame。
Args:
base_path (str): 开始搜索的根目录路径。
filename_pattern (str): 目标Python文件的文件名模式，例如 "form.py"。
dict_key_indicators (tuple): 用于识别包含目标字典的行的键指示器，例如 ("name", "age")。
dict_var_name (str): 字典变量在文件中的赋值前缀，例如 "def_options ="。
Returns:
pd.DataFrame: 包含所有提取字典数据的Pandas DataFrame。
如果未找到或解析任何字典，则返回空的DataFrame。
"""
target_files = []
for root, dirs, files in os.walk(base_path):
for file in files:
if file.endswith(filename_pattern):
file_path = os.path.join(root, file)
target_files.append(file_path)
extracted_dictionaries_data = []
for file_path in target_files:
with open(file_path, "r", encoding="utf-8") as f:
for line in f:
stripped_line = line.strip()
# 检查行是否包含所有指示键和变量名
if all(key in stripped_line for key in dict_key_indicators) and dict_var_name in stripped_line:
try:
# 提取字典字符串
dictionary_str = stripped_line.split(dict_var_name, 1)[1].strip()
# 安全解析字典字符串
dictionary_obj = ast.literal_eval(dictionary_str)
if isinstance(dictionary_obj, dict):
extracted_dictionaries_data.append(dictionary_obj)
break # 假设每个文件只包含一个目标字典
else:
print(f"警告: 文件 {file_path} 中解析结果不是字典类型: {dictionary_str}")
except (ValueError, SyntaxError) as e:
print(f"错误: 无法解析文件 {file_path} 中的字典字符串 '{dictionary_str}': {e}")
except IndexError:
print(f"警告: 文件 {file_path} 的行 '{stripped_line}' 无法正确分割字典字符串。")
except Exception as e:
print(f"警告: 处理文件 {file_path} 的行时发生未知错误: {e} - 行内容: {stripped_line}")
if extracted_dictionaries_data:
return pd.DataFrame(extracted_dictionaries_data)
else:
print("未找到或成功解析任何字典数据。")
return pd.DataFrame()
# 示例使用：
# 假设您的项目结构如下：
# ./charts_data/
# ├── ahc_visits/
# │ └── booking_breakdown_per_age_group/
# │ └── form.py (内容：def_options = {'name': 'Alice', 'age': 30, 'city': 'NY'})
# └── other_charts/
# └── some_report/
# └── form.py (内容：def_options = {'name': 'Bob', 'age': 25, 'city': 'LA'})
# 创建一些模拟文件用于测试
os.makedirs("./charts_data/ahc_visits/booking_breakdown_per_age_group", exist_ok=True)
with open("./charts_data/ahc_visits/booking_breakdown_per_age_group/form.py", "w") as f:
f.write("def_options = {'name': 'Alice', 'age': 30, 'city': 'New York'}\n")
os.makedirs("./charts_data/other_charts/some_report", exist_ok=True)
with open("./charts_data/other_charts/some_report/form.py", "w") as f:
f.write("def_options = {'name': 'Bob', 'age': 25, 'city': 'Los Angeles'}\n")
os.makedirs("./charts_data/another_folder", exist_ok=True)
with open("./charts_data/another_folder/form.py", "w") as f:
f.write("def_options = {'name': 'Charlie', 'age': 35, 'city': 'Chicago', 'occupation': 'Engineer'}\n")
# 调用函数
base_dir = "./charts_data/"
df = import_dictionaries_to_dataframe(base_dir, dict_key_indicators=("name", "age"), dict_var_name="def_options =")
if not df.empty:
print("\n最终生成的DataFrame:")
print(df)
else:
print("DataFrame为空。")
# 清理模拟文件
import shutil
if os.path.exists(base_dir):
shutil.rmtree(base_dir)通过上述步骤，您可以有效地从分散在多层目录中的Python文件中提取结构化字典数据，并将其整合到一个易于分析和操作的Pandas DataFrame中。
以上就是高效导入多目录Python文件字典至Pandas DataFrame的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
812
收藏
