diff --git a/logger/modules/log_file_merger.py b/logger/modules/log_file_merger.py new file mode 100644 index 0000000..59e5f96 --- /dev/null +++ b/logger/modules/log_file_merger.py @@ -0,0 +1,75 @@ +""" +Merges log files. +""" + +import os +import pathlib +from datetime import datetime + +from ..read_yaml.modules import read_yaml + +CONFIG_FILE_PATH = pathlib.Path(os.path.dirname(__file__), "config_logger.yaml") +MERGED_LOGS_FILENAME = "merged_logs" + + +def merge_log_files(log_file_directory: str) -> None: + """ + Reads, sorts, and writes log files in the specified directory. + """ + # Read log files + log_files = [ + file + for file in os.listdir(log_file_directory) + if file.endswith(".log") and file != f"{MERGED_LOGS_FILENAME}.log" + ] + log_entries = [] + for log_file in log_files: + with open(os.path.join(log_file_directory, log_file), "r", encoding="utf-8") as file: + log_entries.extend(file.readlines()) + + # Sort log entries + log_entries.sort(key=lambda entry: datetime.strptime(entry.split(": ")[0], "%H:%M:%S")) + + # Write merged logs + merged_log_file = os.path.join(log_file_directory, f"{MERGED_LOGS_FILENAME}.log") + with open(merged_log_file, "w", encoding="utf-8") as file: + file.writelines(log_entries) + + +def get_current_run_directory() -> str: + """ + Gets directory of current run. + """ + # Configuration settings + result, config = read_yaml.open_config(CONFIG_FILE_PATH) + if not result: + print("ERROR: Failed to load configuration file") + + try: + log_directory_path = config["logger"]["directory_path"] + file_datetime_format = config["logger"]["file_datetime_format"] + except KeyError as exception: + print(f"Config key(s) not found: {exception}") + + # Get the path to the logs directory + entries = os.listdir(log_directory_path) + + if len(entries) == 0: + print("ERROR: The directory for this log session was not found.") + + log_names = [ + entry for entry in entries if os.path.isdir(os.path.join(log_directory_path, entry)) + ] + + # Find the log directory for the current run, which is the most recent timestamp + current_directory = max( + log_names, + key=lambda datetime_string: datetime.strptime(datetime_string, file_datetime_format), + ) + + return os.path.join(log_directory_path, current_directory) + + +if __name__ == "__main__": + current_run_directory = get_current_run_directory() + merge_log_files(current_run_directory) diff --git a/logger/test_log_file_merger.py b/logger/test_log_file_merger.py new file mode 100644 index 0000000..8767f88 --- /dev/null +++ b/logger/test_log_file_merger.py @@ -0,0 +1,84 @@ +""" +Log file merger unit tests. +""" + +import os +import tempfile +import shutil +import pytest + + +from .modules import log_file_merger + + +@pytest.fixture(name="dummy_logs") +def fixture_dummy_logs() -> str: # type: ignore + """ + Creates a temporary directory with dummy log files. + """ + temp_directory = tempfile.mkdtemp() + subdirectory = os.path.join(temp_directory, "subdirectory") + os.makedirs(subdirectory) + + # Create three dummy log files in the subdirectory + log_file_1 = os.path.join(subdirectory, "log1.log") + log_file_2 = os.path.join(subdirectory, "log2.log") + log_file_3 = os.path.join(subdirectory, "log3.log") + + with open(log_file_1, "w", encoding="utf-8") as f: + f.write("12:59:28: [INFO] [foo1.py | foo1 | 43] Foo1 initialized\n") + f.write("13:00:04: [ERROR] [foo1.py | foo1 | 30] Foo1 could not be created\n") + f.write("13:00:22: [ERROR] [foo1.py | foo1 | 49] Foo1 failed to create class object\n") + + with open(log_file_2, "w", encoding="utf-8") as f: + f.write("12:59:40: [INFO] [foo2.py | foo2 | 43] Foo2 initialized\n") + f.write("13:00:06: [ERROR] [foo2.py | foo2 | 30] Foo2 could not be created\n") + f.write("13:00:09: [ERROR] [foo2.py | foo2 | 49] Foo2 failed to create class object\n") + + with open(log_file_3, "w", encoding="utf-8") as f: + f.write("12:59:59: [INFO] [foo3.py | foo3 | 43] Foo3 initialized\n") + f.write("13:00:12: [ERROR] [foo3.py | foo3 | 30] Foo3 could not be created\n") + f.write("13:00:30: [ERROR] [foo3.py | foo3 | 49] Foo3 failed to create class object\n") + + yield temp_directory + + # Cleanup + shutil.rmtree(temp_directory) + + +class TestLogFileMerger: + """ + Test suite for the log file merger. + """ + + def test_merge_log_files(self, dummy_logs: str) -> None: + """ + Test if merger correctly combines log files. + """ + temp_directory = dummy_logs + subdirectory = os.path.join(temp_directory, "subdirectory") + + # Merge log files in the subdirectory + log_file_merger.merge_log_files(subdirectory) + + # Check if merged_logs.log is created + merged_log_file = os.path.join(subdirectory, "merged_logs.log") + assert os.path.exists(merged_log_file) + + # Check the contents of the merged log file + with open(merged_log_file, "r", encoding="utf-8") as f: + merged_logs = f.readlines() + + expected_logs = [ + "12:59:28: [INFO] [foo1.py | foo1 | 43] Foo1 initialized\n", + "12:59:40: [INFO] [foo2.py | foo2 | 43] Foo2 initialized\n", + "12:59:59: [INFO] [foo3.py | foo3 | 43] Foo3 initialized\n", + "13:00:04: [ERROR] [foo1.py | foo1 | 30] Foo1 could not be created\n", + "13:00:06: [ERROR] [foo2.py | foo2 | 30] Foo2 could not be created\n", + "13:00:09: [ERROR] [foo2.py | foo2 | 49] Foo2 failed to create class object\n", + "13:00:12: [ERROR] [foo3.py | foo3 | 30] Foo3 could not be created\n", + "13:00:22: [ERROR] [foo1.py | foo1 | 49] Foo1 failed to create class object\n", + "13:00:30: [ERROR] [foo3.py | foo3 | 49] Foo3 failed to create class object\n", + ] + + assert merged_logs == expected_logs