95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""Журнал действий UI админки (JSONL): только события с клиента (setStatus). Сервер — в wesp.log, см. activity-feed."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
_LOG = logging.getLogger(__name__)
|
|
_LOCK = threading.Lock()
|
|
|
|
|
|
def _activity_log_path(app) -> str:
|
|
return (app.config.get("WESP_ADMIN_UI_ACTIVITY_LOG_PATH") or "").strip()
|
|
|
|
|
|
def _line_unix_ts(obj: Dict[str, Any]) -> float | None:
|
|
u = obj.get("u")
|
|
if isinstance(u, (int, float)):
|
|
return float(u)
|
|
return None
|
|
|
|
|
|
def append_admin_ui_activity(app, *, text: str, level: str, user_login: str) -> None:
|
|
"""Дописывает одну строку JSON; строки старше WESP_ADMIN_UI_ACTIVITY_RETENTION_DAYS удаляются."""
|
|
try:
|
|
_append_admin_ui_activity_impl(app, text=text, level=level, user_login=user_login)
|
|
except Exception:
|
|
_LOG.exception("Сбой журнала UI-активности (WESP_ADMIN_UI_ACTIVITY_LOG_PATH).")
|
|
|
|
|
|
def _append_admin_ui_activity_impl(app, *, text: str, level: str, user_login: str) -> None:
|
|
path_str = _activity_log_path(app)
|
|
if not path_str or app.config.get("TESTING"):
|
|
return
|
|
|
|
path = Path(path_str)
|
|
if not path.is_absolute():
|
|
base = app.config.get("BASE_DIR", ".")
|
|
path = Path(base) / path
|
|
|
|
row = {
|
|
"u": int(time.time()),
|
|
"level": level if level in ("ok", "err", "warn") else "ok",
|
|
"user": (user_login or "")[:128],
|
|
"text": (text or "")[:2000],
|
|
}
|
|
line = json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n"
|
|
|
|
days = int(app.config.get("WESP_ADMIN_UI_ACTIVITY_RETENTION_DAYS", 3) or 3)
|
|
days = max(1, min(days, 30))
|
|
cutoff = time.time() - float(days * 86400)
|
|
|
|
with _LOCK:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
_rewrite_prune_and_append(path, cutoff, line)
|
|
|
|
|
|
def _rewrite_prune_and_append(path: Path, cutoff_ts: float, new_line: str) -> None:
|
|
lines_kept: List[str] = []
|
|
if path.is_file():
|
|
try:
|
|
with path.open("r", encoding="utf-8") as f:
|
|
for ln in f:
|
|
raw = ln.strip()
|
|
if not raw:
|
|
continue
|
|
try:
|
|
obj = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
ts = _line_unix_ts(obj)
|
|
if ts is not None and ts >= cutoff_ts:
|
|
lines_kept.append(raw + "\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
|