107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
import json
|
|
import logging
|
|
import logging.handlers
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
|
|
DEFAULT_LOG_DIR = Path("data") / "logs"
|
|
MODULE_LOGS_ENABLED = False
|
|
LOG_JSON = False
|
|
|
|
|
|
class JSONFormatter(logging.Formatter):
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
payload = {
|
|
"time": self.formatTime(record, "%Y-%m-%d %H:%M:%S"),
|
|
"level": record.levelname,
|
|
"name": record.name,
|
|
"message": record.getMessage(),
|
|
}
|
|
if record.exc_info:
|
|
payload["exc_info"] = self.formatException(record.exc_info)
|
|
return json.dumps(payload, ensure_ascii=True)
|
|
|
|
|
|
def setup_logging(
|
|
level: str = "INFO",
|
|
log_dir: Optional[Path] = None,
|
|
json_logs: bool = False,
|
|
module_logs: bool = False,
|
|
remote_url: str = "",
|
|
) -> None:
|
|
log_dir = log_dir or DEFAULT_LOG_DIR
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
global MODULE_LOGS_ENABLED, LOG_JSON
|
|
MODULE_LOGS_ENABLED = module_logs
|
|
LOG_JSON = json_logs
|
|
|
|
logger = logging.getLogger()
|
|
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
|
|
|
|
formatter: logging.Formatter
|
|
if json_logs:
|
|
formatter = JSONFormatter()
|
|
else:
|
|
formatter = logging.Formatter(
|
|
fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
|
|
file_handler = logging.handlers.RotatingFileHandler(
|
|
log_dir / "overub.log",
|
|
maxBytes=2_000_000,
|
|
backupCount=5,
|
|
encoding="utf-8",
|
|
)
|
|
file_handler.setFormatter(formatter)
|
|
|
|
stream_handler = logging.StreamHandler()
|
|
stream_handler.setFormatter(formatter)
|
|
|
|
if not logger.handlers:
|
|
logger.addHandler(file_handler)
|
|
logger.addHandler(stream_handler)
|
|
if remote_url:
|
|
try:
|
|
from urllib.parse import urlparse
|
|
|
|
parsed = urlparse(remote_url)
|
|
if parsed.scheme in {"http", "https"} and parsed.netloc:
|
|
http_handler = logging.handlers.HTTPHandler(
|
|
host=parsed.netloc,
|
|
url=parsed.path or "/",
|
|
method="POST",
|
|
)
|
|
http_handler.setFormatter(formatter)
|
|
logger.addHandler(http_handler)
|
|
except Exception:
|
|
logger.debug("Remote logging not configured")
|
|
|
|
|
|
def get_logger(name: str) -> logging.Logger:
|
|
logger = logging.getLogger(name)
|
|
if MODULE_LOGS_ENABLED:
|
|
log_dir = DEFAULT_LOG_DIR / "modules"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
handler_name = f"module_file_{name}"
|
|
if not any(getattr(h, "name", "") == handler_name for h in logger.handlers):
|
|
handler = logging.handlers.RotatingFileHandler(
|
|
log_dir / f"{name.replace('.', '_')}.log",
|
|
maxBytes=1_000_000,
|
|
backupCount=3,
|
|
encoding="utf-8",
|
|
)
|
|
handler.name = handler_name
|
|
if LOG_JSON:
|
|
handler.setFormatter(JSONFormatter())
|
|
else:
|
|
handler.setFormatter(
|
|
logging.Formatter(
|
|
fmt="%(asctime)s %(levelname)s %(name)s %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
)
|
|
logger.addHandler(handler)
|
|
return logger
|