refact: improve logger handling

This commit is contained in:
sebthom
2025-02-10 20:34:58 +01:00
parent ec7ffedcd6
commit 367ef07798
4 changed files with 64 additions and 24 deletions

View File

@@ -3,14 +3,13 @@ SPDX-FileCopyrightText: © Sebastian Thomschke and contributors
SPDX-License-Identifier: AGPL-3.0-or-later SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/ SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
""" """
import asyncio, atexit, copy, importlib.metadata, json, logging, os, re, signal, shutil, sys, textwrap, time import asyncio, atexit, copy, importlib.metadata, json, os, re, signal, shutil, sys, textwrap, time
import getopt # pylint: disable=deprecated-module import getopt # pylint: disable=deprecated-module
import urllib.parse as urllib_parse import urllib.parse as urllib_parse
import urllib.request as urllib_request import urllib.request as urllib_request
from collections.abc import Iterable from collections.abc import Iterable
from datetime import datetime from datetime import datetime
from gettext import gettext as _ from gettext import gettext as _
from logging.handlers import RotatingFileHandler
from typing import Any, Final from typing import Any, Final
import certifi, colorama, nodriver import certifi, colorama, nodriver
@@ -28,8 +27,8 @@ from ._version import __version__
# W0406: possibly a bug, see https://github.com/PyCQA/pylint/issues/3933 # W0406: possibly a bug, see https://github.com/PyCQA/pylint/issues/3933
LOG:Final[logging.Logger] = loggers.get_logger(__name__) LOG:Final[loggers.Logger] = loggers.get_logger(__name__)
LOG.setLevel(logging.INFO) LOG.setLevel(loggers.INFO)
colorama.just_fix_windows_console() colorama.just_fix_windows_console()
@@ -51,7 +50,7 @@ class KleinanzeigenBot(WebScrapingMixin):
self.categories:dict[str, str] = {} self.categories:dict[str, str] = {}
self.file_log:logging.FileHandler | None = None self.file_log:loggers.LogFileHandle | None = None
log_file_basename = is_frozen() and os.path.splitext(os.path.basename(sys.executable))[0] or self.__module__ log_file_basename = is_frozen() and os.path.splitext(os.path.basename(sys.executable))[0] or self.__module__
self.log_file_path:str | None = abspath(f"{log_file_basename}.log") self.log_file_path:str | None = abspath(f"{log_file_basename}.log")
@@ -61,9 +60,8 @@ class KleinanzeigenBot(WebScrapingMixin):
def __del__(self) -> None: def __del__(self) -> None:
if self.file_log: if self.file_log:
self.file_log.flush()
loggers.LOG_ROOT.removeHandler(self.file_log)
self.file_log.close() self.file_log.close()
self.file_log = None
self.close_browser_session() self.close_browser_session()
def get_version(self) -> str: def get_version(self) -> str:
@@ -239,8 +237,8 @@ class KleinanzeigenBot(WebScrapingMixin):
case "--lang": case "--lang":
set_current_locale(Locale.of(value)) set_current_locale(Locale.of(value))
case "-v" | "--verbose": case "-v" | "--verbose":
LOG.setLevel(logging.DEBUG) LOG.setLevel(loggers.DEBUG)
logging.getLogger("nodriver").setLevel(logging.INFO) loggers.get_logger("nodriver").setLevel(loggers.INFO)
match len(arguments): match len(arguments):
case 0: case 0:
@@ -258,10 +256,7 @@ class KleinanzeigenBot(WebScrapingMixin):
return return
LOG.info("Logging to [%s]...", self.log_file_path) LOG.info("Logging to [%s]...", self.log_file_path)
self.file_log = RotatingFileHandler(filename = self.log_file_path, maxBytes = 10 * 1024 * 1024, backupCount = 10, encoding = "utf-8") self.file_log = loggers.configure_file_logging(self.log_file_path)
self.file_log.setLevel(logging.DEBUG)
self.file_log.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
loggers.LOG_ROOT.addHandler(self.file_log)
LOG.info("App version: %s", self.get_version()) LOG.info("App version: %s", self.get_version())
LOG.info("Python version: %s", sys.version) LOG.info("Python version: %s", sys.version)
@@ -611,7 +606,7 @@ class KleinanzeigenBot(WebScrapingMixin):
LOG.info("Publishing ad '%s'...", ad_cfg["title"]) LOG.info("Publishing ad '%s'...", ad_cfg["title"])
if LOG.isEnabledFor(logging.DEBUG): if loggers.is_debug(LOG):
LOG.debug(" -> effective ad meta:") LOG.debug(" -> effective ad meta:")
YAML().dump(ad_cfg, sys.stdout) YAML().dump(ad_cfg, sys.stdout)

View File

@@ -5,6 +5,7 @@ SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanze
""" """
import copy, logging, re, sys import copy, logging, re, sys
from gettext import gettext as _ from gettext import gettext as _
from logging.handlers import RotatingFileHandler
from typing import Any, Final # @UnusedImport from typing import Any, Final # @UnusedImport
import colorama import colorama
@@ -12,12 +13,14 @@ from . import i18n, reflect
__all__ = [ __all__ = [
"Logger", "Logger",
"LOG_ROOT", "LogFileHandle",
"DEBUG", "DEBUG",
"INFO", "INFO",
"configure_console_logging", "configure_console_logging",
"configure_file_logging",
"flush_all_handlers", "flush_all_handlers",
"get_logger" "get_logger",
"is_debug"
] ]
Logger = logging.Logger Logger = logging.Logger
@@ -91,6 +94,46 @@ def configure_console_logging() -> None:
LOG_ROOT.addHandler(stderr_log) LOG_ROOT.addHandler(stderr_log)
class LogFileHandle:
"""Encapsulates a log file handler with close and status methods."""
def __init__(self, file_path: str, handler: RotatingFileHandler, logger: logging.Logger):
self.file_path = file_path
self._handler:RotatingFileHandler | None = handler
self._logger = logger
def close(self) -> None:
"""Flushes, removes, and closes the log handler."""
if self._handler:
self._handler.flush()
self._logger.removeHandler(self._handler)
self._handler.close()
self._handler = None
def is_closed(self) -> bool:
"""Returns whether the log handler has been closed."""
return not self._handler
def configure_file_logging(log_file_path:str) -> LogFileHandle:
"""
Sets up a file logger and returns a callable to flush, remove, and close it.
@param log_file_path: Path to the log file.
@return: Callable[[], None]: A function that cleans up the log handler.
"""
fh = RotatingFileHandler(
filename = log_file_path,
maxBytes = 10 * 1024 * 1024, # 10 MB
backupCount = 10,
encoding = "utf-8"
)
fh.setLevel(logging.DEBUG)
fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s"))
LOG_ROOT.addHandler(fh)
return LogFileHandle(log_file_path, fh, LOG_ROOT)
def flush_all_handlers() -> None: def flush_all_handlers() -> None:
for handler in LOG_ROOT.handlers: for handler in LOG_ROOT.handlers:
handler.flush() handler.flush()

View File

@@ -41,10 +41,11 @@ class TestKleinanzeigenBot:
bot.configure_file_logging() bot.configure_file_logging()
file_log = bot.file_log file_log = bot.file_log
assert not file_log.stream.closed # type: ignore[union-attr] assert file_log is not None
assert not file_log.is_closed()
# Delete and garbage collect the bot instance to ensure the destructor (__del__) is called # Delete and garbage collect the bot instance to ensure the destructor (__del__) is called
del bot del bot
gc.collect() gc.collect()
assert file_log.stream is None # type: ignore[union-attr] assert file_log.is_closed()

View File

@@ -3,7 +3,7 @@ SPDX-FileCopyrightText: © Jens Bergmann and contributors
SPDX-License-Identifier: AGPL-3.0-or-later SPDX-License-Identifier: AGPL-3.0-or-later
SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/ SPDX-ArtifactOfProjectHomePage: https://github.com/Second-Hand-Friends/kleinanzeigen-bot/
""" """
import copy, logging, os, tempfile import copy, os, tempfile
from collections.abc import Generator from collections.abc import Generator
from datetime import datetime, timedelta from datetime import datetime, timedelta
from pathlib import Path from pathlib import Path
@@ -16,6 +16,7 @@ from ruamel.yaml import YAML
from kleinanzeigen_bot import LOG, KleinanzeigenBot from kleinanzeigen_bot import LOG, KleinanzeigenBot
from kleinanzeigen_bot._version import __version__ from kleinanzeigen_bot._version import __version__
from kleinanzeigen_bot.ads import calculate_content_hash from kleinanzeigen_bot.ads import calculate_content_hash
from kleinanzeigen_bot.utils import loggers
@pytest.fixture @pytest.fixture
@@ -224,7 +225,7 @@ class TestKleinanzeigenBotCommandLine:
def test_parse_args_handles_verbose_flag(self, test_bot: KleinanzeigenBot) -> None: def test_parse_args_handles_verbose_flag(self, test_bot: KleinanzeigenBot) -> None:
"""Verify that verbose flag sets correct log level.""" """Verify that verbose flag sets correct log level."""
test_bot.parse_args(["dummy", "--verbose"]) test_bot.parse_args(["dummy", "--verbose"])
assert LOG.level == logging.DEBUG assert loggers.is_debug(LOG)
def test_parse_args_handles_config_path(self, test_bot: KleinanzeigenBot, test_data_dir: str) -> None: def test_parse_args_handles_config_path(self, test_bot: KleinanzeigenBot, test_data_dir: str) -> None:
"""Verify that config path is set correctly.""" """Verify that config path is set correctly."""
@@ -432,10 +433,10 @@ class TestKleinanzeigenBotBasics:
def test_get_log_level(self, test_bot: KleinanzeigenBot) -> None: def test_get_log_level(self, test_bot: KleinanzeigenBot) -> None:
"""Test log level configuration.""" """Test log level configuration."""
# Reset log level to default # Reset log level to default
LOG.setLevel(logging.INFO) LOG.setLevel(loggers.INFO)
assert LOG.level == logging.INFO assert not loggers.is_debug(LOG)
test_bot.parse_args(['script.py', '-v']) test_bot.parse_args(['script.py', '-v'])
assert LOG.level == logging.DEBUG assert loggers.is_debug(LOG)
def test_get_config_file_path(self, test_bot: KleinanzeigenBot) -> None: def test_get_config_file_path(self, test_bot: KleinanzeigenBot) -> None:
"""Test config file path handling.""" """Test config file path handling."""
@@ -476,7 +477,7 @@ class TestKleinanzeigenBotArgParsing:
def test_parse_args_verbose(self, test_bot: KleinanzeigenBot) -> None: def test_parse_args_verbose(self, test_bot: KleinanzeigenBot) -> None:
"""Test parsing verbose flag.""" """Test parsing verbose flag."""
test_bot.parse_args(['script.py', '-v', 'help']) test_bot.parse_args(['script.py', '-v', 'help'])
assert logging.getLogger('kleinanzeigen_bot').level == logging.DEBUG assert loggers.is_debug(loggers.get_logger('kleinanzeigen_bot'))
def test_parse_args_config_path(self, test_bot: KleinanzeigenBot) -> None: def test_parse_args_config_path(self, test_bot: KleinanzeigenBot) -> None:
"""Test parsing config path.""" """Test parsing config path."""