app/api/log_query.py

"""日志查询 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})