diff --git a/ise/ise_logparser/.python-version b/ise/ise_logparser/.python-version index 871f80a..39f773d 100644 --- a/ise/ise_logparser/.python-version +++ b/ise/ise_logparser/.python-version @@ -1 +1 @@ -3.12.3 +pypy@3.10.14 diff --git a/ise/ise_logparser/pyproject.toml b/ise/ise_logparser/pyproject.toml index 76f9e2b..07b89b6 100644 --- a/ise/ise_logparser/pyproject.toml +++ b/ise/ise_logparser/pyproject.toml @@ -9,6 +9,10 @@ dependencies = [ "parsy>=2.1", "icecream>=2.1.3", "python-dotenv>=1.0.1", + "mypy>=1.11.1", + "setuptools>=72.1.0", + "colorama>=0.4.6", + "types-colorama>=0.4.15.20240311", ] readme = "README.md" requires-python = ">= 3.8" diff --git a/ise/ise_logparser/requirements-dev.lock b/ise/ise_logparser/requirements-dev.lock index 91c16ad..bd2a046 100644 --- a/ise/ise_logparser/requirements-dev.lock +++ b/ise/ise_logparser/requirements-dev.lock @@ -14,15 +14,28 @@ asttokens==2.4.1 # via icecream colorama==0.4.6 # via icecream + # via ise-logparser executing==2.0.1 # via icecream icecream==2.1.3 # via ise-logparser +mypy==1.11.1 + # via ise-logparser +mypy-extensions==1.0.0 + # via mypy parsy==2.1 # via ise-logparser pygments==2.18.0 # via icecream python-dotenv==1.0.1 # via ise-logparser +setuptools==72.1.0 + # via ise-logparser six==1.16.0 # via asttokens +tomli==2.0.1 + # via mypy +types-colorama==0.4.15.20240311 + # via ise-logparser +typing-extensions==4.12.2 + # via mypy diff --git a/ise/ise_logparser/requirements.lock b/ise/ise_logparser/requirements.lock index 91c16ad..bd2a046 100644 --- a/ise/ise_logparser/requirements.lock +++ b/ise/ise_logparser/requirements.lock @@ -14,15 +14,28 @@ asttokens==2.4.1 # via icecream colorama==0.4.6 # via icecream + # via ise-logparser executing==2.0.1 # via icecream icecream==2.1.3 # via ise-logparser +mypy==1.11.1 + # via ise-logparser +mypy-extensions==1.0.0 + # via mypy parsy==2.1 # via ise-logparser pygments==2.18.0 # via icecream python-dotenv==1.0.1 # via ise-logparser +setuptools==72.1.0 + # via ise-logparser six==1.16.0 # via asttokens +tomli==2.0.1 + # via mypy +types-colorama==0.4.15.20240311 + # via ise-logparser +typing-extensions==4.12.2 + # via mypy diff --git a/ise/ise_logparser/src/main.py b/ise/ise_logparser/src/main.py index 589e7e1..9052ce3 100755 --- a/ise/ise_logparser/src/main.py +++ b/ise/ise_logparser/src/main.py @@ -1,93 +1,225 @@ #!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import logging +import os import re from pathlib import Path -from dotenv import dotenv_values +from typing import cast + import parsy as ps -from icecream import ic -import os -from pprint import pprint as print -from parsy import regex, seq +from colorama import Back, Fore, Style +from dotenv import dotenv_values +from parsy import Parser, regex, seq, string -# TODO: (akristip) refactor these to use `<<` instead of capture groups -parse_header = regex( - r"(.*)(^[^\n]*-+\n.*EASYPY JOB START.*?-+\|\n)", - re.MULTILINE | re.DOTALL, - 1, -) -parse_test_output = regex( - r"(.*)(^[^\n]*-+\n.*EASYPY JOB END.*?-+\|)", - re.MULTILINE | re.DOTALL, - 1, -) -parse_post_test = regex( - r"(.*)(\+-+\+\n\| +Easypy Report +\|\n\+-+\+)", - re.MULTILINE | re.DOTALL, - 1, -) -parse_report = regex( - r"(.*)(\+-+\+\n\| +Task Result Summary +\|\n\+-+\+)", - re.MULTILINE | re.DOTALL, - 1, -) -parse_summary = regex( - r"(.*)(\+-+\+\n\| +Task Result Details +\|\n\+-+\+)", - re.MULTILINE | re.DOTALL, - 1, -) - -timestamp = regex(r"\d{4}-\d{2}-\d{2}.\d{2}:\d{2}:\d{2}(,\d{3})?") -log_level = regex(r" (.+?) ", 0, 1) -source = regex(r"\[([A-Za-z_]+)\]", 0, 1) -messsage = ps.any_char.until(ps.peek(timestamp) | ps.eof) - - -parse_details = regex(r".*", re.MULTILINE | re.DOTALL) -parse_sections = seq( - header=parse_header, - test_output=parse_test_output, - post_test=parse_post_test, - report=parse_report, - summary=parse_summary, - details=parse_details, -) -parse_lines = seq( - ts=timestamp, - level=log_level, - src=source, - msg=messsage.map(lambda x: "".join(x).strip()), -).many() +LEVEL_UNKNOWN = 255 +NEWLINE = string("\n") +SPACE = string(" ") class Config(dict[str, str]): - def __init__(self): - self.dict = dotenv_values(".env") + class LevelAction(argparse.Action): + def __call__(self, parser, namespace, values, option_string=None) -> None: + _ = (parser, option_string) + values = cast(str, values) + setattr(namespace, self.dest, format_debug_level(values.upper())[0]) + + def __init__(self) -> None: + self.dict = {k: v for k, v in dotenv_values(".env").items() if v is not None} + parser = argparse.ArgumentParser() + if not ("LOGFILE" in self.dict or "LOGFILE" in os.environ): + parser.add_argument( + "LOGFILE", + type=Path, + ) + parser.add_argument("-q", "--query", dest="QUERY") + parser.add_argument( + "-l", + "--level", + dest="LOGLEVEL", + type=str, + choices=["debug", "info", "warn", "err", "crit", "unk"], + default=logging.INFO, + action=self.LevelAction, + ) + + self.dict.update(parser.parse_args().__dict__) def __getitem__(self, key: str, /) -> str: try: - return self.dict.__getitem__(key) + return self.dict[key] except KeyError: return os.environ[key] - def __setitem__(self, _key: str, _value: str, /): - raise NotImplementedError("Config is read-only") + def __setitem__(self, _key: str, _value: str, /) -> None: + msg = "Config is read-only" + raise NotImplementedError(msg) - def __delitem__(self, _: str, /): - raise NotImplementedError("Config is read-only") + def __delitem__(self, _: str, /) -> None: + msg = "Config is read-only" + raise NotImplementedError(msg) def __iter__(self): return iter(self.dict) - def __len__(self): + def __len__(self) -> int: return len(self.dict) - def __repr__(self): + def __repr__(self) -> str: return self.dict.__repr__() +""" +Formatting functions +""" + + +def colorize_status(status: str) -> str: + if "PASS" in status: + return Fore.GREEN + status + Style.RESET_ALL + elif "FAIL" in status: + return Fore.RED + status + Style.RESET_ALL + elif "ERROR" in status: + return Fore.BLACK + Back.RED + status + Style.RESET_ALL + return status + + +def format_debug_level(level: str) -> tuple[int, str]: + if level == "DEBUG": + return (logging.DEBUG, level) + elif level == "INFO": + return (logging.INFO, level) + elif "WARN" in level: + return (logging.WARNING, level) + elif "ERR" in level: + return (logging.ERROR, level) + elif "CRIT" in level: + return (logging.CRITICAL, level) + else: + return (LEVEL_UNKNOWN, level) + + +def colorize_by_level(level: str, text: str) -> str: + if level == "DEBUG": + return Style.DIM + text + Style.RESET_ALL + elif level == "INFO": + return Style.BRIGHT + text + Style.RESET_ALL + elif level == "WARNING": + return Fore.YELLOW + text + Style.RESET_ALL + elif level == "ERROR": + return Fore.RED + text + Style.RESET_ALL + elif level == "CRITICAL": + return Style.BRIGHT + Back.RED + Fore.BLACK + text + Style.RESET_ALL + else: + return Fore.BLUE + text + Style.RESET_ALL + # return level + + +""" +Parser defs +""" + + +def block_parser(text: str) -> Parser: + return regex(rf"^.*\n^.*{text}.*$\n.*", re.MULTILINE) + + +timestamp: Parser = regex(r"\d{4}-\d{2}-\d{2}.\d{2}:\d{2}:\d{2}(,\d{3})?") +debug_level: Parser = regex(r"\S+") + + +start_block = block_parser("EASYPY JOB START") +end_block = block_parser("EASYPY JOB END") +report_block = block_parser("Easypy Report") +summary_block = block_parser("Task Result Summary") +details_block = block_parser("Task Result Details") + + +sections = seq( + header=ps.any_char.until(start_block).concat() << start_block << NEWLINE, + test_output=ps.any_char.until(end_block).concat() << end_block << NEWLINE, + post_test=ps.any_char.until(report_block).concat() << report_block, + report=ps.any_char.until(summary_block).concat() << summary_block, + summary=ps.any_char.until(details_block).concat() << details_block, + details=ps.any_char.until(NEWLINE + ps.letter).concat() << NEWLINE, + etc=ps.any_char.until(ps.eof).concat(), +) +messsage = ps.any_char.until(ps.peek(timestamp) | ps.eof).concat() +test_output = seq( + ts=timestamp << SPACE, + level=debug_level.map(format_debug_level) << SPACE, + src=string("[") >> regex(r"[A-Za-z_0-9]+") << string("] "), + msg=messsage.map(lambda x: x.strip()), +).many() + + +test_details: Parser = seq( + test_name=( + NEWLINE.optional() + << (SPACE * 4) + << (string("|--") | string("`--")) + << ps.whitespace + ) + >> ps.any_char.until(ps.whitespace).concat() + << ps.whitespace, + test_results=(ps.letter.many().concat()).map(colorize_status), +) + +subtask_details: Parser = seq( + subtask_name=(string("`--") << ps.whitespace) + >> ps.any_char.until(ps.whitespace).concat() + << ps.whitespace, + subtask_result=(ps.letter.many().concat()).map(colorize_status), + tests=test_details.many(), +) + +task_details: Parser = seq( + task_number=ps.any_char.until(string(":")).concat() << string(":") << ps.whitespace, + task_name=ps.any_char.until(ps.whitespace).concat() << ps.whitespace, + subtasks=subtask_details.many(), +) +details: Parser = task_details.many() + + +""" +Output functions +""" + + +def output_details(log_details: list) -> None: + for task in log_details: + print( + f"{task['task_number']}\t {Style.BRIGHT + task['task_name'] + Style.RESET_ALL}" + ) + for subtask in task["subtasks"]: + print(f" {subtask['subtask_result']}\t{subtask['subtask_name']}") + for test in subtask["tests"]: + print(f" | {test['test_results']}\t{test['test_name']}") + + +def output_tests( + tests: list, + filter_level: int = logging.NOTSET, + search_term: str | re.Pattern | None = None, +) -> None: + for test in tests: + level = test["level"] + if (search_term and not re.search(search_term, test["msg"])) or level[ + 0 + ] >= filter_level: + print( + colorize_by_level( + level[1], f"{test['ts']} {level[1]} {test['src']} {test['msg']}" + ) + ) + + if __name__ == "__main__": config = Config() with Path(config["LOGFILE"]).open() as f: log = f.read() - sections = parse_sections.parse(log) - lines = parse_lines.parse(sections["test_output"]) - ic(lines) + parsed_sections = sections.parse(log) + output_details(details.parse(parsed_sections["details"])) + print("-------------\v") + output_tests(test_output.parse(parsed_sections["test_output"]), config["LOGLEVEL"]) # noqa