首页 / 资讯中心 / 文章详情

手写Python日志查看工具:实现关键字检索与实时跟踪

手写Python日志查看工具:实现关键字检索与实时跟踪 ★ FEATURED ARTICLE
简介面向日常开发、测试与运维场景的轻量级日志查看工具主程序体积小巧但处理性能出色作者实测可流畅打开4G以上日志文件试验46G大文件亦无压力。压缩包共含5个文件整体仅551KB其中exe为主程序免安装可直接运行chm为详细帮助手册html与txt分别提供版本说明与使用指引manifest用于程序运行配置结构紧凑、开箱即用。目前已有938人学习下载适合需要快速定位系统日志、排查线上异常的技术人员使用相比常规文本编辑器它在超大日志场景下能显著减少加载等待帮助更快定位关键信息提升排障效率。配套说明文档覆盖常见操作与查看技巧部署到Windows本机即可上手同时保留原始发布记录便于追溯工具版本与更新内容是一款兼顾轻量与性能的实用日志处理小工具。1. 日志查看工具与其翻文件不如写个顺手的 logviewer排查线上问题时最磨人的往往不是问题本身而是日志动不动几百 MB打开卡死搜索慢得像在等编译。我最早也是用编辑器硬翻后来发现时间全耗在等待和误看上。当时手头正好需要一个日志查看工具干脆自己写了一个轻量版 logviewer核心诉求很简单——能像 tail -f 一样实时跟踪能快速按关键字检索还能对时间窗口做过滤。这套脚本后来在几次现场事故里帮我省下大量时间也让我把日志处理的很多细节摸透了。如果你也经常跟 production log 打交道或者在做运维、后端开发时总被日志检索折磨这套思路和代码可以直接落地复用。日志工具的价值不在功能多而在命中准、开箱快。本质上它要解决三件事怎么读大文件不卡、怎么在持续写入的文件里抓新增内容、怎么过滤出真正关心的那几行。下面几章我会从选型开始把整个工具拆开讲清楚。2. 先搞清需求与选型日志工具的核心是读、追、滤动手写之前我没急着堆功能而是先列了需求清单。毕竟日志查看工具这种东西做大了就成了日志平台做小了就是个脚本。对我来说最核心的诉求是能快速定位问题而不是把日志做成可视化大盘。2.1 为什么不用现成的 tail 和 grepLinux 下 tail 和 grep 确实能干不少事但组合起来有几个痛点一是 tail -f 只能看实时输出想回看前面的内容还得另开命令二是 grep 不带上下文命中行前后发生了什么看不到排查异常时往往得再翻大文件三是多文件检索时想看哪个文件哪一行输出里的文件归属标记不够直观。这些都是我在用一段时间后逐渐感受到的。后来需求明确才决定自己写一个脚本化工具核心就是三个动作跟踪、检索、过滤。如果你只是临时看一眼日志那 tail 和 grep 完全够用但如果日志是持续写入的而且你需要在同一个界面里完成“先回看、再跟踪、再检索”的循环那一个专门的 logviewer 脚本更顺手。我的场景是现场排查接口超时和崩溃需要同时看应用日志和系统日志还要能在发现问题时立刻按关键字回捞上下文。2.2 技术选型Python 脚本零依赖优先选 Python 而不是 Go 或 Node理由很实际目标机器上基本都有 Python 环境哪怕没有装一个也比重编译 Go 二进制快另一个原因是日志处理的逻辑大多是文本解析和正则匹配Python 的标准库足够覆盖不必引入第三方包。零依赖意味着拷贝过去就能跑不折腾环境。版本上我用了 Python 3.6主要因为它对 pathlib 和 f-string 的支持较好写起来干净。脚本本身不要求图形界面做成命令行工具SSH 到机器上直接用。下面是脚本的初始骨架包含参数解析和入口逻辑#!/usr/bin/env python3 # logviewer.py - 轻量日志查看工具 # 用法示例: # python3 logviewer.py --file app.log --follow # python3 logviewer.py --file app.log --grep ERROR --context 5 import argparse import time from pathlib import Path def parse_args(): parser argparse.ArgumentParser(description轻量日志查看工具 logviewer) parser.add_argument(--file, -f, requiredTrue, help日志文件路径) parser.add_argument(--grep, -g, help关键字或正则表达式命中才输出) parser.add_argument(--context, -C, typeint, default0, help命中行前后各显示几行) parser.add_argument(--follow, -F, actionstore_true, help实时跟踪文件新增内容) parser.add_argument(--start, help只显示该时间之后的日志行格式如 2024-01-01 00:00:00) parser.add_argument(--encoding, defaultutf-8, help文件编码默认 utf-8常需要改为 gbk) return parser.parse_args() def main(): args parse_args() path Path(args.file) if not path.exists(): print(f[错误] 文件不存在: {path}) return if __name__ __main__: main()这段代码把参数全部集中在一个函数里后续扩展只要在这里加选项就行。--file是必填项其他都可选。--encoding很多时候容易被忽略但中文日志用 GBK 写入时默认 utf-8 读取会直接报错或乱码这个参数是踩坑后的补救。--context控制命中行的上下文行数排查异常时很有用。选型逻辑就这么定下来了先用 argparse 管参数再用 pathlib 处理路径读取文件时按行迭代配合 seek 和 tell 做增量追踪。这样整个工具没有外部依赖拷贝到任何一台有 Python 的机器上就能跑对生产环境的侵入也最小。3. 实现关键字检索与上下文输出日志查看工具的核心功能很多场景下查日志就是查关键字报错。比如接口超时、数据库连接失败、空指针异常这些都是明确的关键字。所以我把检索做成了第一优先级功能而不是先做 tail -f。3.1 按行迭代与大文件读取的取舍读取大日志文件第一反应可能是readlines()但这会把整个文件一次性加载进内存500 MB 的日志直接吃掉几百兆内存得不偿失。我的做法是直接用for line in f按行迭代Python 内部会做缓冲读取既省内存又不会太慢。另外打开文件时用errorsignore处理编码异常避免读到半个字符时中断。def scan_file(path, encodingutf-8, keywordNone, context0, start_timeNone): 扫描日志文件按条件过滤行 results [] buffer [] import re pattern re.compile(keyword) if keyword else None with open(path, r, encodingencoding, errorsignore) as f: for line in f: line line.rstrip(\n) # 时间过滤: 如果指定了 start_time则按字符串前缀比较 if start_time and line start_time: continue # 关键字过滤 if pattern and not pattern.search(line): if buffer: buffer.append(line) if len(buffer) context: buffer.pop(0) continue # 命中处理 if pattern: results.append({line: line, match: True, buffer: buffer.copy()}) else: results.append({line: line, match: False, buffer: []}) buffer [] return results这段逻辑的重点在buffer的使用它是一个长度为 context 的滑动窗口保存命中行之前的最近 N 行这样输出命中行时前面几行上下文也能一起展示。时间过滤用的是字符串比较前提是日志行开头的时间戳格式统一比如2024-01-01 10:00:00这种格式字符串对比和日期时间对比结果一致。但如果遇到混合格式的时间戳字符串比较就不可靠后面的坑会细说。正则匹配这里用的是re.compile(keyword)也就是说--grep参数支持正则。如果你只想搜纯关键字比如ERROR正则也能匹配但要注意正则元字符的转义。像想搜Caused by:冒号没问题但如果搜java.lang.NullPointerException也没问题最怕的是搜包含(或[的内容直接用会报错。后面我会给一个自动转义的开关。3.2 上下文输出与多文件命中标记扫描结果准备好了接下来是输出格式。单文件场景下直接打印行内容就够了但多文件场景比如同时看应用日志和系统日志输出时至少要带文件路径或文件名否则根本分不清哪行来自哪个文件。我实现时给输出加了一个前缀选项。def print_results(results, path, show_filenameFalse, colorTrue): 格式化输出扫描结果 RED \033[31m if color else RESET \033[0m if color else for item in results: if show_filename: prefix f[{path}] else: prefix # 输出上下文行 for ctx_line in item[buffer]: print(f{prefix}{ctx_line}) # 输出命中行命中部分标红 if item[match] and color: import re match re.search(r(ERROR|Exception|Timeout), item[line]) if match: start, end match.span() print(f{prefix}{item[line][:start]}{RED}{item[line][start:end]}{RESET}{item[line][end:]}) else: print(f{prefix}{item[line]}) else: print(f{prefix}{item[line]})这里的 ANSI 颜色转义在大多数 Linux 终端里有效Windows 的 CMD 老版本不支持但 PowerShell 和 Windows Terminal 基本都支持了。show_filename参数适合多文件同时扫描时打开单文件时关掉避免每行刷目录路径。用的时候是这样python3 logviewer.py --file app.log --grep ERROR|Exception --context 3 --start 2024-06-01 10:00:00 --encoding gbk注意--grep里传入了正则表达式ERROR|Exception命中任意一个都会输出。--context 3表示命中行前后各显示 3 行--start过滤时间--encoding gbk应对中文日志。这套组合基本覆盖了日常排查 70% 以上的场景剩下的核心操作是实时跟踪。4. 实时跟踪从轮询到 seek/tell 增量读取日志工具的另一半核心功能是实时跟踪相当于增强版 tail -f。生产环境排查时经常要边复现问题边盯日志如果只是重复跑扫描命令每次都要从头读文件几百 MB 的文件会越来越慢。所以实时跟踪必须做增量读取——只读新增部分。4.1 记录文件偏移量与编码问题增量的关键是用tell()记录当前文件位置下次读取时seek()到上次位置接着读。Python 文件对象在文本模式下用 tell/seek 会有编码相关的限制尤其是 GBK 等多字节编码seek 到任意字节位置可能切到半个汉字导致读出来乱码。我的做法是把文件切成二进制模式读取然后手动解码按字节偏移控制位置这样对多字节编码更可控。def follow_file(path, encodingutf-8, keywordNone, sleep_interval0.5): 实时跟踪文件新增内容, 支持编码容错 with open(path, rb) as f: # 先跳到文件末尾 f.seek(0, 2) current_pos f.tell() print(f[跟踪] 从文件末尾开始, 偏移量{current_pos}) while True: line f.readline() if line: try: text line.decode(encoding, errorsignore) except UnicodeDecodeError: text line.decode(encoding, errorsreplace) text text.rstrip(\r\n) if keyword: import re if re.search(keyword, text): print(f{text}) else: print(text) else: # 没有新内容, 睡一会再试 time.sleep(sleep_interval) # 处理文件被轮转(truncate)的情况 if f.tell() current_pos: print([提示] 检测到文件被截断或轮转, 重新从头读取) f.seek(0, 0) current_pos f.tell()这段代码里有两个值得注意的细节第一readline()在二进制模式下返回 bytes手动decode时用errorsignore宁可丢一个坏字符也不能让整个跟踪中断第二文件轮转检测——如果日志服务按天滚动或按大小截断文件文件大小会变小此时f.tell()的位置会大于文件尾部异常情况下要回退到文件头重新读。日志轮转是运维场景里的高频问题不处理的话跟踪会停在旧位置表现为日志不更新实际上新日志已经写进新文件了。4.2 不带文件头读取与 tail -F 视角对比关注点是输出格式。系统自带的tail -F能处理文件轮转但没法做关键字过滤和上下文。我把过滤和跟踪合在一起这样复现问题时可以只盯关键字相关的行避免刷屏。使用方式python3 logviewer.py --file app.log --follow --grep ERROR --encoding gbk这样做之后终端只会滚动输出包含ERROR的行其他日志刷得再快也不干扰视线。如果只想看新增的全部内容去掉--grep即可。sleep_interval参数控制轮询频率默认 0.5 秒实时性要求高的场景可以调到 0.1 秒但会增加 CPU 占用低频场景可以调到 1 秒以上。跟踪和检索打通之后日志查看工具的核心闭环已经完成。但有几个坑是我实际用了很久才总结出来的单独开一章记录。5. 日志查看工具使用避坑文件编码、性能与时间过滤工具能用和工具好用是两码事。这章列几个我在实际使用中踩过的坑每条都是“现象 → 原因 → 解决”的结构希望能帮你少走弯路。5.1 中文日志乱码现象用默认 utf-8 读 GBK 编码的日志输出全是乱码甚至直接报UnicodeDecodeError。原因早期 Windows 下很多 Java 应用默认用 GBK 写日志Linux 下则多用 utf-8。机器环境不一样编码就乱了。解决我把--encoding参数暴露给调用方脏活让使用者定。另外读取时统一errorsignore保证即使编码判断错了也只是丢几个乱码字符不会让整个工具中断。# 读取时的最终兜底 with open(path, rb) as f: raw f.read() try: text raw.decode(args.encoding) except UnicodeDecodeError: text raw.decode(args.encoding, errorsreplace)有人可能会想自动检测编码但标准库里没有可靠的检测方案第三方包 chardet 又要额外依赖。我的建议是在启动参数里明确传编码实在不确定时先file app.log看一眼再决定。5.2 大文件首次扫描太慢现象300 MB 的日志文件首次扫描要等十几秒甚至更久。原因逐行正则匹配本身不慢慢的是频繁打印。终端滚动输出是有成本的尤其是带 ANSI 颜色码时。解决先把结果收集到内存列表扫描完成后再统一输出另外--grep匹配时使用re.compile预编译正则避免每行都重新编译。compiled_pattern re.compile(keyword) # 只编译一次 for line in f: if compiled_pattern.search(line): results.append(line)如果文件大到几百 MB 甚至几 GB我的建议是配合 split 或 grep 先粗筛一遍再用这个工具精查别硬扛。5.3 时间过滤用了字符串比较结果不对现象指定--start 2024-06-01 10:00:00结果早于该时间的行也打印出来了。原因日志行的时间戳格式不统一有的带毫秒有的不带有的前面带级别前缀比如INFO 2024-06-01 10:00:00这样字符串前缀比较就失效了。解决先把时间戳从行里抽取出来再统一解析成datetime对象比较而不是直接比较整个行字符串。from datetime import datetime def extract_time(line): 从日志行里抽出时间字符串, 没有则返回 None parts line.split() for part in parts: if part and part[0].isdigit() and : in part: try: return datetime.strptime(part, %Y-%m-%d %H:%M:%S) except ValueError: continue return None严格的做法是把时间解析独立成函数然后和 start_time 做 datetime 比较。这样即使日志行里混着毫秒、时区也不容易出错。5.4 grep 正则包含特殊字符报错现象搜索java.util.regex.PatternSyntaxException时报错re.error: nothing to repeat。原因关键是搜的内容里有(、)、[、]这类正则元字符直接作为正则传进去会被解析成语法结构。解决加一个--literal开关开启后把关键字用re.escape包一层纯字面量匹配。if args.literal: pattern re.compile(re.escape(keyword)) else: pattern re.compile(keyword)需要全文搜索时建议优先开--literal只有确实要用正则匹配时再关掉这个开关能省掉一大半报错。6. 把 logviewer 扩展成多文件检索的批处理入口单文件场合下这个工具已经够用但真正让我觉得它“值了”的是把它扩展成多文件检索的批处理入口。排查分布式系统问题时日志分散在好多台机器上光用单文件模式就得反复切路径效率很低。我的做法是写一个包装脚本接收目录或通配符批量调用 logviewer 的核心函数统一输出。6.1 目录扫描与结果合并# multi_logviewer.py - 多文件日志检索入口 import argparse import glob from pathlib import Path # 复用上一章的 scan_file 和 print_results from logviewer_core import scan_file, print_results def scan_directory(directory, pattern, keyword, context0, start_timeNone, encodingutf-8): 扫描目录下所有匹配 pattern 的日志文件 files glob.glob(str(Path(directory) / pattern)) if not files: print(f[提示] 目录 {directory} 下没有匹配 {pattern} 的文件) return for file_path in sorted(files): # 跳过非日志文件 if not file_path.endswith((.log, .txt)): continue results scan_file(file_path, encodingencoding, keywordkeyword, contextcontext, start_timestart_time) if results: print(f\n {file_path} ) print_results(results, file_path, show_filenameTrue, colorTrue) if __name__ __main__: parser argparse.ArgumentParser() parser.add_argument(--dir, requiredTrue, help日志目录) parser.add_argument(--pattern, default*.log, help文件名通配符) parser.add_argument(--grep, requiredTrue) parser.add_argument(--context, typeint, default3) parser.add_argument(--encoding, defaultutf-8) args parser.parse_args() scan_directory(args.dir, args.pattern, args.grep, args.context, encodingargs.encoding)使用时python3 multi_logviewer.py --dir /var/log/myapp/ --pattern *.log --grep ConnectTimeout --context 2 --encoding gbk这个入口会把所有匹配文件扫描出来的结果按文件分组输出命中文件优先显示。很多情况下不是每个文件都有命中空文件会跳过只展示有内容的文件避免刷屏。实际用下来排查跨服务问题时比一台台机器手动翻效率高出不少。最后说一个习惯。从那以后我每次排查日志问题时都会先把日志文件编码确认一遍再决定是用--encoding utf-8还是--encoding gbk同时随手带上--grep和--context而不是先全量铺开再慢慢看。这个工具后来改了又改但最开始那版骨架一直保留着因为它的结构足够简单参数足够明确遇到新场景直接在上面加功能就行。希望这份拆解对你有帮助也欢迎你按自己的场景改出一版更趁手的 logviewer。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助?
咨询建站