本文详细介绍了如何从多层嵌套目录中的python文件里提取包含特定格式的字典数据，并利用`os.walk`遍历文件系统、`ast.literal_eval`安全解析字典字符串，最终将其高效整合并构建成一个统一的pandas dataframe。教程涵盖了文件路径获取、内容读取、字典解析及dataframe构建与合并的完整流程。
在数据处理和自动化任务中，我们经常会遇到需要从散布在文件系统各处的配置文件或数据文件中提取结构化信息的情况。特别是在复杂的项目结构中，如果数据以Python字典的形式存储在多个.py文件中，并需要将其汇总到一个统一的Pandas DataFrame中进行分析，这就需要一套系统性的处理方法。本教程将指导您完成这一过程，包括文件系统遍历、文件内容读取、字典字符串的安全解析以及最终的DataFrame构建与合并。
1. 遍历文件系统以定位目标文件
首先，我们需要一种机制来扫描指定根目录下的所有子目录，找到包含目标字典的Python文件。Python的os模块提供了强大的文件系统交互功能，其中os.walk()函数是实现这一目标的核心工具。它会递归地遍历目录树，为每个目录生成一个三元组：(root, dirs, files)，分别代表当前目录路径、子目录列表和文件列表。
假设我们的目标文件是form.py，并且它们可能位于任意深度的子目录中。我们可以使用os.walk()结合文件后缀名过滤来找到这些文件。
import osimport pandas as pdimport 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) # 打印找到的文件路径列表
2. 读取文件内容并提取字典字符串
找到所有目标文件后，下一步是打开每个文件，读取其内容，并从中识别出我们需要的字典字符串。在许多场景中，字典可能以变量赋值的形式存在，例如 def_options = {‘key1’: ‘value1’, ‘key2’: ‘value2’}。我们需要解析这些行，提取出字典的纯字符串表示。
立即学习“Python免费学习笔记（深入）”；
笔目鱼英文论文写作器
写高质量英文论文，就用笔目鱼
87 查看详情
为了确保提取的准确性，我们可以结合字符串查找和分割操作。例如，如果已知字典包含特定的键（如”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)
3. 安全解析字典字符串为Python字典
直接使用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)} 个字典对象。")
4. 构建并合并Pandas DataFrame
有了所有解析好的字典对象列表，最后一步是将其转换为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 osimport pandas as pdimport astdef 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 shutilif os.path.exists(base_dir): shutil.rmtree(base_dir)
注意事项与总结
字典格式一致性： 本教程假设所有form.py文件中的字典都具有相同的键结构。如果键不一致，pd.DataFrame会自动填充NaN值。字典识别逻辑： 识别包含字典的行 (if “name” in stripped_line and “age” in stripped_line and “def_options =” in stripped_line:) 是关键。您需要根据实际文件中字典的定义方式来调整此处的条件判断和字符串分割逻辑。例如，如果字典没有前缀变量名，或者有其他独特的标识符，需要相应修改。错误处理： 在实际应用中，文件可能损坏、格式不符或不包含预期的字典。代码中包含了try-except块来捕获ast.literal_eval可能引发的ValueError或SyntaxError，以及其他潜在的异常，从而提高程序的健壮性。编码： 读取文件时指定encoding=”utf-8″是一个好习惯，可以避免因编码问题导致的错误。性能优化： 对于极其庞大的文件系统和海量文件，可以考虑使用多线程或多进程来并行处理文件，以提高数据提取效率。数据存储替代方案： 如果form.py文件仅用于存储数据而非执行代码，考虑使用更适合数据存储的格式，如JSON或YAML，它们有更完善的解析库，且通常更明确地表示数据结构。
通过上述步骤，您可以有效地从分散在多层目录中的Python文件中提取结构化字典数据，并将其整合到一个易于分析和操作的Pandas DataFrame中。
以上就是高效导入多目录Python文件字典至Pandas DataFrame的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/588403.html
微信扫一扫 
支付宝扫一扫 
