103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
|
import os
|
||
|
|
import time
|
||
|
|
import logging
|
||
|
|
import functools
|
||
|
|
from logging.handlers import RotatingFileHandler
|
||
|
|
from contextlib import contextmanager
|
||
|
|
|
||
|
|
_logger_initialized = False
|
||
|
|
|
||
|
|
LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "logs")
|
||
|
|
LOG_FILE = os.path.join(LOG_DIR, "app.log")
|
||
|
|
|
||
|
|
SLOW_THRESHOLD = 1.0
|
||
|
|
CRITICAL_THRESHOLD = 5.0
|
||
|
|
|
||
|
|
LOG_FORMAT = "[%(asctime)s] [%(levelname)-8s] [%(name)s:%(funcName)s:%(lineno)d] %(message)s"
|
||
|
|
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
|
||
|
|
|
||
|
|
|
||
|
|
def setup_logging(level=logging.DEBUG):
|
||
|
|
global _logger_initialized
|
||
|
|
if _logger_initialized:
|
||
|
|
return
|
||
|
|
_logger_initialized = True
|
||
|
|
|
||
|
|
os.makedirs(LOG_DIR, exist_ok=True)
|
||
|
|
|
||
|
|
root_logger = logging.getLogger("app")
|
||
|
|
root_logger.setLevel(level)
|
||
|
|
|
||
|
|
if root_logger.handlers:
|
||
|
|
return
|
||
|
|
|
||
|
|
formatter = logging.Formatter(LOG_FORMAT, datefmt=DATE_FORMAT)
|
||
|
|
|
||
|
|
file_handler = RotatingFileHandler(
|
||
|
|
LOG_FILE, maxBytes=5 * 1024 * 1024, backupCount=3, encoding="utf-8"
|
||
|
|
)
|
||
|
|
file_handler.setLevel(logging.DEBUG)
|
||
|
|
file_handler.setFormatter(formatter)
|
||
|
|
|
||
|
|
console_handler = logging.StreamHandler()
|
||
|
|
console_handler.setLevel(logging.INFO)
|
||
|
|
console_handler.setFormatter(formatter)
|
||
|
|
|
||
|
|
root_logger.addHandler(file_handler)
|
||
|
|
root_logger.addHandler(console_handler)
|
||
|
|
|
||
|
|
|
||
|
|
def get_logger(name):
|
||
|
|
return logging.getLogger(f"app.{name}")
|
||
|
|
|
||
|
|
|
||
|
|
def log_timing(func=None, *, logger_name=None):
|
||
|
|
def decorator(fn):
|
||
|
|
_logger = get_logger(logger_name or fn.__module__)
|
||
|
|
|
||
|
|
@functools.wraps(fn)
|
||
|
|
def wrapper(*args, **kwargs):
|
||
|
|
fn_name = fn.__qualname__
|
||
|
|
_logger.debug(f"[ENTER] {fn_name}")
|
||
|
|
start = time.perf_counter()
|
||
|
|
try:
|
||
|
|
result = fn(*args, **kwargs)
|
||
|
|
return result
|
||
|
|
except Exception as e:
|
||
|
|
_logger.error(f"[ERROR] {fn_name}: {e}", exc_info=True)
|
||
|
|
raise
|
||
|
|
finally:
|
||
|
|
elapsed = time.perf_counter() - start
|
||
|
|
if elapsed >= CRITICAL_THRESHOLD:
|
||
|
|
_logger.critical(f"[SLOW!] {fn_name} took {elapsed:.3f}s (>={CRITICAL_THRESHOLD}s)")
|
||
|
|
elif elapsed >= SLOW_THRESHOLD:
|
||
|
|
_logger.warning(f"[SLOW] {fn_name} took {elapsed:.3f}s (>={SLOW_THRESHOLD}s)")
|
||
|
|
else:
|
||
|
|
_logger.debug(f"[EXIT] {fn_name} took {elapsed:.3f}s")
|
||
|
|
|
||
|
|
return wrapper
|
||
|
|
|
||
|
|
if func is not None:
|
||
|
|
return decorator(func)
|
||
|
|
return decorator
|
||
|
|
|
||
|
|
|
||
|
|
@contextmanager
|
||
|
|
def log_timing_context(operation, logger_name=None):
|
||
|
|
_logger = get_logger(logger_name or "timing")
|
||
|
|
_logger.debug(f"[START] {operation}")
|
||
|
|
start = time.perf_counter()
|
||
|
|
try:
|
||
|
|
yield
|
||
|
|
except Exception as e:
|
||
|
|
_logger.error(f"[ERROR] {operation}: {e}", exc_info=True)
|
||
|
|
raise
|
||
|
|
finally:
|
||
|
|
elapsed = time.perf_counter() - start
|
||
|
|
if elapsed >= CRITICAL_THRESHOLD:
|
||
|
|
_logger.critical(f"[SLOW!] {operation} took {elapsed:.3f}s (>={CRITICAL_THRESHOLD}s)")
|
||
|
|
elif elapsed >= SLOW_THRESHOLD:
|
||
|
|
_logger.warning(f"[SLOW] {operation} took {elapsed:.3f}s (>={SLOW_THRESHOLD}s)")
|
||
|
|
else:
|
||
|
|
_logger.debug(f"[DONE] {operation} took {elapsed:.3f}s")
|