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) -> None: if not cfg.get("incidents", {}).get("enabled", True): return 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)