Spaces:
Running
Running
File size: 2,560 Bytes
86211da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """日志查询 API（凭 log API key 鉴权）：/api/logs/*
用途：不依赖 admin key，凭可吊销的随机 API key 查询请求日志，
方便外部工具/脚本排查线上问题（key 由 /admin/api/log-keys 生成）。
鉴权方式（二选一）：
- 请求头 X-XTC-Log-Key: <key>（推荐，避免 key 进代理/访问日志）
- query 参数 ?key=<key>
注意：/api/logs 不在 request_log 中间件的记录范围内（只记录 /v1/ /admin/api/ /u/api/），
查询行为本身不会被记录，避免 key 泄露到日志存储。
"""
from __future__ import annotations
from typing import Optional
from fastapi import APIRouter, Query, Request
from ..errors import HttpError
from ..services import log_api_key_store, request_log_store
from ._common import ok_with_cors
router = APIRouter(prefix="/api/logs", tags=["log-query"])
def _require_log_key(request: Request) -> str:
key = request.headers.get("x-xtc-log-key") or request.query_params.get("key") or ""
if not key:
raise HttpError(
"missing log api key: pass header X-XTC-Log-Key or query ?key=",
status=401, code="unauthorized",
)
if not log_api_key_store.verify_key(key):
raise HttpError(
"log api key invalid or revoked",
status=401, code="unauthorized",
)
return key
@router.get("")
async def list_logs(
request: Request,
limit: int = Query(default=50, ge=1, le=500),
offset: int = Query(default=0, ge=0),
path: Optional[str] = Query(default=None),
method: Optional[str] = Query(default=None),
status: Optional[str] = Query(default=None),
ok: Optional[int] = Query(default=None, ge=0, le=1),
hours: Optional[int] = Query(default=None, ge=1, le=720),
keyword: Optional[str] = Query(default=None),
category: Optional[str] = Query(default="business"),
device_id: Optional[str] = Query(default=None),
):
_require_log_key(request)
result = request_log_store.list_records(
limit=limit, offset=offset,
path=path, method=method, status=status,
ok=bool(ok) if ok is not None else None,
hours=hours, keyword=keyword, category=category,
device_id=device_id,
)
return ok_with_cors(result)
@router.get("/{rid}")
async def get_log_detail(rid: int, request: Request):
_require_log_key(request)
rec = request_log_store.get_record(rid)
if not rec:
raise HttpError(f"request log not found: {rid}", status=404, code="not_found")
return ok_with_cors({"record": rec})
|