"""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}