大体积TXT文件Python检索代码优化方案咨询

核心优化方向及实现 #

1. 改用进程池替代线程池 #

Python线程池受GIL限制，对于大文件的IO+CPU混合操作，进程池能更好利用多核CPU，避免线程切换的额外开销，提升并行处理效率。

2. 预转换查询词为小写，减少重复计算 #

提前将查询词统一转成小写，行内容仅做一次小写转换用于匹配，避免循环内重复执行字符串转换操作。

3. 按块读取文件，降低IO调用频率 #

逐行读取大文件会产生大量系统IO调用，改用64KB块读取再分割成行，同时处理跨行内容，显著减少IO次数，提升读取速度。

4. 用生成器逐步返回结果，降低内存占用 #

将原函数的列表存储改为生成器yield返回结果，避免大文件匹配结果过多导致的内存溢出，同时让主线程实时拿到结果，消除“未响应”的错觉。

5. 修复匹配与格式化逻辑 #

原代码中格式化函数仅能高亮完整query字符串，优化后支持对每个查询词单独高亮，适配多词分散匹配的场景。

优化后的完整代码 #

import os
import concurrent.futures

def format_matching_line(line, query_words):
    # 对每个查询词单独高亮，支持多词分散匹配
    formatted_line = line
    for word in query_words:
        lower_line = formatted_line.lower()
        idx = lower_line.find(word.lower())
        if idx != -1:
            original_word = formatted_line[idx:idx+len(word)]
            formatted_line = formatted_line.replace(
                original_word,
                f"<span style='background-color:#685581'>{original_word}</span>",
                1
            )
    return formatted_line

def search_in_file(directory, filename, query_words):
    # 预转换查询词为小写，避免重复计算
    query_lower = [word.lower() for word in query_words]
    filepath = os.path.join(directory, filename)
    buffer = ""
    line_number = 0

    # 按64KB块读取文件，配合buffer处理跨行内容
    with open(filepath, "r", encoding="utf-8", buffering=1024*1024) as file:
        while True:
            chunk = file.read(65536)
            if not chunk:
                # 处理最后剩余的buffer内容
                if buffer:
                    line_number +=1
                    line_stripped = buffer.strip()
                    if all(word in line_stripped.lower() for word in query_lower):
                        yield (filename, f"Line {line_number}: {line_stripped}")
                break
            lines = chunk.split('\n')
            lines[0] = buffer + lines[0]
            buffer = lines.pop()
            for line in lines:
                line_number +=1
                line_stripped = line.strip()
                if all(word in line_stripped.lower() for word in query_lower):
                    yield (filename, f"Line {line_number}: {line_stripped}")

def search(query, directory):
    if not query or not directory:
        return
    query_words = query.split()
    # 使用进程池，默认利用所有CPU核心
    with concurrent.futures.ProcessPoolExecutor() as executor:
        future_to_file = {
            executor.submit(search_in_file, directory, filename, query_words): 
            filename for filename in os.listdir(directory) if filename.endswith(".txt")
        }
        for future in concurrent.futures.as_completed(future_to_file):
            filename = future_to_file[future]
            try:
                # 实时输出当前文件的匹配结果，避免等待所有文件完成
                print(f"Results in {filename}:")
                print('-'*70)
                for _, line in future.result():
                    formatted_line = format_matching_line(line, query_words)
                    print(formatted_line)
                    print('-'*70)
                print()
            except Exception as e:
                print(f"Error while searching in {filename}: {e}")

if __name__ == "__main__":
    query = input("Введите текст для поиска: ")
    directory = input("Введите путь к папке с файлами: ")
    search(query, directory)

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