"""操作审计记录服务，统一处理详情脱敏与序列化。"""

from __future__ import annotations

import json
from collections.abc import Mapping
from typing import Any

from sqlalchemy.orm import Session

from ..models import AuditLog, User

_SENSITIVE_KEY_PARTS = (
    "cookie",
    "token",
    "password",
    "passwd",
    "secret",
    "signature",
    "sign",
    "qr_data",
    "authorization",
    "credential",
    "private_key",
)


def _safe_value(value: Any, key: str = "") -> Any:
    """递归过滤敏感字段，避免把凭据、二维码和签名写入审计表。"""
    key_lower = key.lower()
    if any(part in key_lower for part in _SENSITIVE_KEY_PARTS):
        return "[已脱敏]"
    if isinstance(value, Mapping):
        return {str(k): _safe_value(v, str(k)) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [_safe_value(item) for item in value]
    if isinstance(value, (str, int, float, bool)) or value is None:
        return value
    return str(value)


def record_audit(
    db: Session,
    current: User | None,
    *,
    action: str,
    target: str = "",
    detail: Mapping[str, Any] | str | None = None,
    success: bool = True,
) -> AuditLog:
    """加入一条审计记录；调用方负责与业务变更一起提交事务。"""
    if isinstance(detail, Mapping):
        detail_text = json.dumps(
            _safe_value(detail), ensure_ascii=False, separators=(",", ":")
        )
    elif detail is None:
        detail_text = ""
    else:
        detail_text = str(_safe_value(detail))
    if not success:
        detail_text = json.dumps(
            {"message": detail_text}, ensure_ascii=False, separators=(",", ":")
        )
    entry = AuditLog(
        user_id=current.id if current else None,
        username=current.username if current else "supplier",
        action=action,
        target=target,
        detail=detail_text,
        success=success,
    )
    db.add(entry)
    return entry