Files
DEMO-AGENT/apps/audit/views.py

71 lines
2.5 KiB
Python

from django.shortcuts import get_object_or_404, render
from .models import AgentAuditLog, NotificationRecord
from apps.chat.models import Conversation
from .services import build_detail_summary, build_history_rows
def log_list(request):
# 处理历史页支持按批次、产品和状态筛选。
scenario_id = (request.GET.get("scenario_id") or "").strip()
keyword = (request.GET.get("keyword") or "").strip()
notify_status = (request.GET.get("notify_status") or "").strip()
risk_status = (request.GET.get("risk_status") or "").strip()
logs = AgentAuditLog.objects.all()
if scenario_id:
logs = logs.filter(scenario_id=scenario_id)
if keyword:
logs = logs.filter(product_name__icontains=keyword) | logs.filter(batch_id__icontains=keyword)
if notify_status:
matched_pairs = list(
NotificationRecord.objects.filter(message_status=notify_status).values_list(
"batch_id",
"conversation_id",
)
)
logs = [
log
for log in logs
if (log.batch_id, log.conversation_id) in matched_pairs
]
if risk_status:
logs = [
log
for log in logs
if (log.structured_output or {}).get("highest_risk_level") == risk_status
or (log.structured_output or {}).get("risk_level") == risk_status
]
return render(
request,
"audit/log_list.html",
{
"history_rows": build_history_rows(logs),
"selected_scenario_id": scenario_id,
"keyword": keyword,
"notify_status": notify_status,
"risk_status": risk_status,
},
)
def log_detail(request, log_id: int):
# 详情页只负责按主键加载审计快照并渲染;
# 所有脱敏和字段映射都应在服务层完成。
audit_log = get_object_or_404(AgentAuditLog, pk=log_id)
notifications = NotificationRecord.objects.filter(
conversation_id=audit_log.conversation_id,
batch_id=audit_log.batch_id,
)
conversation = Conversation.objects.filter(conversation_id=audit_log.conversation_id).first()
detail_summary = build_detail_summary(audit_log, conversation, notifications)
return render(
request,
"audit/log_detail.html",
{
"log": audit_log,
"notifications": notifications,
"conversation": conversation,
"detail_summary": detail_summary,
},
)