site/WESP_REL/app/services/admin_hardware_log.py

"""Журнал снимков нагрузки железа (JSONL) для графиков в админке."""

from __future__ import annotations

import json
import logging
import os
import threading
import time
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

_LOG = logging.getLogger(__name__)
_LOCK = threading.Lock()
_LAST_SAMPLE_MONO: Optional[float] = None


def _hardware_log_path(app) -> str:
    raw = (app.config.get("WESP_ADMIN_HARDWARE_LOG_PATH") or "").strip()
    return raw


def _line_unix_ts(obj: Dict[str, Any]) -> Optional[float]:
    u = obj.get("u")
    if isinstance(u, (int, float)):
        return float(u)
    return None


def maybe_append_hardware_sample(app, metrics: Dict[str, Any]) -> None:
    """
    Дописывает одну строку JSON в лог, не чаще чем раз в WESP_ADMIN_HARDWARE_LOG_INTERVAL_SEC.
    Удаляет из файла записи старше WESP_ADMIN_HARDWARE_LOG_RETENTION_DAYS (по полю u — unix time).
    Ошибки записи логируются и не пробрасываются (ответ /system-metrics остаётся 200).
    """
    try:
        _maybe_append_hardware_sample_impl(app, metrics)
    except Exception:
        _LOG.exception(
            "Сбой журнала нагрузки железа (WESP_ADMIN_HARDWARE_LOG_PATH). "
            "Метрики дашборда не затронуты."
        )


def _maybe_append_hardware_sample_impl(app, metrics: Dict[str, Any]) -> None:
    global _LAST_SAMPLE_MONO

    path_str = _hardware_log_path(app)
    if not path_str:
        return

    interval = float(app.config.get("WESP_ADMIN_HARDWARE_LOG_INTERVAL_SEC", 60) or 60)
    interval = max(5.0, min(interval, 3600.0))

    now_mono = time.monotonic()
    with _LOCK:
        prev = _LAST_SAMPLE_MONO
        if isinstance(prev, (int, float)) and (now_mono - float(prev)) < interval:
            return
        _LAST_SAMPLE_MONO = float(now_mono)

    path = Path(path_str)
    if not path.is_absolute():
        base = app.config.get("BASE_DIR", ".")
        path = Path(base) / path

    cpu = metrics.get("cpu") if isinstance(metrics.get("cpu"), dict) else {}
    mem = metrics.get("memory") if isinstance(metrics.get("memory"), dict) else {}
    sw = metrics.get("swap") if isinstance(metrics.get("swap"), dict) else {}
    disk = metrics.get("disk") if isinstance(metrics.get("disk"), dict) else {}

    row = {
        "u": int(time.time()),
        "cpu_pct": float(cpu.get("percent") or 0),
        "cpu_cores": int(cpu.get("cores") or 0),
        "ram_pct": float(mem.get("percent") or 0),
        "ram_used": int(mem.get("used") or 0),
        "ram_total": int(mem.get("total") or 0),
        "swap_pct": float(sw.get("percent") or 0) if int(sw.get("total") or 0) else 0.0,
        "swap_used": int(sw.get("used") or 0),
        "swap_total": int(sw.get("total") or 0),
        "disk_pct": float(disk.get("percent") or 0) if disk else None,
        "disk_used": int(disk.get("used") or 0) if disk else None,
        "disk_total": int(disk.get("total") or 0) if disk else None,
    }
    line = json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n"

    path.parent.mkdir(parents=True, exist_ok=True)
    cutoff = time.time() - float(
        max(1, min(int(app.config.get("WESP_ADMIN_HARDWARE_LOG_RETENTION_DAYS", 3) or 3), 30)) * 86400
    )

    _rewrite_prune_and_append(path, cutoff, line)


def _rewrite_prune_and_append(path: Path, cutoff_ts: float, new_line: str) -> None:
    """Читает JSONL, оставляет строки с u >= cutoff_ts, дописывает new_line."""
    lines_kept: List[str] = []
    if path.is_file():
        try:
            with path.open("r", encoding="utf-8") as f:
                for ln in f:
                    ln = ln.strip()
                    if not ln:
                        continue
                    try:
                        obj = json.loads(ln)
                    except json.JSONDecodeError:
                        continue
                    ts = _line_unix_ts(obj)
                    if ts is not None and ts >= cutoff_ts:
                        lines_kept.append(ln + "\n")
        except OSError:
            lines_kept = []

    tmp = path.with_name(f"{path.name}.tmp.{os.getpid()}")
    try:
        with tmp.open("w", encoding="utf-8") as out:
            for ln in lines_kept:
                out.write(ln)
            out.write(new_line)
        os.replace(str(tmp), str(path))
    except OSError:
        try:
            if tmp.is_file():
                tmp.unlink(missing_ok=True)
        except OSError:
            pass


def read_hardware_history(app) -> Tuple[List[Dict[str, Any]], str]:
    """
    Точки за окно хранения (по конфигу, по умолчанию 3 суток).
    Возвращает (points, path_or_empty).
    """
    path_str = _hardware_log_path(app)
    if not path_str:
        return [], ""

    path = Path(path_str)
    if not path.is_absolute():
        base = app.config.get("BASE_DIR", ".")
        path = Path(base) / path

    if not path.is_file():
        return [], str(path)

    days = int(app.config.get("WESP_ADMIN_HARDWARE_LOG_RETENTION_DAYS", 3) or 3)
    days = max(1, min(days, 30))
    cutoff = time.time() - float(days * 86400)

    out: List[Dict[str, Any]] = []
    try:
        with path.open("r", encoding="utf-8") as f:
            for ln in f:
                ln = ln.strip()
                if not ln:
                    continue
                try:
                    obj = json.loads(ln)
                except json.JSONDecodeError:
                    continue
                ts = _line_unix_ts(obj)
                if ts is None or ts < cutoff:
                    continue
                out.append(
                    {
                        "t": int(ts),
                        "cpu": float(obj.get("cpu_pct") or 0),
                        "ram": float(obj.get("ram_pct") or 0),
                        "swap": float(obj.get("swap_pct") or 0),
                        "disk": float(obj.get("disk_pct") or 0)
                        if obj.get("disk_pct") is not None
                        else None,
                    }
                )
    except OSError:
        return [], str(path)

    out.sort(key=lambda x: x["t"])
    return out, str(path)