src/Logger.py - master - Flipper/Marginal

src/Logger.py
527 строк17 KB
Alex
Bug fixes.
12 янв 2026, 13:17
12 янв 2026, 13:17e907759
import os.path
import traceback
import re
from datetime import datetime

from utils import generateFullPath, getBasePath


class Logger:
    """
    Central logging subsystem.

    Handles session logging, error logging, calculation-specific logs,
    and live console output with optional translation support.
    """

    def __init__(self, backbone):
        """
        Initialize logger state and bind translator.

        Args:
            backbone: Application backbone instance.
        """
        self.translator = backbone.translator

        self.logActive = False
        self.errorLogActive = False
        self.calcActive = False

        self.sessionTimestamp = None

        self.logDirLimit = 10
        self.logDir = generateFullPath(getBasePath(), "logs")
        self.logFile = None

        self.errorDir = self.logDir
        self.errorFile = None

        self.currentCalcFile = None

        self.consoleMWWriteMethod = None
        self.connected = None

    def startLog(self):
        """
        Start the main session log if not already active.
        """
        if self.logActive:
            return

        if not os.path.isdir(self.logDir):
            os.mkdir(self.logDir)

        timestamp = datetime.fromtimestamp(
            datetime.timestamp(datetime.now())
        )
        self.sessionTimestamp = timestamp

        filename = "LOGENTRY_" + timestamp.strftime("%d%m%Y")
        logPath = f"{self.logDir}\\{filename}.txt"

        self.logFile = open(logPath, "a")
        self.logActive = True

        self.logFile.write(
            f"{'-' * 50}Session:{timestamp}{'-' * 50}\n"
        )

        self.cullLogs()

    def closeLog(self):
        """
        Close the main session log and error log if active.
        """
        if self.logActive:
            self.log("System", "Session terminated.", True)
            self.logFile.write(f"{'-' * 150}\n")
            self.logFile.flush()
            self.logFile.close()
            self.logFile = None
            self.logActive = False

        if self.errorLogActive:
            self.closeErrorLog()

    def startErrorLog(self):
        """
        Start the error log, ensuring the main log exists first.
        """
        self.startLog()

        if self.errorLogActive:
            return

        timestamp = self.sessionTimestamp
        filename = "ERROR_LOGENTRY_" + timestamp.strftime("%d%m%Y")
        logPath = f"{self.errorDir}\\{filename}.txt"

        self.errorFile = open(logPath, "a")
        self.errorLogActive = True

        self.errorFile.write(
            f"{'=' * 50}Session:{timestamp}{'=' * 50}\n"
        )
        self.errorFile.flush()

    def closeErrorLog(self):
        """
        Close the error log file.
        """
        self.errorFile.write(f"{'=' * 150}")
        self.errorFile.flush()
        self.errorFile.close()
        self.errorFile = None
        self.errorLogActive = False

    def log(self,author="System",message="Ping",hideInConsole: bool = False,*args,**kwargs):
        """
        Write a formatted message to the session log and optionally the console.

        Args:
            author (str): Log entry source.
            message (str): Message template.
            hideInConsole (bool): Suppress console output if True.
        """
        if not hideInConsole:
            translatedMessage = self.translator.translate(message)

        if args or kwargs:
            try:
                if "{" in message and "}" in message:
                    if kwargs:
                        message = message.format(**kwargs)
                        if not hideInConsole:
                            translatedMessage = translatedMessage.format(**kwargs)
                    elif args:
                        if len(args) == 1 and isinstance(args[0], (list, tuple)):
                            message = message.format(*args[0])
                            if not hideInConsole:
                                translatedMessage = translatedMessage.format(*args[0])
                        else:
                            message = message.format(*args)
                            if not hideInConsole:
                                translatedMessage = translatedMessage.format(*args)
            except Exception:
                self.log(
                    "System",
                    "Failed to format message:\n{}",
                    False,
                    message
                )

        if not hideInConsole:
            self.consoleLog(author, translatedMessage)

        if not self.logActive:
            self.startLog()

        timestamp = datetime.now().strftime("%d/%m/%Y|%H:%M:%S")
        self.logFile.write(
            f"[{timestamp}] {author} - {message}\n"
        )
        self.logFile.flush()

    def customLog(self,content: str,translate: bool = True,hideInConsole: bool = False,*args,**kwargs):
        """
        Write a raw log entry without author or timestamp formatting.
        """
        if not hideInConsole:
            translatedContent = self.translator.translate(content)

        if args or kwargs:
            try:
                if "{" in content and "}" in content:
                    if kwargs:
                        content = content.format(**kwargs)
                        if not hideInConsole:
                            translatedContent = translatedContent.format(**kwargs)
                    elif args:
                        if len(args) == 1 and isinstance(args[0], (list, tuple)):
                            content = content.format(*args[0])
                            if not hideInConsole:
                                translatedContent = translatedContent.format(*args[0])
                        else:
                            content = content.format(*args)
                            if not hideInConsole:
                                translatedContent = translatedContent.format(*args)
            except Exception:
                self.log(
                    "System",
                    "Failed to format custom message:\n{}",
                    True,
                    content
                )

        if not hideInConsole:
            self.customConsoleLog(
                translatedContent if translate else content
            )

        if not self.logActive:
            self.startLog()

        self.logFile.write(f"{content}\n")
        self.logFile.flush()

    def consoleLog(self, author="System", message="Ping"):
        """
        Write a formatted log message to the UI console.
        """
        if not self.connected:
            return

        timestamp = datetime.now().strftime("%d/%m/%Y | %H:%M:%S")
        write = f"[{timestamp}] {author} - {message}"
        self.consoleMWWriteMethod(write)

    def customConsoleLog(self, content: str = "Ping"):
        """
        Write a raw message to the UI console.
        """
        if self.connected:
            self.consoleMWWriteMethod(content)

    def logError(self, excType, excValue, excTraceback, context):
        """
        Write an exception traceback to the error log.
        """
        if not self.errorLogActive:
            self.startErrorLog()

        timestamp = datetime.now().strftime("%d/%m/%Y|%H:%M:%S")
        self.errorFile.write(
            f"[{timestamp}] ExceptionManager - Exception in {context}:\n"
        )
        traceback.print_exception(
            excType,
            excValue,
            excTraceback,
            file=self.errorFile
        )
        self.errorFile.write(f"{'~' * 50}\n")
        self.errorFile.flush()

    def cullLogs(self):
        """
        Enforce log file retention limits by deleting oldest logs.
        """
        logFolderContents = os.listdir(self.logDir)

        logFiles = [
            file for file in logFolderContents
            if ("ERROR" not in file)
            and ("LOGENTRY" in file)
            and os.path.isfile(os.path.join(self.logDir, file))
        ]

        filesToDelete = len(logFiles) - self.logDirLimit

        if filesToDelete > 0:
            self.log(
                "Logger",
                f"Log files limit exceeded by {filesToDelete}. "
                "Deleting outdated logs...",
                True
            )

        deletedLogCount = 0
        deletedErrorCount = 0

        while filesToDelete > 0:
            oldestFile = None
            oldestFilePath = None

            for file in logFiles:
                filePath = os.path.join(self.logDir, file)
                if (
                    oldestFilePath is None
                    or os.path.getctime(filePath)
                    < os.path.getctime(oldestFilePath)
                ):
                    oldestFile = file
                    oldestFilePath = filePath

                # Fail guard: never delete active log
                if self.logFile and self.logFile.name == oldestFilePath:
                    self.log(
                        "Logger",
                        "Deletion aborted: fail guard triggered.",
                        True
                    )
                    return

            os.remove(oldestFilePath)
            logFiles.remove(oldestFile)

            deletedLogCount += 1
            filesToDelete -= 1

            relatedErrorFile = os.path.join(
                self.errorDir,
                f"ERROR_{oldestFile}"
            )
            if os.path.exists(relatedErrorFile):
                os.remove(relatedErrorFile)
                deletedErrorCount += 1

        if deletedLogCount > 0:
            if deletedErrorCount > 0:
                self.log(
                    "Logger",
                    f"Successfully deleted {deletedLogCount} log files + "
                    f"{deletedErrorCount} related error files.",
                    True
                )
            else:
                self.log(
                    f"Successfully deleted {deletedLogCount} log files.",
                    hideInConsole=True
                )

    def clearOtherLogs(self):
        """
        Delete all log files except the current active session log and error log.
        This helps clean up storage space while preserving the current session.
        """
        try:
            if not os.path.isdir(self.logDir):
                return

            # Get current log file names if they exist
            currentLogName = None
            currentErrorName = None

            if self.logFile and hasattr(self.logFile, 'name'):
                currentLogName = os.path.basename(self.logFile.name)

            if self.errorFile and hasattr(self.errorFile, 'name'):
                currentErrorName = os.path.basename(self.errorFile.name)

            # Get all files in the log directory
            allFiles = os.listdir(self.logDir)

            deletedLogs = 0
            deletedErrors = 0
            skippedCurrent = 0

            for file in allFiles:
                filePath = os.path.join(self.logDir, file)

                # Only process files, not directories
                if not os.path.isfile(filePath):
                    continue

                # Skip the current session logs
                if file == currentLogName or file == currentErrorName:
                    skippedCurrent += 1
                    continue

                # Identify log files (regular and error)
                isLogFile = "LOGENTRY" in file and "ERROR" not in file
                isErrorFile = "ERROR_LOGENTRY" in file

                if isLogFile or isErrorFile:
                    try:
                        os.remove(filePath)
                        if isLogFile:
                            deletedLogs += 1
                        elif isErrorFile:
                            deletedErrors += 1
                    except OSError as e:
                        self.log(
                            "Logger",
                            f"Failed to delete log file {file}: {e}",
                            True
                        )

            # Log the results
            if deletedLogs > 0 or deletedErrors > 0:
                messageParts = []
                if deletedLogs > 0:
                    messageParts.append(f"{deletedLogs} log file(s)")
                if deletedErrors > 0:
                    messageParts.append(f"{deletedErrors} error file(s)")

                self.log(
                    "Logger",
                    f"Deleted old logs: {', '.join(messageParts)}.\n" +
                    f"Current session logs preserved ({skippedCurrent} file(s) kept).",
                    True
                )
            else:
                self.log(
                    "Logger",
                    "No old logs to delete. Current session logs preserved.",
                    True
                )

        except Exception as e:
            self.log(
                "Logger",
                f"Error during log cleanup: {e}",
                True
            )

    def getSessionTimeStamp(self) -> datetime:
        """
        Return the session timestamp.

        Returns:
            datetime -> Session start time.
        """
        return self.sessionTimestamp

    def startCalcLog(self, filename, fileOutput, profile: dict = None, input: dict = None):
        """
        Start a calculation-specific log file.
        """
        if self.calcActive:
            return

        if not os.path.isdir(fileOutput):
            os.mkdir(fileOutput)

        logPath = f"{fileOutput}\\{filename}_CALC.txt"

        self.currentCalcFile = open(logPath, "w")
        self.calcActive = True

        self.currentCalcFile.write("=" * 60 + "\n")
        self.currentCalcFile.write(
            f"CALCULATION LOG - {self.sessionTimestamp}\n"
        )
        self.currentCalcFile.write("=" * 60 + "\n\n")

        if profile:
            self.currentCalcFile.write(
                f"PROFILE: {profile.get('profileName', 'Unnamed')}\n"
            )
            self.currentCalcFile.write(
                f"DESCRIPTION: {profile.get('description', 'No description')}\n"
            )
            self.currentCalcFile.write(
                f"VERSION: {profile.get('metadata', {}).get('version', 'N/A')}\n"
            )
            self.currentCalcFile.write(
                f"AUTHOR: {profile.get('metadata', {}).get('author', 'Unknown')}\n\n"
            )

        if input:
            self.currentCalcFile.write("INPUT DATA SUMMARY:\n")
            self.currentCalcFile.write("-" * 40 + "\n")

            for key, value in input.items():
                if isinstance(value, list):
                    if len(value) > 5:
                        self.currentCalcFile.write(
                            f"  {key}: "
                            f"[{value[0]}, {value[1]}, ..., {value[-1]}] "
                            f"(len={len(value)})\n"
                        )
                    else:
                        self.currentCalcFile.write(
                            f"  {key}: {value}\n"
                        )
                else:
                    self.currentCalcFile.write(
                        f"  {key}: {value}\n"
                    )

            self.currentCalcFile.write("\n")

        self.currentCalcFile.write("CALCULATION DETAILS:\n")
        self.currentCalcFile.write("=" * 60 + "\n\n")
        self.currentCalcFile.flush()

    def logCalculation(self, entry: str):
        """
        Append a calculation step entry to the calculation log.
        """
        if self.calcActive:
            self.currentCalcFile.write(f"{entry}\n")
            self.currentCalcFile.flush()

    def closeCalcLog(self):
        """
        Finalize and close the calculation log file.
        """
        if not self.calcActive:
            return

        self.currentCalcFile.write("=" * 60 + "\n\n")
        self.currentCalcFile.write("END OF CALCULATIONS.\n")
        self.currentCalcFile.write("=" * 60 + "\n")

        self.currentCalcFile.flush()
        self.currentCalcFile.close()
        self.currentCalcFile = None
        self.calcActive = False

    def generateCalcExpression(self, operation: str, localVars: dict):
        """
        Generate a human-readable expression with variables substituted.

        NOTE: This is a string transformation utility only and does NOT
        evaluate or execute the expression.
        """
        result = operation

        nonCallables = {
            k: v for k, v in localVars.items()
            if not callable(v)
        }

        # Replace longer keys first to avoid partial matches
        for key in sorted(nonCallables.keys(), key=len, reverse=True):
            pattern = r"(?<!\\w)" + re.escape(key) + r"(?!\\w)"
            result = re.sub(pattern, str(localVars[key]), result)

        return result

    def connectConsole(self, method: callable):
        """
        Connect a UI console write method to the logger.
        """
        self.connected = True
        self.consoleMWWriteMethod = method