如何在Python中读取三个文件、提取指定列到字典并合并过滤主文件

一、正确读取单文件并提取指定列到字典 #

你提供的示例代码存在几个问题：未导入csv模块、未初始化reader对象、print语句里的["col4"]写法错误（应为row["col4"]）。以下是正确的实现方式，通过csv.DictReader可以直接按列名访问数据：

import csv

# 定义函数：读取文件，提取指定列存入字典（键为key_col，值为value_col）
def extract_to_dict(file_path, key_col, value_col):
    result_dict = {}
    with open(file_path, "r", encoding="utf-8") as f:
        # 使用DictReader按列名读取数据
        reader = csv.DictReader(f)
        for row in reader:
            # 避免列名不存在引发的KeyError
            if key_col in row and value_col in row:
                result_dict[row[key_col]] = row[value_col]
    return result_dict

# 调用函数读取第一个文件
part1_dict = extract_to_dict("CSB/python/data/part1.txt", "col1", "col4")

二、合并多个提取后的字典 #

合并字典的方式取决于你对重复键的处理需求：

• 重复键覆盖：如果后面文件的键值需要覆盖前面的，用字典解包或update方法：

# 先读取另外两个文件的字典
part2_dict = extract_to_dict("CSB/python/data/part2.txt", "col1", "col5")
part3_dict = extract_to_dict("CSB/python/data/part3.txt", "col1", "col6")

# 方式1：字典解包创建新字典
merged_dict = {**part1_dict, **part2_dict, **part3_dict}

# 方式2：用update方法修改原字典
merged_dict = part1_dict.copy()
merged_dict.update(part2_dict)
merged_dict.update(part3_dict)

• 保留重复键的所有值：如果需要保留同一键对应的所有值，用collections.defaultdict存储列表：

from collections import defaultdict

merged_dict = defaultdict(list)
for single_dict in [part1_dict, part2_dict, part3_dict]:
    for key, val in single_dict.items():
        merged_dict[key].append(val)

三、用合并后的字典过滤主文件 #

假设主文件是CSV格式，我们可以根据合并字典的键（或值）过滤主文件的行，保留符合条件的数据并输出：

def filter_main_file(main_path, merged_dict, filter_col):
    filtered_data = []
    with open(main_path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        # 先保存表头
        filtered_data.append(reader.fieldnames)
        # 遍历主文件每行，匹配合并字典的键
        for row in reader:
            if row[filter_col] in merged_dict:
                # 可选：将合并字典对应的值追加到当前行
                row["matched_value"] = merged_dict[row[filter_col]]
                filtered_data.append(row)
    
    # 将过滤结果写入新文件
    with open("filtered_main_result.txt", "w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=filtered_data[0])
        writer.writeheader()
        writer.writerows(filtered_data[1:])

# 调用函数：主文件路径、合并后的字典、用于匹配的列名
filter_main_file("CSB/python/data/main.txt", merged_dict, "col1")

内容的提问来源于stack exchange，提问作者curiouscoder