如何使用pyparsing逐行读取大文件并在匹配到目标后停止读取
方案1:自行维护滑动缓冲区
无需修改现有解析规则,逻辑简单易调试:
from pyparsing import ParseException # 你已定义的解析表达式 final 无需修改 match_result = None # 可根据目标内容的最大预估行数调整该值,保证缓冲区能装下完整的匹配块 lookback_lines = 30 recent_lines = [] with open(p, errors="ignore", newline='') as f: for line in f: recent_lines.append(line) # 仅保留最近N行内容,控制内存占用 if len(recent_lines) > lookback_lines: recent_lines.pop(0) buffer = "".join(recent_lines) try: matches = final.search_string(buffer, 1) if matches: match_result = matches[0] break except ParseException: # 内容未完整,继续读取下一行 continue # 后续处理match_result即可
方案2:使用pyparsing原生增量解析器
pyparsing 3.0及以上版本提供了IncrementalParser原生支持增量输入,无需自行维护缓冲区:
from pyparsing import IncrementalParser, ParseException # 包装现有解析表达式,开启增量解析能力 incr_parser = IncrementalParser(final) match_result = None with open(p, errors="ignore", newline='') as f: for line in f: try: # 逐行喂入解析器 incr_parser.feed(line) # 尝试解析完整内容 match_result = incr_parser.parse() break except ParseException: # 内容不完整,继续喂入下一行 continue incr_parser.close() # 后续处理match_result即可
注意事项
- 解析规则中避免使用
StringEnd这类严格依赖输入流末尾的语法,否则会等待全部文件读取完成才会返回结果 - 若解析规则包含明确的目标内容结束标记(如示例中的
}),解析器会在读取到完整块后立刻终止,无多余IO操作
内容的提问来源于stack exchange,提问作者SimpleOne
相关产品推荐
相关产品推荐

