Created script to merge log files

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)

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