220 lines
7.6 KiB
Python
220 lines
7.6 KiB
Python
"""审计日志查询 API（迭代第 4 轮）。
|
|
|
|
迭代第 3 轮把审计写入了数据库，但**没有查询入口**——写了看不到等于没写。
|
|
本模块提供：
|
|
- `GET /api/v1/core/audit-logs/` 操作轨迹列表（按目标/动作/用户筛选）
|
|
- `GET /api/v1/core/audit-logs/summary/` 按动作类型的统计（近 N 天）
|
|
|
|
设计：只读、按租户隔离、默认最近 100 条。审计日志本身**不可修改/删除**
|
|
（没有任何写接口），保证证据链完整。
|
|
"""
|
|
|
|
import json
|
|
|
|
from adrf.decorators import api_view
|
|
from asgiref.sync import sync_to_async
|
|
from django.db.models import Count
|
|
from django.utils import timezone
|
|
from datetime import timedelta
|
|
from rest_framework.response import Response
|
|
from rest_framework.exceptions import ValidationError
|
|
|
|
from apps.core.viewset import resolve_tenant
|
|
|
|
# 业务动作 → 中文标签（前端展示用）
|
|
ACTION_LABELS = {
|
|
"post": "过账",
|
|
"cancel": "作废",
|
|
"force": "管控放行",
|
|
"price_override": "改价",
|
|
"quota_exceeded": "配额超限",
|
|
"issue": "签发",
|
|
"plan_change": "套餐变更",
|
|
"seed": "数据初始化",
|
|
}
|
|
|
|
# 风险等级：越权/放行类动作需要重点关注
|
|
ACTION_RISK = {
|
|
"force": "high",
|
|
"price_override": "medium",
|
|
"plan_change": "medium",
|
|
"quota_exceeded": "low",
|
|
"post": "info",
|
|
"cancel": "info",
|
|
"issue": "info",
|
|
"seed": "info",
|
|
}
|
|
|
|
|
|
async def _tenant_or_error(request):
|
|
code = request.META.get("HTTP_X_TENANT_ID", "")
|
|
tenant = await sync_to_async(resolve_tenant)(code)
|
|
if tenant is None:
|
|
raise ValidationError({"tenant": "无法识别租户"})
|
|
return tenant
|
|
|
|
|
|
@api_view(["GET"])
|
|
async def audit_log_list(request):
|
|
"""操作轨迹列表。
|
|
|
|
参数：target_type / target_id / action / user_id / days / limit
|
|
"""
|
|
tenant = await _tenant_or_error(request)
|
|
from apps.core.audit import query_logs
|
|
|
|
target_type = request.query_params.get("target_type", "").strip()
|
|
target_id = request.query_params.get("target_id", "").strip()
|
|
action = request.query_params.get("action", "").strip()
|
|
days = request.query_params.get("days", "")
|
|
try:
|
|
limit = min(int(request.query_params.get("limit", 100)), 500)
|
|
except (TypeError, ValueError):
|
|
limit = 100
|
|
|
|
def _load():
|
|
from apps.core.models import AuditLog
|
|
|
|
qs = AuditLog.objects.filter(tenant=tenant).select_related("user")
|
|
if target_type:
|
|
qs = qs.filter(target_type=target_type)
|
|
if target_id:
|
|
qs = qs.filter(target_id=str(target_id))
|
|
if action:
|
|
qs = qs.filter(detail__business_action=action)
|
|
if days:
|
|
try:
|
|
since = timezone.now() - timedelta(days=int(days))
|
|
qs = qs.filter(created_at__gte=since)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
# 双键排序：同一秒内创建的多条记录 created_at 相同，
|
|
# 仅按时间排序不稳定（实测出现 SEQ4/SEQ2/SEQ3 的乱序），
|
|
# 加 -id 保证"最新在前"的顺序确定
|
|
return list(qs.order_by("-created_at", "-id")[:limit])
|
|
|
|
logs = await sync_to_async(_load)()
|
|
return Response({
|
|
"count": len(logs),
|
|
"results": [
|
|
{
|
|
"id": log.id,
|
|
"created_at": log.created_at.isoformat(),
|
|
"action": log.detail.get("business_action", log.action),
|
|
"action_label": ACTION_LABELS.get(
|
|
log.detail.get("business_action", ""), log.get_action_display()
|
|
),
|
|
"risk": ACTION_RISK.get(log.detail.get("business_action", ""), "info"),
|
|
"target_type": log.target_type,
|
|
"target_id": log.target_id,
|
|
"user": (
|
|
{"id": log.user_id, "username": log.user.username}
|
|
if log.user_id else None
|
|
),
|
|
"ip": str(log.ip) if log.ip else None,
|
|
"user_agent": log.user_agent,
|
|
"detail": log.detail,
|
|
}
|
|
for log in logs
|
|
],
|
|
})
|
|
|
|
|
|
@api_view(["GET"])
|
|
async def audit_log_summary(request):
|
|
"""按动作类型统计（近 N 天，默认 30）。
|
|
|
|
用于首页/风控页展示"最近有多少次管控放行"这类指标。
|
|
"""
|
|
tenant = await _tenant_or_error(request)
|
|
try:
|
|
days = int(request.query_params.get("days", 30))
|
|
except (TypeError, ValueError):
|
|
days = 30
|
|
|
|
def _load():
|
|
from apps.core.models import AuditLog
|
|
|
|
since = timezone.now() - timedelta(days=days)
|
|
qs = AuditLog.objects.filter(tenant=tenant, created_at__gte=since)
|
|
rows = (
|
|
qs.values("detail__business_action")
|
|
.annotate(n=Count("id"))
|
|
.order_by("-n")
|
|
)
|
|
total = qs.count()
|
|
# 高风险动作计数（放行类）
|
|
high_risk = sum(
|
|
r["n"] for r in rows
|
|
if ACTION_RISK.get(r["detail__business_action"] or "", "info") == "high"
|
|
)
|
|
return rows, total, high_risk
|
|
|
|
rows, total, high_risk = await sync_to_async(_load)()
|
|
return Response({
|
|
"days": days,
|
|
"total": total,
|
|
"high_risk_count": high_risk,
|
|
"items": [
|
|
{
|
|
"action": r["detail__business_action"] or "unknown",
|
|
"label": ACTION_LABELS.get(
|
|
r["detail__business_action"] or "", r["detail__business_action"] or "未知"
|
|
),
|
|
"risk": ACTION_RISK.get(r["detail__business_action"] or "", "info"),
|
|
"count": r["n"],
|
|
}
|
|
for r in rows
|
|
],
|
|
})
|
|
|
|
|
|
@api_view(["GET"])
|
|
async def audit_log_for_target(request):
|
|
"""某张单据的完整轨迹（"这单都发生了什么"）。
|
|
|
|
参数：target_type（必填）、target_id（必填）
|
|
"""
|
|
tenant = await _tenant_or_error(request)
|
|
target_type = request.query_params.get("target_type", "").strip()
|
|
target_id = request.query_params.get("target_id", "").strip()
|
|
if not (target_type and target_id):
|
|
raise ValidationError({"detail": "target_type 与 target_id 必填"})
|
|
|
|
from apps.core.audit import query_logs
|
|
|
|
def _load():
|
|
from django.db.models import Q
|
|
|
|
from apps.core.models import AuditLog
|
|
|
|
# 轨迹要覆盖"与这张单相关的所有操作"，而不仅是 target_type 完全匹配的：
|
|
# 例如强制放行记录挂在管控类型下（target_type=credit_limit），
|
|
# 但 detail.bill_no 才是它真正作用的单据。只按 target_type 过滤会把这些
|
|
# "同一单据的其他动作"漏掉，用户看到的轨迹是割裂的。
|
|
qs = AuditLog.objects.filter(tenant=tenant).filter(
|
|
Q(target_type=target_type, target_id=str(target_id))
|
|
| Q(detail__bill_no=str(target_id))
|
|
).select_related("user")
|
|
# 按 id 正序：同一秒内的操作也能还原真实先后顺序
|
|
return list(qs.order_by("id")[:200])
|
|
|
|
logs = await sync_to_async(_load)()
|
|
return Response({
|
|
"target": {"type": target_type, "id": target_id},
|
|
"count": len(logs),
|
|
"timeline": [
|
|
{
|
|
"at": log.created_at.isoformat(),
|
|
"action": log.detail.get("business_action"),
|
|
"action_label": ACTION_LABELS.get(
|
|
log.detail.get("business_action", ""), "操作"
|
|
),
|
|
"user": log.user.username if log.user_id else "系统",
|
|
"ip": str(log.ip) if log.ip else None,
|
|
"detail": log.detail,
|
|
}
|
|
for log in logs # 已按 id 正序（真实操作顺序）
|
|
],
|
|
})
|