Approach: stream the file line-by-line, parse each newline-delimited JSON entry, filter entries with level == "error" whose timestamp is within the last 24 hours, count per component in a dict, and finally print the top 5. The script is single-pass and keeps only counts in memory (O(number of unique components)).python
#!/usr/bin/env python3
import sys
import json
from datetime import datetime, timedelta, timezone
from collections import Counter
import heapq
def parse_iso8601(s):
# Accepts RFC3339/ISO8601 like "2025-01-01T12:34:56Z" or with offset.
if s.endswith("Z"):
s = s[:-1] + "+00:00"
try:
return datetime.fromisoformat(s)
except Exception:
return None
def main(path):
now = datetime.now(timezone.utc)
cutoff = now - timedelta(days=1)
counts = Counter()
with open(path, 'r', encoding='utf-8') as f:
for lineno, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
# malformed JSON: skip (could log to stderr)
print(f"Skipping malformed JSON at line {lineno}", file=sys.stderr)
continue
ts_raw = obj.get("timestamp")
level = obj.get("level")
comp = obj.get("component")
if not (ts_raw and level and comp):
print(f"Skipping incomplete entry at line {lineno}", file=sys.stderr)
continue
ts = parse_iso8601(ts_raw)
if ts is None:
print(f"Skipping unparsable timestamp at line {lineno}", file=sys.stderr)
continue
# Make timezone-aware; assume naive timestamps are UTC
if ts.tzinfo is None:
ts = ts.replace(tzinfo=timezone.utc)
if level.lower() == "error" and ts >= cutoff and ts <= now:
counts[comp] += 1
for comp, cnt in heapq.nlargest(5, counts.items(), key=lambda x: x[1]):
print(f"{comp}\t{cnt}")
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: script.py /path/to/log.ndjson", file=sys.stderr)
sys.exit(2)
main(sys.argv[1])
