Golem Yapapi Log Summarizer

golem_log_parser.py

import json
import argparse
from pathlib import Path
from collections import defaultdict
from datetime import datetime

# Small log-parsing CLI with no external dependencies
# Save file somewhere, see options with `python golem_log_parser.py -h`

class LogLine:
    """Individual log event parser for easier manipulation"""

    def __init__(self, line) -> None:

        line_split = line.split(" ")
        self.timestamp = datetime.strptime(
            line_split[0].lstrip("[").split(".")[0], "%Y-%m-%dT%H:%M:%S"
        )
        self.log_level = line_split[1]
        self.log_type = line_split[2].rstrip("]")
        self.content = " ".join(line_split[3:])

    def __str__(self) -> str:
        return (
            f"{self.timestamp} - {self.log_level} - {self.log_type}\n{self.content}\n"
        )


class LogParser:
    """Lazy-ish logfile parser, create LogLine objects and assembles/displays stats"""

    def __init__(self, log_path: Path, level: str, sum_chars: int) -> None:
        self.log_path = log_path
        self.sum_chars = sum_chars
        self.levels = []
        self.unparsable = []
        self.line_objs = []

        # Init summary tree
        self.sum_tree = defaultdict(lambda: defaultdict(lambda: defaultdict(int)))

        # Set log-level
        for l in ["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"]:
            if l == level:
                self.levels.append(l)
                break
            else:
                self.levels.append(l)

    def parse_log(self) -> None:
        with open(self.log_path, "r") as lf:
            lines = lf.read().split("\n[")
            for line in lines:
                try:
                    lo = LogLine(line)
                    self.line_objs.append(lo)
                except ValueError:
                    print("Found unparsable line!")
                    self.unparsable.append(line)

                # Update summary tree
                if lo.log_level in self.levels:
                    self.sum_tree[lo.log_level][lo.log_type][
                        lo.content[: self.sum_chars]
                    ] += 1

    def get_runtime(self):
        self.line_objs.sort(key=lambda x: x.timestamp)
        return self.line_objs[-1].timestamp - self.line_objs[0].timestamp

    def print_stats(self):
        print(f"Runtime: {self.get_runtime()}")
        print(f"Total log events: {len(self.line_objs)}")
        print(json.dumps(self.sum_tree, indent=4))


argparser = argparse.ArgumentParser()
argparser.add_argument("path", type=str, help="Path to the logfile to parse")
argparser.add_argument(
    "log_level",
    type=str,
    help="Only summarize logs belonging to this level and above",
    default="WARNING",
)
argparser.add_argument(
    "content_chars",
    type=int,
    help="Number of characters of content the summary is based on, increase for more granularity",
    default=15,
)
args = argparser.parse_args()

logparser = LogParser(args.path, args.log_level, args.content_chars)
logparser.parse_log()
logparser.print_stats()