如何串联Python处理流程 实现配对提取、文件合并、词频统计一体化

场景说明 #

• 待处理数据集：600余个存储「两个字符串配对+浮点数值」格式的txt文件，单条内容样例为JAK2,LONP1,3.941044066754e-10，文件命名格式为「识别前缀+数值后缀」，例如AV8GF00_0.01、AV8FG00_0.0001。
• 原有流程痛点：三个独立处理环节需要依次创建中间文件夹、切换工作目录存储结果，操作链路冗长、磁盘读写开销大；尝试将三个环节直接嵌套时，出现循环层级错位导致的逻辑错误。

原独立环节核心逻辑边界 #

环节1：单文件数据清洗 #

• 读取单文件原始内容，拆分所有条目后过滤含小数点的浮点值元素
• 剩余字符串按两两一组拆分为配对列表
• 循环提取全局众数字符串，留存匹配的配对结果，移除已匹配的配对条目直至列表为空

环节2：同前缀文件内容合并 #

• 提取所有文件的文件名前6位作为分组标识
• 将同前缀的所有文件内容合并为单一组目标文件

此前嵌套报错的核心诱因：将文件分组逻辑错误嵌套进了单文件内部的配对提取循环，每处理一个众数条目就重复遍历全目录生成合并文件，属于逻辑层级完全错位。

环节3：字符串频次统计 #

• 读取合并完成的文件内容，拆分所有字符串条目
• 统计每个字符串的出现频次，输出降序排列的计数列表，格式样例为[('WAS', 11), ('GTX1', 5)]

一体化无中间存储实现方案 #

核心设计思路 #

• 取消所有中间结果落盘、文件夹切换的操作，所有中间计算结果全部在内存对象中流转，仅最后一步将最终结果写入磁盘
• 修正嵌套逻辑：把文件前缀分组的逻辑从单文件内部循环中提至全局外层，消除重复遍历的冗余

可直接运行的流程代码 #

from pathlib import Path
from collections import Counter
from statistics import multimode

# 路径初始化，无需反复切换工作目录
INPUT_DIR = Path("存放原始600个txt文件的根路径")
OUTPUT_DIR = Path("最终统计结果的输出路径")
OUTPUT_DIR.mkdir(exist_ok=True, parents=True)

# 全局预处理：提前按文件名前6位完成分组，替代第二环节的重复遍历逻辑
all_original_files = list(INPUT_DIR.glob("*.txt"))
file_groups = {}
for file_path in all_original_files:
    prefix = file_path.name[:6]
    file_groups.setdefault(prefix, []).append(file_path)

# 逐组执行全流程处理
for group_prefix, group_file_list in file_groups.items():
    # 内存归集容器：存放整个分组内所有清洗后的字符串，替代第一环节写盘、第二环节读盘的操作
    group_collected_items = []

    # 遍历组内每个原始文件，执行第一环节的清洗逻辑
    for single_file in group_file_list:
        with open(single_file, "r", encoding="utf-8") as f:
            # 拆分所有内容条目
            raw_entries = [
                item for line in f.read().splitlines() 
                for item in line.split(",")
            ]
            # 过滤带小数点的浮点值
            str_entries = [x for x in raw_entries if "." not in x]
            # 两两拆分为配对列表
            pair_list = [
                str_entries[i:i+2] 
                for i in range(0, len(str_entries), 2)
            ]

            # 循环提取众数、留存配对、移除已匹配条目
            while len(pair_list) > 0:
                current_mode = multimode(
                    item for sublist in pair_list 
                    for item in sublist
                )
                # 将当前轮次的众数条目直接存入内存归集容器
                group_collected_items.extend(current_mode)
                # 匹配所有包含当前众数的配对
                matched_pairs = [
                    pair for pair in pair_list
                    for mode_item in current_mode
                    if mode_item in pair
                ]
                # 从剩余配对列表中移除已匹配的条目
                for pair_a, pair_b in matched_pairs:
                    pair_list = [
                        remain_pair for remain_pair in pair_list
                        if (pair_a not in remain_pair) or (pair_b not in remain_pair)
                    ]

    # 直接基于内存中归集的清洗结果执行第三环节的频次统计，跳过中间文件读写
    freq_result = Counter(group_collected_items).most_common()

    # 仅在流程最后写入最终结果文件
    result_path = OUTPUT_DIR / f"{group_prefix}.txt"
    with open(result_path, "w", encoding="utf-8") as f:
        f.write(str(freq_result))

原嵌套错误的针对性修复 #

• 逻辑层级修正：将原来写在单文件内部while循环里的文件分组逻辑上移到全局初始化阶段，避免每提取一次众数就重复遍历全目录、重复生成合并文件的无效操作
• 冗余环节移除：原来需要「第一环节写本地txt→第二环节读本地txt合并」的两次磁盘读写，现在通过内存容器直接传递处理结果，减少IO开销
• 路径管理优化：全程使用Path对象管理文件路径，不再反复调用os.chdir切换工作目录，避免路径错位

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