Python统计日志中流程块(Process Blocks)数量的实现方法

核心思路 #

嵌套流程的匹配完全等价于括号匹配逻辑，用栈做单次遍历即可实现，时间复杂度O(n)，内存占用极低，完全适配超大规模日志场景：

• 逐行扫描日志，提前编译正则提取每行的操作名、执行状态
• 遇到Executing状态的操作：如果当前栈为空，说明这是一个顶层流程的起点，记录起点信息；随后将操作名压入栈
• 遇到Closed状态的操作：弹出栈顶元素（合法日志结构下必然和当前操作名匹配）；如果弹出后栈为空，说明这是对应顶层流程的终点，记录完整块信息
• 遍历完成后结果列表的长度就是顶层块总数，同时自带每个块的起止描述，不需要二次扫描

代码实现 #

import re
from typing import List, Tuple

# 提前编译正则，提升大日志处理效率
LOG_PATTERN = re.compile(r'^\[[^\]]+\]\[([^\[]+)\[(Executing|Closed)\]\s*\]')

def count_top_level_blocks(log_file_path: str) -> Tuple[int, List[str]]:
    stack = []
    blocks = []
    current_top_start = None

    with open(log_file_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue  # 自动跳过空行，兼容示例里的展示空行、实际日志无空行的场景
            match = LOG_PATTERN.match(line)
            if not match:
                continue  # 跳过格式不匹配的异常行，提升鲁棒性
            op_name = match.group(1).strip()
            status = match.group(2)

            if status == 'Executing':
                if not stack:
                    # 栈空时入栈的是顶层流程
                    current_top_start = f"{op_name} [Executing]"
                stack.append(op_name)
            elif status == 'Closed':
                stack.pop()
                if not stack:
                    # 出栈后栈空，说明匹配到顶层流程终点
                    blocks.append(f"{current_top_start} to {op_name} [Closed]")
    
    return len(blocks), blocks

if __name__ == "__main__":
    # 替换为实际日志文件路径即可运行
    block_count, block_info = count_top_level_blocks("test.log")
    print(f"number_of_blocks = {block_count}")
    print("对应流程块说明：")
    for idx, desc in enumerate(block_info, 1):
        print(f"block {idx}: {desc}")

效果验证 #

用题目给出的日志样例测试，输出完全符合预期：

number_of_blocks = 3
对应流程块说明：
block 1: Open WebSite [Executing] to Open WebSite [Closed]
block 2: Login [Executing] to Login [Closed]
block 3: Read Data [Executing] to Read Data [Closed]

性能说明 #

• 正则提前编译，逐行流式读取文件，不需要把整个日志加载到内存，GB级日志也能稳定运行
• 内存占用仅和日志最大嵌套深度相关，和日志总大小无关
• 遍历过程中直接完成统计和信息记录，无多余计算开销

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