buff3.0/backend-client-bak/app/utils/api_logger.py

"""
API接口日志记录器
记录入参密文、入参明文、出参明文、出参密文
日志文件按天滚动，每天一个日志文件
"""
import os
import json
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Optional
from loguru import logger as loguru_logger

# 日志配置
API_LOG_DIR = Path("logs/api")
API_LOG_DIR.mkdir(parents=True, exist_ok=True)

# 日志文件按天滚动
def get_api_log_path() -> Path:
    """获取今天的API日志文件路径"""
    return API_LOG_DIR / f"api_{datetime.now().strftime('%Y-%m-%d')}.log"

# 敏感字段脱敏
REDACTED_FIELDS = {
    "password",
    "card_code",
    "shared_secret",
    "identity_secret",
    "aes_key",
    "encrypted_aes_key",
    "signature",
    "access_token",
    "refresh_token",
    "session_token",
    "steamLoginSecure",
    "machine_code",
}


def _redact_value(value: Any) -> Any:
    """脱敏单个值（敏感字段统一显示为 ****）"""
    if value is None:
        return None
    if isinstance(value, str):
        if len(value) > 8:
            return value[:2] + "****" + value[-2:] if len(value) > 4 else "****"
        return "****" if value else ""
    return "****"


def redact_for_db(data: Any, max_str_len: int = 2000) -> Optional[str]:
    """
    将数据脱敏后序列化为 JSON 字符串，用于写入 api_logs 表（明文记录，符合文档 4.5）。
    敏感字段会替换为 ****。
    """
    if data is None:
        return None
    try:
        redacted = _redact_data(data)
        truncated = _truncate_data(redacted, max_str_len)
        s = json.dumps(truncated, ensure_ascii=False, default=str)
        return s[:max_str_len] + ("..." if len(s) > max_str_len else "")
    except Exception:
        return None


def _redact_data(data: Any, max_length: int = 10000) -> Any:
    """递归脱敏数据中的敏感字段"""
    if data is None:
        return None
    if isinstance(data, dict):
        redacted = {}
        for k, v in data.items():
            if k in REDACTED_FIELDS:
                redacted[k] = _redact_value(v)
            else:
                redacted[k] = _redact_data(v)
        return redacted
    if isinstance(data, list):
        return [_redact_data(item) for item in data]
    return data


def _truncate_data(data: Any, max_length: int = 10000) -> Any:
    """截断过长的数据"""
    if data is None:
        return None
    if isinstance(data, dict):
        truncated = {}
        for k, v in data.items():
            truncated[k] = _truncate_data(v)
        return truncated
    if isinstance(data, list):
        return data[:1000]  # 最多1000个元素
    if isinstance(data, str) and len(data) > max_length:
        return data[:max_length] + "...[truncated]"
    return data


class APILogger:
    """API接口日志记录器"""

    def __init__(self):
        self._log_path = get_api_log_path()
        self._setup_logger()

    def _setup_logger(self):
        """配置日志处理器"""
        # 移除默认处理器
        try:
            loguru_logger.remove()
        except Exception:
            pass

        # 添加文件处理器
        self._handler_id = loguru_logger.add(
            str(self._log_path),
            rotation="00:00",  # 每天午夜滚动
            retention="30 days",
            format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {message}",
            level="DEBUG",
            encoding="utf-8",
            serialize=False,
        )

    def log_api_call(
        self,
        endpoint: str,
        method: str,
        encrypted_request: Optional[str] = None,
        decrypted_request: Optional[Dict[str, Any]] = None,
        encrypted_response: Optional[str] = None,
        decrypted_response: Optional[Dict[str, Any]] = None,
        processing_time_ms: int = 0,
        status_code: int = 200,
        user_id: Optional[int] = None,
        client_ip: Optional[str] = None,
        error_message: Optional[str] = None,
    ):
        """
        记录完整的API调用信息

        Args:
            endpoint: 接口路径
            method: HTTP方法
            encrypted_request: 加密的请求数据
            decrypted_request: 解密后的请求数据
            encrypted_response: 加密的响应数据
            decrypted_response: 解密后的响应数据
            processing_time_ms: 处理时间(毫秒)
            status_code: HTTP状态码
            user_id: 用户ID
            client_ip: 客户端IP
            error_message: 错误信息
        """
        try:
            # 构建日志条目
            log_entry = {
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
                "endpoint": endpoint,
                "method": method,
                "user_id": user_id,
                "client_ip": client_ip,
                "processing_time_ms": processing_time_ms,
                "status_code": status_code,
                "error": error_message,
                "request": {
                    "encrypted": encrypted_request,
                    "decrypted": _truncate_data(_redact_data(decrypted_request)),
                },
                "response": {
                    "encrypted": encrypted_response,
                    "decrypted": _truncate_data(_redact_data(decrypted_response)),
                },
            }

            # 写入日志文件
            log_line = json.dumps(log_entry, ensure_ascii=False)
            loguru_logger.debug(log_line)

        except Exception as e:
            loguru_logger.error(f"记录API日志失败: {e}")

    def log_request_received(
        self,
        endpoint: str,
        method: str,
        encrypted_request: str,
        user_id: Optional[int] = None,
        client_ip: Optional[str] = None,
    ):
        """记录收到请求（仅入参密文）"""
        try:
            log_entry = {
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
                "event": "REQUEST_RECEIVED",
                "endpoint": endpoint,
                "method": method,
                "user_id": user_id,
                "client_ip": client_ip,
                "encrypted_request": encrypted_request,
            }
            loguru_logger.debug(json.dumps(log_entry, ensure_ascii=False))
        except Exception as e:
            loguru_logger.error(f"记录请求日志失败: {e}")

    def log_request_decrypted(
        self,
        endpoint: str,
        method: str,
        decrypted_request: Dict[str, Any],
        user_id: Optional[int] = None,
    ):
        """记录请求解密完成（入参明文）"""
        try:
            log_entry = {
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
                "event": "REQUEST_DECRYPTED",
                "endpoint": endpoint,
                "method": method,
                "user_id": user_id,
                "decrypted_request": _truncate_data(_redact_data(decrypted_request)),
            }
            loguru_logger.debug(json.dumps(log_entry, ensure_ascii=False))
        except Exception as e:
            loguru_logger.error(f"记录请求解密日志失败: {e}")

    def log_response_ready(
        self,
        endpoint: str,
        method: str,
        decrypted_response: Dict[str, Any],
        user_id: Optional[int] = None,
    ):
        """记录响应准备完成（出参明文）"""
        try:
            log_entry = {
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
                "event": "RESPONSE_READY",
                "endpoint": endpoint,
                "method": method,
                "user_id": user_id,
                "decrypted_response": _truncate_data(_redact_data(decrypted_response)),
            }
            loguru_logger.debug(json.dumps(log_entry, ensure_ascii=False))
        except Exception as e:
            loguru_logger.error(f"记录响应准备日志失败: {e}")

    def log_response_sent(
        self,
        endpoint: str,
        method: str,
        encrypted_response: str,
        processing_time_ms: int,
        status_code: int,
        user_id: Optional[int] = None,
        error_message: Optional[str] = None,
    ):
        """记录响应发送完成（出参密文）"""
        try:
            log_entry = {
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],
                "event": "RESPONSE_SENT",
                "endpoint": endpoint,
                "method": method,
                "user_id": user_id,
                "processing_time_ms": processing_time_ms,
                "status_code": status_code,
                "error": error_message,
                "encrypted_response": encrypted_response[:500] + "..." if len(encrypted_response or "") > 500 else encrypted_response,
            }
            loguru_logger.debug(json.dumps(log_entry, ensure_ascii=False))
        except Exception as e:
            loguru_logger.error(f"记录响应发送日志失败: {e}")


# 全局实例
api_logger = APILogger()


def log_api_call(
    endpoint: str,
    method: str,
    encrypted_request: Optional[str] = None,
    decrypted_request: Optional[Dict[str, Any]] = None,
    encrypted_response: Optional[str] = None,
    decrypted_response: Optional[Dict[str, Any]] = None,
    processing_time_ms: int = 0,
    status_code: int = 200,
    user_id: Optional[int] = None,
    client_ip: Optional[str] = None,
    error_message: Optional[str] = None,
):
    """便捷的API调用日志记录函数"""
    api_logger.log_api_call(
        endpoint=endpoint,
        method=method,
        encrypted_request=encrypted_request,
        decrypted_request=decrypted_request,
        encrypted_response=encrypted_response,
        decrypted_response=decrypted_response,
        processing_time_ms=processing_time_ms,
        status_code=status_code,
        user_id=user_id,
        client_ip=client_ip,
        error_message=error_message,
    )