139 lines
4.2 KiB
Python
139 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
# Standard library imports
|
|
from collections import defaultdict
|
|
from functools import wraps
|
|
import logging
|
|
import logging.config
|
|
import logging.handlers
|
|
import os
|
|
import sys
|
|
import traceback
|
|
import yaml
|
|
|
|
# PyQt imports
|
|
from PyQt6.QtWidgets import QApplication, QMessageBox
|
|
|
|
# Third party imports
|
|
import stackprinter # type: ignore
|
|
|
|
# App imports
|
|
from config import Config
|
|
from classes import ApplicationError
|
|
|
|
|
|
class FunctionFilter(logging.Filter):
|
|
"""Filter to allow category-based logging to stderr."""
|
|
|
|
def __init__(self, module_functions: dict[str, list[str]]):
|
|
super().__init__()
|
|
|
|
self.modules: list[str] = []
|
|
self.functions: defaultdict[str, list[str]] = defaultdict(list)
|
|
|
|
if module_functions:
|
|
for module in module_functions.keys():
|
|
if module_functions[module]:
|
|
for function in module_functions[module]:
|
|
self.functions[module].append(function)
|
|
else:
|
|
self.modules.append(module)
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
if not getattr(record, "levelname", None) == "DEBUG":
|
|
# Only prcess DEBUG messages
|
|
return False
|
|
|
|
module = getattr(record, "module", None)
|
|
if not module:
|
|
# No module in record
|
|
return False
|
|
|
|
# Process if this is a module we're tracking
|
|
if module in self.modules:
|
|
return True
|
|
|
|
# Process if this is a function we're tracking
|
|
if getattr(record, "funcName", None) in self.functions[module]:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
class LevelTagFilter(logging.Filter):
|
|
"""Add leveltag"""
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
# Extract the first character of the level name
|
|
record.leveltag = record.levelname[0]
|
|
# We never actually filter messages out, just add an extra field
|
|
# to the LogRecord
|
|
return True
|
|
|
|
|
|
# Load YAML logging configuration
|
|
with open("app/logging.yaml", "r") as f:
|
|
config = yaml.safe_load(f)
|
|
logging.config.dictConfig(config)
|
|
|
|
# Get logger
|
|
log = logging.getLogger(Config.LOG_NAME)
|
|
|
|
|
|
def handle_exception(exc_type, exc_value, exc_traceback):
|
|
error = str(exc_value)
|
|
if issubclass(exc_type, ApplicationError):
|
|
log.error(error)
|
|
else:
|
|
# Handle unexpected errors (log and display)
|
|
error_msg = "".join(traceback.format_exception(exc_type, exc_value, exc_traceback))
|
|
|
|
print(stackprinter.format(exc_value, suppressed_paths=['/.venv'], style='darkbg'))
|
|
|
|
msg = stackprinter.format(exc_value)
|
|
log.error(msg)
|
|
log.error(error_msg)
|
|
print("Critical error:", error_msg) # Consider logging instead of print
|
|
|
|
if os.environ["MM_ENV"] == "PRODUCTION":
|
|
from helpers import send_mail
|
|
|
|
send_mail(
|
|
Config.ERRORS_TO,
|
|
Config.ERRORS_FROM,
|
|
"Exception (log_uncaught_exceptions) from musicmuster",
|
|
msg,
|
|
)
|
|
if QApplication.instance() is not None:
|
|
fname = os.path.split(exc_traceback.tb_frame.f_code.co_filename)[1]
|
|
msg = f"ApplicationError: {error}\nat {fname}:{exc_traceback.tb_lineno}"
|
|
QMessageBox.critical(None, "Application Error", msg)
|
|
|
|
|
|
def truncate_large(obj, limit=5):
|
|
"""Helper to truncate large lists or other iterables."""
|
|
if isinstance(obj, (list, tuple, set)):
|
|
if len(obj) > limit:
|
|
return f"{type(obj).__name__}(len={len(obj)}, items={list(obj)[:limit]}...)"
|
|
|
|
return repr(obj)
|
|
|
|
|
|
def log_call(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
args_repr = [truncate_large(a) for a in args]
|
|
kwargs_repr = [f"{k}={truncate_large(v)}" for k, v in kwargs.items()]
|
|
params_repr = ", ".join(args_repr + kwargs_repr)
|
|
log.debug(f"call {func.__name__}({params_repr})")
|
|
try:
|
|
result = func(*args, **kwargs)
|
|
log.debug(f"return {func.__name__}: {truncate_large(result)}")
|
|
return result
|
|
except Exception as e:
|
|
log.debug(f"exception in {func.__name__}: {e}")
|
|
raise
|
|
return wrapper
|
|
|
|
|
|
sys.excepthook = handle_exception
|