shanxia2.0/backend/app/core/api_request_log_middleware.py

"""统一记录 API 请求/响应出入参的中间件"""
import asyncio
import json
import logging
import time
from typing import Any, Optional, Tuple
from urllib.parse import parse_qsl, urlencode

from fastapi import HTTPException
from jose import jwt
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

from app.core.config import settings
from app.core.database import SessionLocal
from app.core.oper_log_context import clear_oper_log_meta, snapshot_oper_log_meta
from app.core.oper_log_infer import business_type_from_path, title_from_path

logger = logging.getLogger("sqxp")

_SENSITIVE_KEYS = frozenset(
    {
        "password",
        "oldpassword",
        "newpassword",
        "encryptedlogin",
        "authorization",
        "token",
        "refreshtoken",
        "apikey",
        "sessionid",
        "authorizeurl",
        "verificationcode",
        "openid",
        "appsecret",
        "wechat_oa_secret",
    }
)

_OMIT_REQ_BYTES = 10 * 1024 * 1024


def _truncate(s: str, max_len: int) -> str:
    if not s:
        return s
    if len(s) <= max_len:
        return s
    return s[:max_len] + f"\n... [truncated, total {len(s)} chars]"


def _redact_notice_query(path: str, query: str) -> str:
    if "/wechat-notices/" not in path:
        return query
    return urlencode([(key, "***" if key.lower() in ("code", "state", "sessionid") else value)
                      for key, value in parse_qsl(query, keep_blank_values=True)])


# MySQL TEXT 列上限为 65535 字节（非字符）；utf8mb4 下中文等多字节字符需按字节截断。
_DB_TEXT_MAX_BYTES = 60000


def _truncate_utf8_bytes(s: str, max_bytes: int = _DB_TEXT_MAX_BYTES) -> str:
    if not s:
        return s
    encoded = s.encode("utf-8")
    if len(encoded) <= max_bytes:
        return s
    truncated = encoded[:max_bytes]
    while truncated and (truncated[-1] & 0xC0) == 0x80:
        truncated = truncated[:-1]
    text = truncated.decode("utf-8", errors="ignore")
    return (
        text
        + f"\n... [truncated, total {len(s)} chars, {len(encoded)} bytes]"
    )


def _redact_obj(obj: Any) -> Any:
    if isinstance(obj, dict):
        out = {}
        for k, v in obj.items():
            lk = str(k).lower()
            if (
                lk in _SENSITIVE_KEYS
                or "password" in lk
                or lk in ("encryptedlogin", "refreshtoken")
            ):
                out[k] = "***"
            else:
                out[k] = _redact_obj(v)
        return out
    if isinstance(obj, list):
        return [_redact_obj(x) for x in obj[:200]] + (["..."] if len(obj) > 200 else [])
    return obj


def _format_body_bytes(raw: bytes, content_type: str, max_bytes: int) -> str:
    if not raw:
        return ""
    if len(raw) > _OMIT_REQ_BYTES:
        return f"[body omitted: {len(raw)} bytes]"
    if len(raw) > max_bytes:
        raw = raw[:max_bytes]
    ct = (content_type or "").lower()
    if "multipart/form-data" in ct:
        return "[multipart/form-data body omitted]"
    if "application/json" in ct or ct == "" or "text/" in ct:
        try:
            text = raw.decode("utf-8", errors="replace")
            try:
                parsed = json.loads(text)
                parsed = _redact_obj(parsed)
                return _truncate(
                    json.dumps(parsed, ensure_ascii=False),
                    settings.API_REQUEST_LOG_MAX_BODY,
                )
            except json.JSONDecodeError:
                return _truncate(text, settings.API_REQUEST_LOG_MAX_BODY)
        except Exception:
            return f"[decode error, {len(raw)} bytes]"
    return _truncate(raw.decode("utf-8", errors="replace"), settings.API_REQUEST_LOG_MAX_BODY)


def _claims_from_auth(
    auth_header: Optional[str],
) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str], Optional[str]]:
    if not auth_header or not auth_header.startswith("Bearer "):
        return None, None, None, None, None
    token = auth_header[7:].strip()
    if not token:
        return None, None, None, None, None
    try:
        payload = jwt.decode(
            token,
            settings.SECRET_KEY,
            algorithms=[settings.ALGORITHM],
            options={"verify_signature": False, "verify_exp": False},
        )
        return (
            payload.get("sub"),
            payload.get("username"),
            payload.get("role"),
            payload.get("real_name"),
            payload.get("tenant_no"),
        )
    except Exception:
        return None, None, None, None, None


def _should_skip_exclude(path: str) -> bool:
    raw = (settings.API_REQUEST_LOG_EXCLUDE_PATHS or "").strip()
    if not raw:
        return False
    for item in raw.split(","):
        prefix = item.strip()
        if not prefix:
            continue
        if path == prefix or path.startswith(prefix):
            return True
    return False


def _should_skip(request: Request) -> bool:
    p = request.url.path
    if request.method == "OPTIONS":
        return True
    if p in ("/", "/health", "/docs", "/redoc", "/openapi.json"):
        return True
    if p.startswith("/docs") or p.startswith("/redoc"):
        return True
    if _should_skip_exclude(p):
        return True
    return False


def _persist_log(
    *,
    http_method: str,
    path: str,
    query_string: Optional[str],
    request_body: Optional[str],
    response_body: Optional[str],
    status_code: int,
    duration_ms: int,
    user_id: Optional[str],
    username: Optional[str],
    role: Optional[str],
    ip_address: Optional[str],
    user_agent: Optional[str],
    create_name: Optional[str] = None,
    tenant_no: Optional[str] = None,
    business_title: Optional[str] = None,
    business_type: Optional[int] = None,
    business_id: Optional[str] = None,
    oper_summary: Optional[str] = None,
) -> None:
    from datetime import datetime

    from app.core.utils import generate_id
    from app.models.models import SysApiRequestLog

    db = SessionLocal()
    try:
        row = SysApiRequestLog(
            id=generate_id(),
            http_method=http_method,
            path=_truncate(path, 500),
            query_string=_truncate(query_string or "", 2000) or None,
            request_body=_truncate_utf8_bytes(request_body) if request_body else None,
            response_body=_truncate_utf8_bytes(response_body) if response_body else None,
            status_code=status_code,
            duration_ms=duration_ms,
            user_id=user_id,
            username=_truncate(username or "", 100) if username else None,
            role=role,
            ip_address=ip_address,
            user_agent=_truncate(user_agent or "", 500) if user_agent else None,
            tenant_no=_truncate(tenant_no or "", 64) if tenant_no else None,
            business_title=_truncate(business_title or "", 100) if business_title else None,
            business_type=business_type,
            business_id=_truncate(business_id or "", 64) if business_id else None,
            oper_summary=_truncate(oper_summary or "", 500) if oper_summary else None,
            is_del=0,
            create_at=datetime.now(),
            create_by=user_id,
            create_name=_truncate(create_name or "", 100) if create_name else None,
        )
        db.add(row)
        db.commit()
    except Exception as e:
        logger.warning("api_request_log persist failed: %s", e)
        db.rollback()
    finally:
        db.close()


def _schedule_persist_log(**kwargs) -> None:
    """异步提交日志写入任务，不阻塞当前请求。"""
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        # 无事件循环时（极少数场景）回退为同步写入，保证日志不丢。
        try:
            _persist_log(**kwargs)
        except Exception as e:
            logger.warning("api_request_log persist fallback failed: %s", e)
        return

    task = loop.create_task(asyncio.to_thread(_persist_log, **kwargs))

    def _on_done(t: asyncio.Task) -> None:
        try:
            t.result()
        except Exception as e:
            logger.warning("api_request_log async task failed: %s", e)

    task.add_done_callback(_on_done)


class ApiRequestLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if not settings.API_REQUEST_LOG_ENABLED or _should_skip(request):
            return await call_next(request)

        max_b = settings.API_REQUEST_LOG_MAX_BODY_BYTES
        content_type = request.headers.get("content-type", "")

        is_doclib = request.url.path.startswith(f"{settings.API_V1_PREFIX}/web/doclib/")
        if request.url.path == f"{settings.API_V1_PREFIX}/web/doclib/upload":
            # 文档由 UploadFile 临时文件接收；审计保留元数据，不复制整份正文到内存。
            req_text = ("[multipart/form-data body omitted]" if "multipart/form-data" in content_type.lower()
                        else "[document upload body omitted]")
        else:
            try:
                body = await request.body()
            except Exception as e:
                body = b""
                logger.warning("api_request_log read body failed: %s", e)
            req_text = _format_body_bytes(body, content_type, max_b)

            async def receive():
                return {"type": "http.request", "body": body, "more_body": False}

            request = Request(request.scope, receive)

        uid, uname, role, real_name, tenant_no = _claims_from_auth(request.headers.get("authorization"))
        ip = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
        if not ip and request.client:
            ip = request.client.host
        ua = request.headers.get("user-agent", "")

        start = time.perf_counter()
        status_code = 500
        resp_text = ""
        raw_resp = b""
        out: Optional[Response] = None
        try:
            response = await call_next(request)
            status_code = response.status_code
            if is_doclib and response.headers.get("content-disposition", "").startswith("attachment;"):
                # ACL 已在路由校验；保留认证/审计及原始流，不把大文件合并成 bytes。
                resp_text = "[document attachment body omitted]"
                out = response
            else:
                if hasattr(response, "body_iterator"):
                    resp_chunks: list[bytes] = []
                    async for chunk in response.body_iterator:
                        resp_chunks.append(chunk)
                    raw_resp = b"".join(resp_chunks)
                elif getattr(response, "body", None) is not None:
                    raw_resp = response.body
                ct_out = response.headers.get("content-type", "")
                resp_text = _format_body_bytes(raw_resp, ct_out, max_b)
                if request.url.path.endswith("/public/wechat-notices/oauth/callback"):
                    resp_text = "[微信绑定确认页面：正文已脱敏]"

                headers = {k: v for k, v in response.headers.items() if k.lower() != "content-length"}
                out = Response(
                    content=raw_resp,
                    status_code=response.status_code,
                    headers=headers,
                    media_type=response.media_type,
                    background=response.background,
                )
        except HTTPException as e:
            status_code = e.status_code
            d = e.detail
            if not isinstance(d, str):
                try:
                    d = json.dumps(d, ensure_ascii=False)
                except Exception:
                    d = str(d)
            resp_text = _truncate(d, settings.API_REQUEST_LOG_MAX_BODY)
            raise
        except Exception as e:
            resp_text = _truncate(str(e), settings.API_REQUEST_LOG_MAX_BODY)
            raise
        finally:
            elapsed_ms = int((time.perf_counter() - start) * 1000)
            path = request.url.path
            qs = _redact_notice_query(path, request.url.query)
            meta = snapshot_oper_log_meta()
            biz_title = meta.business_title if meta and meta.business_title else title_from_path(path)
            biz_type = (
                meta.business_type
                if meta and meta.business_type is not None
                else business_type_from_path(path, request.method)
            )
            biz_id = meta.business_id if meta else None
            oper_summary = meta.oper_summary if meta else None
            effective_tenant = tenant_no or settings.DEFAULT_TENANT_NO
            _schedule_persist_log(
                http_method=request.method,
                path=path,
                query_string=qs or None,
                request_body=req_text or None,
                response_body=resp_text or None,
                status_code=status_code,
                duration_ms=elapsed_ms,
                user_id=uid,
                username=uname,
                role=role,
                ip_address=ip,
                user_agent=ua,
                create_name=real_name,
                tenant_no=effective_tenant,
                business_title=biz_title,
                business_type=biz_type,
                business_id=biz_id,
                oper_summary=oper_summary,
            )
            clear_oper_log_meta()

        return out