101 lines
3.2 KiB
Python
101 lines
3.2 KiB
Python
"""services/logview.py —— 读取行为日志文件(JSONL)+ 内存聚合,供管理后台日志查看。"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def _select_files(log_dir: Path, from_date: str | None, to_date: str | None) -> list[Path]:
|
|
"""按日期范围选出 audit-YYYY-MM-DD.log(升序)。"""
|
|
if not log_dir.is_dir():
|
|
return []
|
|
out: list[Path] = []
|
|
for p in sorted(log_dir.glob("audit-*.log")):
|
|
name = p.name
|
|
if not (name.startswith("audit-") and name.endswith(".log")):
|
|
continue
|
|
stem = name[len("audit-"):-len(".log")]
|
|
if from_date and stem < from_date:
|
|
continue
|
|
if to_date and stem > to_date:
|
|
continue
|
|
out.append(p)
|
|
return out
|
|
|
|
|
|
def _read_lines(path: Path) -> list[dict[str, Any]]:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
text = f.read()
|
|
except (FileNotFoundError, OSError):
|
|
return []
|
|
out: list[dict[str, Any]] = []
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
out.append(json.loads(line))
|
|
except (json.JSONDecodeError, TypeError):
|
|
continue
|
|
return out
|
|
|
|
|
|
def _match(e: dict[str, Any], event, user_id, q, status) -> bool:
|
|
if event and e.get("event") != event:
|
|
return False
|
|
if user_id and e.get("user_id") != user_id:
|
|
return False
|
|
if status is not None and str(e.get("status")) != str(status):
|
|
return False
|
|
if q:
|
|
q = str(q)
|
|
if q not in str(e.get("path") or "") and q not in str(e.get("user_id") or ""):
|
|
return False
|
|
return True
|
|
|
|
|
|
async def query_logs(
|
|
log_dir: str,
|
|
*,
|
|
from_date: str | None = None,
|
|
to_date: str | None = None,
|
|
event: str | None = None,
|
|
user_id: str | None = None,
|
|
q: str | None = None,
|
|
status: str | None = None,
|
|
page: int = 1,
|
|
page_size: int = 50,
|
|
) -> dict[str, Any]:
|
|
files = _select_files(Path(log_dir), from_date, to_date)
|
|
# 文件 I/O 放线程池,不阻塞事件循环
|
|
all_lines = await asyncio.to_thread(lambda: [ln for p in files for ln in _read_lines(p)])
|
|
items = [e for e in all_lines if _match(e, event, user_id, q, status)]
|
|
items.sort(key=lambda e: str(e.get("ts") or ""), reverse=True)
|
|
total = len(items)
|
|
|
|
by_status: dict[str, int] = {}
|
|
latencies: list[float] = []
|
|
denied = 0
|
|
for e in items:
|
|
st = e.get("status")
|
|
if st is not None:
|
|
key = str(st)
|
|
by_status[key] = by_status.get(key, 0) + 1
|
|
if e.get("event") in ("call_denied", "login_fail", "admin_login_fail"):
|
|
denied += 1
|
|
lm = e.get("latency_ms")
|
|
if isinstance(lm, (int, float)):
|
|
latencies.append(float(lm))
|
|
latencies.sort()
|
|
start = (page - 1) * page_size
|
|
aggregates = {
|
|
"by_status": by_status,
|
|
"denied": denied,
|
|
"avg_latency_ms": round(sum(latencies) / len(latencies), 1) if latencies else None,
|
|
"p95_latency_ms": latencies[int(len(latencies) * 0.95)] if latencies else None,
|
|
}
|
|
return {"items": items[start:start + page_size], "total": total, "aggregates": aggregates}
|