81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
import logging
|
|
import os
|
|
from collections import deque
|
|
from datetime import datetime, timedelta, timezone
|
|
from logging.handlers import TimedRotatingFileHandler
|
|
from typing import Any
|
|
|
|
|
|
def _get_path(cfg: dict[str, Any]) -> str:
|
|
return cfg.get("incidents", {}).get("path", "/var/server-bot/incidents.log")
|
|
|
|
|
|
def incidents_path(cfg: dict[str, Any]) -> str:
|
|
return _get_path(cfg)
|
|
|
|
|
|
def _get_logger(cfg: dict[str, Any]) -> logging.Logger:
|
|
logger = logging.getLogger("incidents")
|
|
if logger.handlers:
|
|
return logger
|
|
|
|
path = _get_path(cfg)
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
|
|
rotate_when = cfg.get("incidents", {}).get("rotate_when", "W0")
|
|
backup_count = int(cfg.get("incidents", {}).get("backup_count", 8))
|
|
handler = TimedRotatingFileHandler(
|
|
path,
|
|
when=rotate_when,
|
|
interval=1,
|
|
backupCount=backup_count,
|
|
encoding="utf-8",
|
|
utc=True,
|
|
)
|
|
formatter = logging.Formatter(
|
|
"%(asctime)s\t%(message)s",
|
|
datefmt="%Y-%m-%dT%H:%M:%SZ",
|
|
)
|
|
handler.setFormatter(formatter)
|
|
|
|
logger.setLevel(logging.INFO)
|
|
logger.addHandler(handler)
|
|
logger.propagate = False
|
|
return logger
|
|
|
|
|
|
def log_incident(cfg: dict[str, Any], text: str, category: str | None = None) -> None:
|
|
if not cfg.get("incidents", {}).get("enabled", True):
|
|
return
|
|
if category and "category=" not in text:
|
|
text = f"category={category} {text}"
|
|
logger = _get_logger(cfg)
|
|
logger.info(text)
|
|
|
|
|
|
def _parse_line(line: str) -> tuple[datetime | None, str]:
|
|
if "\t" not in line:
|
|
return None, line.strip()
|
|
ts, msg = line.split("\t", 1)
|
|
try:
|
|
dt = datetime.strptime(ts.strip(), "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
dt = None
|
|
return dt, msg.strip()
|
|
|
|
|
|
def read_recent(cfg: dict[str, Any], hours: int, limit: int = 200) -> list[str]:
|
|
path = _get_path(cfg)
|
|
if not os.path.exists(path):
|
|
return []
|
|
|
|
since = datetime.now(timezone.utc) - timedelta(hours=hours)
|
|
lines = deque(maxlen=limit)
|
|
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
|
for line in f:
|
|
dt, msg = _parse_line(line.rstrip())
|
|
if dt is None or dt < since:
|
|
continue
|
|
lines.append(f"{dt:%Y-%m-%d %H:%M} {msg}")
|
|
return list(lines)
|