diff --git a/ise/ise_logparser/src/main.py b/ise/ise_logparser/src/main.py index 9052ce3..5516d3f 100755 --- a/ise/ise_logparser/src/main.py +++ b/ise/ise_logparser/src/main.py @@ -10,8 +10,7 @@ from typing import cast import parsy as ps from colorama import Back, Fore, Style -from dotenv import dotenv_values -from parsy import Parser, regex, seq, string +from parsy import Parser, generate, peek, regex, seq, string LEVEL_UNKNOWN = 255 NEWLINE = string("\n") @@ -23,10 +22,16 @@ class Config(dict[str, str]): 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]) + setattr(namespace, self.dest, get_debug_level_const(values.upper())[0]) def __init__(self) -> None: - self.dict = {k: v for k, v in dotenv_values(".env").items() if v is not None} + self.dict = {} + if os.path.exists(".env"): + from dotenv import dotenv_values + + 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( @@ -85,10 +90,10 @@ def colorize_status(status: str) -> str: return status -def format_debug_level(level: str) -> tuple[int, str]: - if level == "DEBUG": +def get_debug_level_const(level: str) -> tuple[int, str]: + if "DEBUG" in level: return (logging.DEBUG, level) - elif level == "INFO": + elif "INFO" in level: return (logging.INFO, level) elif "WARN" in level: return (logging.WARNING, level) @@ -101,21 +106,51 @@ def format_debug_level(level: str) -> tuple[int, str]: def colorize_by_level(level: str, text: str) -> str: - if level == "DEBUG": + if "DEBUG" in level: return Style.DIM + text + Style.RESET_ALL - elif level == "INFO": + elif "INFO" in level: return Style.BRIGHT + text + Style.RESET_ALL - elif level == "WARNING": + elif "WARN" in level: return Fore.YELLOW + text + Style.RESET_ALL - elif level == "ERROR": + elif "ERR" in level: return Fore.RED + text + Style.RESET_ALL - elif level == "CRITICAL": + elif "CRIT" in level: return Style.BRIGHT + Back.RED + Fore.BLACK + text + Style.RESET_ALL else: return Fore.BLUE + text + Style.RESET_ALL # return level +""" +Output functions +""" + + +class TestPrinterBuilder: + loglevel: int = logging.NOTSET + query: str | re.Pattern[str] | None = None + + def __call__( + self, ts: str, level: tuple[int, str], src: str, msg: str + ) -> str | None: + level_name = level[1] + if (level[0] >= self.loglevel) or ( + self.query is not None and re.search(self.query, msg) + ): + res = f"{ts}\t{Fore.BLACK + Back.WHITE + level[1] + Style.RESET_ALL}\t{colorize_by_level(level_name,f'{src}: {msg}')}" + + print(res) + return res + + def with_loglevel(self, loglevel: int) -> TestPrinterBuilder: + self.loglevel = loglevel + return self + + def with_query(self, query: str | re.Pattern) -> TestPrinterBuilder: + self.query = query + return self + + """ Parser defs """ @@ -125,8 +160,13 @@ 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+") +timestamp: Parser = ( + regex( + r"^\d{4}-\d{2}-\d{2}.\d{2}:\d{2}:\d{2}(\+\d{2}:\d{2})?(,\d{3})?", re.MULTILINE + ) + << string(":").optional() +) +debug_level: Parser = regex(r"[A-Za-z0-9_%-]+") << string(":").optional() start_block = block_parser("EASYPY JOB START") @@ -136,90 +176,108 @@ 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() +message = ps.any_char.until(ps.peek(timestamp) | ps.eof).concat() -test_details: Parser = seq( - test_name=( - NEWLINE.optional() - << (SPACE * 4) - << (string("|--") | string("`--")) +@generate +def test_output_line(): + ts = yield timestamp << SPACE + level = yield debug_level << SPACE + src = yield (string("[") >> regex(r"[A-Za-z_0-9]+") << string("] ")).optional() + msg = yield message.map(lambda x: x.rstrip()) + + return print_test_line(ts, get_debug_level_const(level), src, msg) + + # return {"ts": ts, "level": level, "src": src, "msg": msg} + + +@generate +def test_details(): + test_name = yield ( + ( + NEWLINE.optional() + << (SPACE * 4) + << (string("|--") | string("`--")) + << ps.whitespace + ) + >> ps.any_char.until(ps.whitespace).concat() << 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() + test_results = yield (ps.letter.many().concat()) + print(f" | {colorize_status(test_results)}\t{test_name}") + return {"test_results": test_results, "test_name": test_name} -""" -Output functions -""" +@generate +def subtask_details(): + subtask_name = ( + yield (string("`--") << ps.whitespace) + >> ps.any_char.until(ps.whitespace).concat() + << ps.whitespace + ) + subtask_result = yield (ps.letter.many().concat()) + print(f" > {colorize_status(subtask_result)}\t{subtask_name}") + tests = yield test_details.many() + return { + "subtask_result": subtask_result, + "subtask_name": subtask_name, + "tests": tests, + } -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']}") +@generate +def task_details(): + task_number = ( + yield ps.any_char.until(string(":")).concat() << string(":") << ps.whitespace + ) + task_name = yield (ps.any_char.until(ps.whitespace).concat() << ps.whitespace) + print(f"{task_number}\t {Style.BRIGHT + task_name + Style.RESET_ALL}") + + subtasks = yield (subtask_details.many()) + return {"task_number": task_number, "task_name": task_name, "subtasks": subtasks} -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']}" - ) - ) +@generate +# NOTE: Uncomment the print statements if you actually care what's in these sections +def sections(): + header = yield ( + ps.any_char.until(peek(timestamp)).concat() # TODO: output this + + test_output_line.until(start_block).concat() + << start_block + << NEWLINE + ) + + test_output = yield test_output_line.until(end_block) << end_block << NEWLINE + post_test = yield (ps.any_char.until(report_block).concat() << report_block) + # print(post_test) + report = yield (ps.any_char.until(summary_block).concat() << summary_block) + # print(report) + summary = yield (ps.any_char.until(details_block).concat() << details_block) + # print(summary) + details = yield task_details.many() + etc = yield (ps.any_char.until(ps.eof).concat()) + return { + "header": header, + "test_output": test_output, + "post_test": post_test, + "report": report, + "summary": summary, + "details": details, + "etc": etc, + } +# TODO: make this work with streams if __name__ == "__main__": config = Config() with Path(config["LOGFILE"]).open() as f: log = f.read() + global output_test_line + print_test_line = TestPrinterBuilder() + if loglevel := config.get("LOGLEVEL"): + print_test_line = print_test_line.with_loglevel(cast(int, loglevel)) + if query := config.get("QUERY"): + print_test_line = print_test_line.with_query(query) + from icecream import ic + 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