"""审计日志查询 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 正序（真实操作顺序） ], })