Python脚本性能优化：检查labels.sql中TEXTID的引用状态

你的脚本性能瓶颈非常明显——每处理一个TEXTID就完整遍历一次整个目录结构，2000个TEXTID就会遍历2000次，这是导致10分钟耗时的核心原因。下面是针对性的优化思路和代码实现：

核心优化思路：反向遍历，一次扫描所有文件 #

把原逻辑的"遍历TEXTID→逐个查文件"改成"扫描所有文件一次→收集所有被使用的TEXTID→批量判断INSERT语句"，这样文件系统只需要遍历一次，时间复杂度从O(N*M)降到O(N+M)（N是TEXTID数量，M是文件数量）。

优化后的代码 #

import os
import re
import shutil

current_dir_location = os.getcwd()
labels_script_location = os.path.join(current_dir_location, 'SQL-scripts/labels.sql')
copy_location = os.path.join(current_dir_location, 'SQL-scripts/labelsCopy.sql')

# 复制原文件做修改
shutil.copy2(labels_script_location, copy_location)

# 读取所有INSERT语句
with open(copy_location, 'r', encoding='UTF-8') as file:
    insert_statements_from_labels = file.readlines()

# 编译提取TEXTID的正则
text_id_pattern = re.compile(r"'[^']+'\s*,\s*'([^']+)'")

def extract_all_text_ids(statements):
    """从所有INSERT语句中提取需要检查的TEXTID集合"""
    text_ids = set()
    for stmt in statements:
        match = text_id_pattern.search(stmt)
        if match:
            text_ids.add(match.group(1))
    return text_ids

def collect_used_text_ids(references_dir, target_text_ids):
    """遍历所有文件一次，收集被引用的TEXTID"""
    used_ids = set()
    target_count = len(target_text_ids)
    
    for root, dirs, files in os.walk(references_dir):
        # 直接修改dirs列表，排除不需要遍历的目录
        dirs[:] = [d for d in dirs if d not in ('node_modules', 'SQL-scripts')]
        
        for file_name in files:
            # 可选：跳过非文本文件，减少IO操作
            if file_name.endswith(('.bin', '.log', '.lock', '.exe')):
                continue
                
            file_path = os.path.join(root, file_name)
            try:
                with open(file_path, 'r', encoding='utf-8') as file:
                    content = file.read()
                    # 检查当前文件包含哪些目标TEXTID
                    for tid in target_text_ids - used_ids:
                        if tid in content:
                            used_ids.add(tid)
                            # 提前终止：如果所有TEXTID都找到，直接返回
                            if len(used_ids) == target_count:
                                return used_ids
            except (UnicodeDecodeError, PermissionError):
                continue
    return used_ids

def get_text_id(insert_statement):    
    match = text_id_pattern.search(insert_statement)
    return match.group(1) if match else None

def search_decide():
    lines_deleted = 0
    used_insert_statements = []
    unused_insert_statement = {}
    
    # 1. 提取所有需要检查的TEXTID
    all_text_ids = extract_all_text_ids(insert_statements_from_labels)
    print(f"共提取到 {len(all_text_ids)} 个TEXTID")
    
    # 2. 一次性扫描所有文件，收集被使用的TEXTID
    used_text_ids = collect_used_text_ids(current_dir_location, all_text_ids)
    print(f"找到 {len(used_text_ids)} 个被引用的TEXTID")
    
    # 3. 批量判断每个INSERT语句是否保留
    for index, insert_statement in enumerate(insert_statements_from_labels):
        text_id = get_text_id(insert_statement)
        if text_id:
            if text_id in used_text_ids:
                used_insert_statements.append(insert_statement)
            else:
                unused_insert_statement[index] = insert_statement
                lines_deleted += 1
        else:
            # 无TEXTID的语句默认保留，可根据需求调整
            used_insert_statements.append(insert_statement)
    
    # 写入处理后的文件
    with open(copy_location, 'w', encoding='utf-8') as file:
        file.writelines(used_insert_statements)
    
    # 输出未使用的语句
    for index, statement in unused_insert_statement.items():
        print(f"未使用的INSERT语句：\n{index}: {statement}\n")
    print("脚本执行完成")
    print(f"共删除 {lines_deleted} 行")

if __name__ == "__main__":
    search_decide()

额外优化点 #

1. 文件类型过滤：在遍历文件时跳过二进制文件、日志文件等非文本文件，减少不必要的IO操作；
2. 提前终止扫描：当所有TEXTID都找到引用后，直接停止遍历文件，节省时间；
3. 字符串查找替代正则：如果TEXTID是精确匹配（不需要正则模糊匹配），用tid in content比正则更快；
4. 大文件处理：对于超大文件，不要一次性读取全部内容，改为按行读取并检查，降低内存占用；
5. 并行处理（可选）：如果文件数量极多，可尝试用multiprocessing并行遍历文件，但要注意避免文件IO竞争，收益可能不如前几点明显。

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