"""日志接收(合并原 xtc-log-receiver.mjs)。 端点: - POST /logs/api/log 单条日志 - POST /logs/api/log/batch 批量日志 - GET /logs/api/stats 统计 - GET /logs/api/logs 查询(admin) """ from __future__ import annotations import json import time from typing import Any, Optional from fastapi import APIRouter, Depends, Query from ..auth import require_admin_key from ..database import get_conn from ._common import ok_with_cors router = APIRouter(prefix="/logs/api", tags=["logs"]) def _insert_log(level: str, module: str, event: str, trace_id: Optional[str], data: Any) -> None: now = int(time.time()) data_str = json.dumps(data, ensure_ascii=False) if data is not None else None with get_conn() as conn: conn.execute( "INSERT INTO app_logs(ts, level, module, event, trace_id, data) VALUES(?,?,?,?,?,?)", (now, level, module, event, trace_id, data_str), ) @router.post("/log") async def post_log(body: dict) -> dict: level = str(body.get("level") or "info").lower() module = str(body.get("module") or "client") event = str(body.get("event") or "log") trace_id = body.get("traceId") or body.get("trace_id") data = body.get("data") or body _insert_log(level, module, event, trace_id, data) return ok_with_cors({"accepted": True}) @router.post("/log/batch") async def post_log_batch(body: dict) -> dict: items = body.get("logs") or body.get("items") or [] if not isinstance(items, list): items = [] count = 0 for item in items: if not isinstance(item, dict): continue _insert_log( str(item.get("level") or "info").lower(), str(item.get("module") or "client"), str(item.get("event") or "log"), item.get("traceId") or item.get("trace_id"), item.get("data") or item, ) count += 1 return ok_with_cors({"accepted": count}) @router.get("/stats") async def stats( hours: int = Query(default=24, ge=1, le=720), _admin: str = Depends(require_admin_key), ) -> dict: since = int(time.time()) - hours * 3600 with get_conn() as conn: by_level = conn.execute( "SELECT level, COUNT(*) as c FROM app_logs WHERE ts >= ? GROUP BY level", (since,), ).fetchall() by_module = conn.execute( "SELECT module, COUNT(*) as c FROM app_logs WHERE ts >= ? GROUP BY module ORDER BY c DESC LIMIT 20", (since,), ).fetchall() total = conn.execute( "SELECT COUNT(*) as c FROM app_logs WHERE ts >= ?", (since,) ).fetchone() return ok_with_cors( { "hours": hours, "total": int(total["c"]) if total else 0, "by_level": {r["level"]: int(r["c"]) for r in by_level}, "by_module": {r["module"]: int(r["c"]) for r in by_module}, } ) @router.get("/logs") async def list_logs( hours: int = Query(default=24, ge=1, le=720), level: Optional[str] = Query(default=None), limit: int = Query(default=200, ge=1, le=2000), offset: int = Query(default=0, ge=0), _admin: str = Depends(require_admin_key), ) -> dict: since = int(time.time()) - hours * 3600 sql = "SELECT id, ts, level, module, event, trace_id, data FROM app_logs WHERE ts >= ?" args: list = [since] if level: sql += " AND level = ?" args.append(level.lower()) sql += " ORDER BY ts DESC, id DESC LIMIT ? OFFSET ?" args += [limit, offset] with get_conn() as conn: rows = conn.execute(sql, args).fetchall() items = [] for r in rows: try: data = json.loads(r["data"]) if r["data"] else None except Exception: data = r["data"] items.append( { "id": int(r["id"]), "ts": int(r["ts"]), "level": r["level"], "module": r["module"], "event": r["event"], "trace_id": r["trace_id"], "data": data, } ) return ok_with_cors({"items": items, "count": len(items)})