如何优化字符串搜索替换脚本性能？含XML批量处理需求

我有一个含两列的电子表格（约4000行）：

• 第一列是唯一的Tag Name字符串，但存在短串是长串严格子集的情况（例如e1\di\BC-B29hiTor和e1\di\BC-B29hiTorq）
• 第二列是对应的替换字符串

核心需求：

1. 不区分大小写匹配，优先匹配长串避免误替换短串
2. 对600余个XML文件执行批量搜索替换，替换时保留原文件其余内容的大小写

我基于Trie结构写了搜索日志脚本，但性能极差：处理100个XML文件耗时约5小时，添加替换逻辑后性能会更差，求性能优化方案。

主脚本 #

# Import required libs
import pandas as pd
import os
import openpyxl
from Trie import Trie
import logging

logging.basicConfig(filename='searchResults.log', level=logging.INFO, format='%(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p')

# Load the hmi tags into a Trie data structure and the addresses into an array.
# The Trie accepts a (key, value) pair, where key is the tag and value is the
# index of the associated array.
df_HMITags = pd.read_excel('Tags.xlsx')
logging.info('Loaded excel file')
HMITags = Trie()
addresses = []
for i in df_HMITags.index:
    HMITags.insert(str(df_HMITags[' Tag Name'][i]).lower(), i)
    addresses.append(str(df_HMITags[' Address'][i]))

# Assign directory
directory = 'Graphics'

# Iterate over the files in the directory
for filename in os.listdir(directory):
    file = os.path.join(directory, filename)
    
    # Checking if it is a file
    if os.path.isfile(file):
        logging.info('Searching File: ' + str(filename))
        print('Searching File:', filename)
        
        # Open the file
        with open(file,'r') as fp:
            
            # Search the file, one line at a time.
            lines = fp.readlines()
            lineNumber = 1
            for line in lines:
                if lineNumber %10 == 0:
                    print('Searching line number:', lineNumber)
                #logging.debug('Searching Line: ' + str(lineNumber))
                #print('Searching Line:', lineNumber)
                # Convert to lower case, as this will simplify searching.
                lineLowered = line.lower()
                
                # Iterate through the line searching for various tags.
                searchString = ''
                potentialMatchFound = False
                charIndex = 0
                while charIndex < len(lineLowered):
                    #logging.debug('charIndex: ' + str(charIndex))
                    #print('charIndex = ', charIndex, '---------------------------------------')
                    searchString = searchString + lineLowered[charIndex]
                    searchResults = HMITags.query(searchString)
                    
                    #if lineNumber == 2424:
                    ###print('searchString:', searchString)
                    ###print('searchResults length:', len(searchResults))
                    
                    # If the first char being searched does not return any results, move on to the next char.
                    if len(searchResults) > 0:
                        potentialMatchFound = True
                        ###print('Potential Match Found:', potentialMatchFound)
                    elif len(searchResults) == 0 and potentialMatchFound:
                        ###print('Determining if exact match exists')
                        # Remove the last char from the string.
                        searchString = searchString[:-1]
                        searchResults = HMITags.query(searchString)
                        
                        #Determine if an exact match exists in the search results
                        exactMatchFound = False
                        exactMatchIndex = 0
                        while exactMatchIndex < len(searchResults) and not exactMatchFound:
                            if searchString == searchResults[exactMatchIndex][0]:
                                exactMatchFound = True
                            exactMatchIndex = exactMatchIndex + 1
                        
                        if exactMatchFound:
                            logging.info('Match Found! File: ' + str(filename) + ' Line Number: ' + str(lineNumber) + ' Column: ' + str(charIndex - len(searchString) + 1) + ' HMI Tag: ' + searchString)
                            print('Found:', searchString)
                            charIndex = charIndex - 1
                        else:
                            ###print('Not Found:', searchString)
                            charIndex = charIndex - len(searchString)
                            
                        searchString = ''
                        potentialMatchFound = False
                    else:
                        searchString = ''
                    charIndex = charIndex + 1
                    
                lineNumber = lineNumber + 1

Trie实现代码 #

class TrieNode:
    """A node in the trie structure"""

    def __init__(self, char):
        # the character stored in this node
        self.char = char

        # whether this can be the end of a key
        self.is_end = False
        
        # The value from the (key, value) pair that is to be stored.
        # (if this node's is_end is True)
        self.value = 0

        # a dictionary of child nodes
        # keys are characters, values are nodes
        self.children = {}
        
class Trie(object):
    """The trie object"""

    def __init__(self):
        """
        The trie has at least the root node.
        The root node does not store any character
        """
        self.root = TrieNode("")
    
    def insert(self, key, value):
        """Insert a key into the trie"""
        node = self.root
        
        # Loop through each character in the key
        # Check if there is no child containing the character, create a new child for the current node
        for char in key:
            if char in node.children:
                node = node.children[char]
            else:
                # If a character is not found,
                # create a new node in the trie
                new_node = TrieNode(char)
                node.children[char] = new_node
                node = new_node
        
        # Mark the end of a key
        node.is_end = True
        
        # Set the value from the (key, value) pair.
        node.value = value
        
    def dfs(self, node, prefix):
        """Depth-first traversal of the trie
        
        Args:
            - node: the node to start with
            - prefix: the current prefix, for tracing a
                key while traversing the trie
        """
        if node.is_end:
            self.output.append((prefix + node.char, node.value))
        
        for child in node.children.values():
            self.dfs(child, prefix + node.char)
        
    def query(self, x):
        """Given an input (a prefix), retrieve all keys stored in
        the trie with that prefix, sort the keys by the number of 
        times they have been inserted
        """
        # Use a variable within the class to keep all possible outputs
        # As there can be more than one key with such prefix
        self.output = []
        node = self.root
        
        # Check if the prefix is in the trie
        for char in x:
            if char in node.children:
                node = node.children[char]
            else:
                # cannot found the prefix, return empty list
                return []
        
        # Traverse the trie to get all candidates
        self.dfs(node, x[:-1])

        # Sort the results in reverse order and return
        return sorted(self.output, key = lambda x: x[1], reverse = True)

方案1：改用预编译正则表达式（最直接有效） #

正则引擎经过高度优化，可通过按字符串长度倒序排列实现长串优先匹配，配合re.IGNORECASE实现不区分大小写匹配，完美适配需求：

1. 读取Excel数据后，构建小写标签到替换值的映射表
2. 将所有标签按长度从长到短排序，保证长串先被匹配
3. 用re.escape()处理标签中的特殊字符，拼接成正则模式并预编译
4. 用re.sub()配合回调函数完成替换，保留原匹配文本的大小写格式

示例代码片段：

import re
import pandas as pd

# 读取标签数据，构建映射表
df = pd.read_excel('Tags.xlsx', usecols=[' Tag Name', ' Address'])
tag_map = {str(row[' Tag Name']).lower(): str(row[' Address']) for _, row in df.iterrows()}

# 按标签长度倒序排列，确保长串优先匹配
sorted_tags = sorted(tag_map.keys(), key=lambda x: -len(x))
# 构建预编译正则模式，忽略大小写
pattern = re.compile('|'.join(re.escape(tag) for tag in sorted_tags), re.IGNORECASE)

# 替换回调：保留原匹配的大小写，返回对应替换值
def replace_match(match):
    return tag_map[match.group(0).lower()]

# 批量处理XML文件
directory = 'Graphics'
for filename in os.listdir(directory):
    file_path = os.path.join(directory, filename)
    if not os.path.isfile(file_path):
        continue
    
    # 一次性读取文件内容，减少IO开销
    with open(file_path, 'r', encoding='utf-8') as f:
        content = f.read()
    
    # 执行替换并写入
    new_content = pattern.sub(replace_match, content)
    with open(file_path, 'w', encoding='utf-8') as f:
        f.write(new_content)

方案2：优化现有Trie实现（若坚持使用Trie） #

当前Trie的性能瓶颈在于频繁的DFS查询和逐字符回溯，可做如下优化：

1. 修改Trie遍历逻辑：遍历文本时，直接追踪Trie节点，同时记录当前遇到的最长有效标签（is_end=True的节点），当无法继续匹配时，直接使用最长标签替换并跳过对应长度的字符，避免逐字符回溯
2. 减少字符串拼接：避免searchString的频繁拼接，直接通过Trie节点追踪匹配路径
3. 批量读取文件：一次性读取整个文件内容，替代逐行处理，减少IO操作次数

方案3：多进程并行处理 #

XML文件之间相互独立，可利用多核CPU并行处理：

• 使用multiprocessing.Pool将文件列表分配给多个进程同时处理
• 注意日志写入的线程安全问题，可让每个进程单独生成日志文件，后续再合并

其他小优化 #

• 日志精简：减少logging.info()的调用频率，或改用异步日志库降低IO开销
• Excel读取优化：指定usecols参数只读取需要的两列，减少内存占用和读取时间
• 编码统一：处理文件时指定明确编码（如utf-8），避免系统默认编码导致的兼容性问题

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