编程题1:JSON 日志管道
This commit is contained in:
@@ -0,0 +1,4 @@
|
||||
# Python 缓存
|
||||
__pycache__/
|
||||
*.pyc
|
||||
*.pyo
|
||||
+47
@@ -0,0 +1,47 @@
|
||||
import json
|
||||
import os
|
||||
|
||||
|
||||
def analyze_log(filepath: str) -> dict:
|
||||
"""解析 JSON Lines 日志文件,返回统计结果。"""
|
||||
# 结果字典,先填好默认值(也就是空结果)
|
||||
result = {
|
||||
"total": 0,
|
||||
"by_level": {},
|
||||
"by_user": {},
|
||||
"last_error": None,
|
||||
}
|
||||
|
||||
# 文件不存在时,直接返回空结果,不报错
|
||||
if not os.path.exists(filepath):
|
||||
return result
|
||||
|
||||
with open(filepath, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
# 空行跳过
|
||||
if not line:
|
||||
continue
|
||||
|
||||
# 某行不是合法 JSON 就跳过这一行,不影响整体解析
|
||||
try:
|
||||
data = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
# 成功解析一条,总数加一
|
||||
result["total"] += 1
|
||||
|
||||
# 按 level 计数
|
||||
level = data["level"]
|
||||
result["by_level"][level] = result["by_level"].get(level, 0) + 1
|
||||
|
||||
# 按 user 计数
|
||||
user = data["user"]
|
||||
result["by_user"][user] = result["by_user"].get(user, 0) + 1
|
||||
|
||||
# 记录最后一条 ERROR 的 message
|
||||
if level == "ERROR":
|
||||
result["last_error"] = data["message"]
|
||||
|
||||
return result
|
||||
Reference in New Issue
Block a user