如何降低Regex处理百万份文本时的计算成本

问题背景 #

当前使用正则+NLTK实现了提取含特定词汇的句子及其前后相邻句子的功能，代码可正常运行，但单份TXT文件处理耗时20秒，需处理约百万份文件，急需在保证结果一致的前提下缩短耗时。现有思路为：先提取包含目标词汇的段落，再用NLTK对目标段落分句，最后提取目标句子及其前后句。

原演示代码：

import re, nltk

txt = '''There is widespread agreement that screening for breast cancer, when combined with appropriate follow-up, will reduce mortality from the disease. How we measure response to treatment is called the 5-year survival rate, or the percentage of people who live 5 years after being diagnosed with cancer. According to information provided by National Cancer Institute, Cancer stage at diagnosis, which refers to extent of a cancer in the body, determines treatment options and has a strong influence on the length of survival. In general, if the cancer is found only in the part of the body where it started it is localized (sometimes referred to as stage 1). If it has spread to a different part of the body, the stage is regional or distant . The earlier female breast cancer is caught, the better chance a person has of surviving five years after being diagnosed. For female breast cancer, 62.5 are diagnosed at the local stage. The 5-year survival for localized female breast cancer is 98.8 . It decreases from 98.8 to 85.5 after the cancer has spread to the lymph nodes (stage 2), and to 27.4 
(stage 4) after it has spread to other organs such as the lung, liver or brain. A major problem with current detection methods is that studies have shown that mammography does not detect 10 -20 of breast cancers that are detected by physical exam alone, which may be attributed to a falsely negative mammogram. 

Breast cancer screening is generally recommended as a routine part of preventive healthcare for women over the age of 20 (approximately 90 million in the United States). Besides skin cancer, breast cancer is the most commonly diagnosed cancer among U.S. women. For these women, the American Cancer Society (ACS) has published guidelines for breast cancer screening including: (i) monthly breast self-examinations for all women over the age of 20; (ii) a clinical breast exam (CBE) every three years for women in their 20s and 30s; (iii) a baseline mammogram for women by the age of 40; and (iv) an annual mammogram for women age 40 or older (according to the American College of Radiology). Unfortunately, the U.S. Preventive Task Force Guidelines have stirred confusion by recommending biennial screening mammography for women ages 50-74. 

Each year, approximately eight million women in the United States require diagnostic testing for breast cancer due to a physical symptom, such as a palpable lesion, pain or nipple discharge, discovered through self or physical examination (approximately seven million) or a non-palpable lesion detected by screening x-ray mammography
(approximately one million). Once a physician has identified a suspicious lesion in a woman's breast, the physician may recommend further diagnostic procedures, including a diagnostic x-ray mammography, an ultrasound study, a magnetic resonance imaging procedure, or a minimally invasive procedure such as fine needle aspiration or large core needle biopsy. In each case, the potential benefits of additional diagnostic testing must be balanced against the costs, risks and discomfort to the patient associated with undergoing the additional procedures. 
'''

target_words = ['risks', 'discomfort', 'surviving', 'risking', 'risks', 'risky']
pattern = r'.*\b(?='+'|'.join(target_words) + r')\b.*'
target_paras = re.findall(pattern, txt, re.IGNORECASE)

# Function to extract sentences containing any target word and its neighbor sentences
def UncertaintySentences (paragraph):
    sent_token = nltk.tokenize.sent_tokenize(paragraph)
    keepsents = []

    for i, sentence in enumerate(sent_token):
        # sentences contain any target word
        if re.search(pattern, sentence, re.IGNORECASE) != None:
            try:
                if i==0: # first sentence in a para, keep it and the one next to it
                    keepsents.extend([sent_token[i], sent_token[i+1]])
                elif i!=len(sent_token)-1: # sentence in the middle, keep it ant the ones before and next to it
                    keepsents.extend([sent_token[i-1], sent_token[i], sent_token[i+1]])
                else: # last sentence, keep it and the one before it
                    keepsents.extend([sent_token[i-1], sent_token[i]])
            except: # para with only one sentence
                keepsents = sent_token

    # drop duplicate sentences 
    del_dup = [] 
    [del_dup.append(x) for x in keepsents if x not in del_dup] 
    return(del_dup) 

for para in target_paras:
    uncertn_sents = UncertaintySentences(para)
    print(uncertn_sents)

优化建议及代码实现 #

核心优化点 #

• 预编译正则表达式：避免重复编译正则，减少每次匹配的性能开销
• 替换低效分句工具：用更轻量快速的分句库替代NLTK的sent_tokenize
• 重构处理流程：跳过段落提取步骤，直接对全文分句后定位目标句子，减少冗余操作
• 优化去重逻辑：用集合替代列表做存在性检查，将O(n)复杂度降为O(1)
• 并行处理文件：针对百万级文件，用多进程充分利用CPU核心，提升整体处理效率

优化后代码示例 #

import re
# 用更轻量的分句库，需先安装：pip install sentence-splitter
from sentence_splitter import split_text_into_sentences

# 预编译正则（仅执行一次）+ 去重目标词
target_words = ['risks', 'discomfort', 'surviving', 'risking', 'risks', 'risky']
unique_targets = list(set(target_words))
pattern = re.compile(r'\b(' + '|'.join(unique_targets) + r')\b', re.IGNORECASE)

def extract_target_sentences(text):
    # 快速分句：sentence-splitter比NLTK更快
    sentences = split_text_into_sentences(text=text, language='en')
    # 若坚持用NLTK，替换为：sentences = nltk.tokenize.sent_tokenize(text)
    
    # 用集合自动去重，无需后续手动处理
    keepsents = set()

    for i, sent in enumerate(sentences):
        if pattern.search(sent):
            keepsents.add(sent)
            # 添加前一句（存在则加）
            if i > 0:
                keepsents.add(sentences[i-1])
            # 添加后一句（存在则加）
            if i < len(sentences) - 1:
                keepsents.add(sentences[i+1])
    
    # 转回列表（若不需要顺序可直接返回集合）
    return list(keepsents)

# 测试
txt = '''[原文本内容，省略]'''
result = extract_target_sentences(txt)
print(result)

批量处理框架 #

针对百万份文件，使用多进程并行处理的示例代码：

import os
from multiprocessing import Pool

def process_single_file(file_path):
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            text = f.read()
        return extract_target_sentences(text)
    except Exception as e:
        print(f"处理文件{file_path}出错: {e}")
        return None

if __name__ == '__main__':
    # 假设所有文件在txt_files目录下
    file_dir = './txt_files'
    file_paths = [os.path.join(file_dir, fname) for fname in os.listdir(file_dir) if fname.endswith('.txt')]
    
    # 根据CPU核心数设置进程数
    with Pool(processes=os.cpu_count()) as pool:
        results = pool.map(process_single_file, file_paths)
    
    # 后续结果处理（如保存到文件/数据库）
    # ...

额外优化提示 #

• 文件IO优化：用with语句保证文件正确关闭；超大文件可考虑流式读取，避免一次性加载到内存
• 匹配逻辑简化：若文本格式规范，可将句子拆分为单词集合，直接检查与目标词集合的交集，跳过正则匹配
• 内存管控：批量处理时避免存储过多中间结果，及时写入输出介质

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