syslog v0.2.0 [base]

"""syslog.Module — журнал бота для кабинета (прямой эфир с паузой)."""
import logging
import os

from core.base_module import BaseModule

logger = logging.getLogger(__name__)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        mod_dir = os.path.dirname(os.path.abspath(__file__))
        self.config = ctx.module_config("syslog", mod_dir)
        self.ctx = ctx

    def health(self) -> dict:
        return {"ok": True, "module": self.name}

"""syslog.ui_web — GET /api/syslog/logs: хвост журнала службы/файла."""

ROUTES = [
    ("GET", "/api/syslog/logs", "logs", {"right": "view_logs",
     "desc": "Хвост журнала (?tail=200|500|1000, ?level=all|info|warning|error)"}),
]

SERVICE = [
    {"key": "logs", "icon": "🧾", "title": "Логи бота", "right": "view_logs",
     "phase": 1, "desc": "Прямой эфир журнала службы с паузой"},
]

PANELS = []


def handle_api(ctx, config, method, req):
    from core.errors import UserError

    if method == "logs":
        query = req.get("query") or {}
        try:
            tail = int(query.get("tail") or 200)
        except (TypeError, ValueError):
            tail = 200
        if tail not in (200, 500, 1000):
            tail = 200
        level = (query.get("level") or "all").strip().lower()
        if level not in ("all", "info", "warning", "error"):
            level = "all"
        lines, source = _tail_log(ctx, config, tail, level)
        return {"ok": True, "lines": lines, "source": source,
                "tail": tail, "level": level}

    raise UserError("неизвестный метод: %s" % method)


def _level_rx(level):
    import re as _re
    return {"info": _re.compile(r"\bDEBUG\b"),
            "warning": _re.compile(r"\b(WARNING|ERROR|CRITICAL)\b"),
            "error": _re.compile(r"\b(ERROR|CRITICAL)\b")}.get(level)


def _apply_level(lines, level):
    """Фильтр по маркеру уровня в тексте строки (надёжнее journalctl -p:
    Python пишет всё в stderr с приоритетом info). info = всё кроме DEBUG."""
    import re as _re
    if level == "all":
        return lines
    rx = _level_rx(level)
    if level == "info":
        return [l for l in lines if not rx.search(l)]
    return [l for l in lines if rx.search(l)]


def _tail_log(ctx, config, tail, level="all"):
    """Хвост журнала: journalctl юнита (SYSTEMD_UNIT) либо файл лога.

    Фильтр уровня — по содержимому строк (маркер LEVELNAME), на стороне
    команды. Берём с запасом x5, отдаём до tail.
    """
    import os as _os
    import subprocess as _sp

    unit = ""
    try:
        unit = ((config.get("SYSTEMD_UNIT", "") or "").strip()
                if config else "")
    except Exception:  # noqa: BLE001
        unit = ""
    if unit:
        try:
            cmd = ["journalctl", "-u", unit, "-n", str(tail * 5),
                   "--no-pager", "-q"]
            r = _sp.run(cmd, capture_output=True, text=True, timeout=15)
            if r.returncode == 0:
                lines = (r.stdout or "").splitlines()
                lines = _apply_level(lines, level)[-tail:]
                src = "journalctl -u %s%s" % (
                    unit, (" level=%s" % level) if level != "all" else "")
                return [l[:2000] for l in lines], src
        except Exception:  # noqa: BLE001
            pass
    log_file = ""
    try:
        log_file = _os.path.join(ctx.data_dir, "logs", "bot.log")
    except Exception:  # noqa: BLE001
        log_file = ""
    if log_file and _os.path.isfile(log_file):
        try:
            with open(log_file, "r", encoding="utf-8",
                      errors="replace") as f:
                # запас на фильтрацию: читаем больше, отдаём tail
                raw = f.read().splitlines()[-(tail * 5):]
            raw = _apply_level(raw, level)
            return [l[:2000] for l in raw[-tail:]], log_file
        except Exception:  # noqa: BLE001
            pass
    return [], "нет источника (SYSTEMD_UNIT пуст, файла нет)"