feat(admin): Phase 6 AI任务监控+成本异常 — 全局任务列/筛/详情抽屉/失败重投
后端:adminpanel tasks(全局 AITask + status/type/team/anomaly 筛分页)+ detail(payload)+ retry(失败重投, 仅图像类 best-effort 否则 400);is_cost_anomaly(实际>预估×1.5)+ F() 异常筛;IsPlatformAdmin + 审计。 前端:adminApi tasks 系列;Admin 任务监控页(状态 tab + 仅成本异常 chip + 表格成本预估→实际+异常标 + 详情抽屉 payload + 失败重投)。 修真 bug:抽屉缺 .show 类停屏外,补上正常滑入。 测试:adminpanel 36 单测过(筛/异常/详情/重投 mock celery + 非失败/不支持/权限拒); 无头 e2e _admin-p6.mjs 6 断言过 + 0 console error(不点真重投);tsc+build 绿。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
seaislee1209
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
5aab8a568f
commit
ca1e50d32c
@@ -1,9 +1,20 @@
|
||||
from decimal import Decimal
|
||||
|
||||
from rest_framework import serializers
|
||||
|
||||
from apps.accounts.models import Team, TeamMember, User
|
||||
from apps.ai.models import QualityWord
|
||||
from apps.ai.models import AITask, QualityWord
|
||||
from apps.assets.models import Asset
|
||||
|
||||
# 成本异常阈值:实际成本 > 预估 × 此倍数(且预估 > 0)即标异常
|
||||
COST_ANOMALY_RATIO = Decimal("1.5")
|
||||
|
||||
|
||||
def is_cost_anomaly(estimated, actual) -> bool:
|
||||
est = estimated or Decimal("0")
|
||||
act = actual or Decimal("0")
|
||||
return bool(est > 0 and act > est * COST_ANOMALY_RATIO)
|
||||
|
||||
|
||||
class QualityWordSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
@@ -73,3 +84,29 @@ class AdminUserSerializer(serializers.ModelSerializer):
|
||||
{"team_id": str(m.team_id), "team_name": m.team.name, "role": m.role}
|
||||
for m in obj.team_memberships.select_related("team").all()
|
||||
]
|
||||
|
||||
|
||||
class AdminTaskSerializer(serializers.ModelSerializer):
|
||||
team_name = serializers.CharField(source="team.name", read_only=True, default=None)
|
||||
model_name = serializers.CharField(source="model_config.name", read_only=True, default=None)
|
||||
cost_anomaly = serializers.SerializerMethodField()
|
||||
|
||||
class Meta:
|
||||
model = AITask
|
||||
fields = [
|
||||
"id", "task_type", "status", "team", "team_name", "model_name",
|
||||
"estimated_cost", "actual_cost", "cost_anomaly", "error_code", "created_at",
|
||||
]
|
||||
read_only_fields = fields
|
||||
|
||||
def get_cost_anomaly(self, obj) -> bool:
|
||||
return is_cost_anomaly(obj.estimated_cost, obj.actual_cost)
|
||||
|
||||
|
||||
class AdminTaskDetailSerializer(AdminTaskSerializer):
|
||||
class Meta(AdminTaskSerializer.Meta):
|
||||
fields = AdminTaskSerializer.Meta.fields + [
|
||||
"project", "idempotency_key", "request_payload", "response_payload",
|
||||
"error_message", "submitted_at", "completed_at",
|
||||
]
|
||||
read_only_fields = fields
|
||||
|
||||
@@ -298,3 +298,72 @@ class AdminAssetReviewTests(TestCase):
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertEqual(r.data["polled"], 1) # 仅 a_proc 处于 processing
|
||||
mock_poll.assert_called_once()
|
||||
|
||||
|
||||
class AdminTaskMonitorTests(TestCase):
|
||||
"""Phase 6:AI 任务监控(列/筛/详情/成本异常)+ 失败重投(celery 全 mock)+ 权限。"""
|
||||
|
||||
def setUp(self):
|
||||
from apps.ai.models import AITask, ModelConfig, ModelProvider
|
||||
|
||||
self.AITask = AITask
|
||||
self.admin = User.objects.create_user(username="padmin6", password="x", is_platform_admin=True)
|
||||
self.normal = User.objects.create_user(username="normal6", password="x")
|
||||
self.team = Team.objects.create(name="T6", owner=self.normal)
|
||||
TeamMember.objects.create(team=self.team, user=self.normal, role=TeamMember.Role.OWNER)
|
||||
prov = ModelProvider.objects.create(name="prov6", display_name="P6")
|
||||
self.mc = ModelConfig.objects.create(provider=prov, name="m6", display_name="M6", capability=ModelConfig.Capability.IMAGE)
|
||||
|
||||
def mk(tt, st, est, act, key):
|
||||
return AITask.objects.create(
|
||||
team=self.team, model_config=self.mc, task_type=tt, status=st,
|
||||
estimated_cost=est, actual_cost=act, idempotency_key=key,
|
||||
)
|
||||
|
||||
self.t_ok = mk(AITask.Type.PRODUCT_IMAGE, AITask.Status.SUCCEEDED, "1.0", "1.0", "k-ok")
|
||||
self.t_failed = mk(AITask.Type.PRODUCT_IMAGE, AITask.Status.FAILED, "1.0", "1.0", "k-failed")
|
||||
self.t_anom = mk(AITask.Type.PERSON_IMAGE, AITask.Status.SUCCEEDED, "1.0", "5.0", "k-anom")
|
||||
self.t_script = mk(AITask.Type.SCRIPT_GENERATION, AITask.Status.FAILED, "1.0", "1.0", "k-script")
|
||||
self.ac = APIClient()
|
||||
self.ac.force_authenticate(self.admin)
|
||||
self.nc = APIClient()
|
||||
self.nc.force_authenticate(self.normal)
|
||||
|
||||
def test_list_permission_and_count(self):
|
||||
self.assertEqual(self.nc.get("/api/admin/tasks/").status_code, 403)
|
||||
r = self.ac.get("/api/admin/tasks/")
|
||||
self.assertEqual(r.status_code, 200)
|
||||
self.assertGreaterEqual(r.data["count"], 4)
|
||||
|
||||
def test_filter_status_type_anomaly(self):
|
||||
self.assertTrue(all(t["status"] == "failed" for t in self.ac.get("/api/admin/tasks/?status=failed").data["results"]))
|
||||
self.assertTrue(all(t["task_type"] == "person_image" for t in self.ac.get("/api/admin/tasks/?task_type=person_image").data["results"]))
|
||||
ids = {t["id"] for t in self.ac.get("/api/admin/tasks/?anomaly=1").data["results"]}
|
||||
self.assertIn(str(self.t_anom.id), ids)
|
||||
self.assertNotIn(str(self.t_ok.id), ids)
|
||||
|
||||
def test_cost_anomaly_flag(self):
|
||||
self.assertTrue(self.ac.get(f"/api/admin/tasks/{self.t_anom.id}/").data["cost_anomaly"])
|
||||
self.assertFalse(self.ac.get(f"/api/admin/tasks/{self.t_ok.id}/").data["cost_anomaly"])
|
||||
|
||||
def test_detail_has_payloads(self):
|
||||
d = self.ac.get(f"/api/admin/tasks/{self.t_failed.id}/")
|
||||
self.assertEqual(d.status_code, 200)
|
||||
self.assertIn("request_payload", d.data)
|
||||
self.assertIn("response_payload", d.data)
|
||||
|
||||
def test_retry_dispatch_and_audit(self):
|
||||
with patch("apps.ai.tasks.generate_standalone_image_task.delay") as mock_delay:
|
||||
r = self.ac.post(f"/api/admin/tasks/{self.t_failed.id}/retry/")
|
||||
self.assertEqual(r.status_code, 200)
|
||||
mock_delay.assert_called_once()
|
||||
self.assertTrue(AdminAuditLog.objects.filter(action="task.retry").exists())
|
||||
|
||||
def test_retry_non_failed_rejected(self):
|
||||
self.assertEqual(self.ac.post(f"/api/admin/tasks/{self.t_ok.id}/retry/").status_code, 400)
|
||||
|
||||
def test_retry_unsupported_type_rejected(self):
|
||||
self.assertEqual(self.ac.post(f"/api/admin/tasks/{self.t_script.id}/retry/").status_code, 400)
|
||||
|
||||
def test_retry_requires_admin(self):
|
||||
self.assertEqual(self.nc.post(f"/api/admin/tasks/{self.t_failed.id}/retry/").status_code, 403)
|
||||
|
||||
@@ -5,6 +5,9 @@ from .views import (
|
||||
admin_asset_reviews_poll,
|
||||
admin_asset_reviews_submit,
|
||||
admin_invitations,
|
||||
admin_task_detail,
|
||||
admin_task_retry,
|
||||
admin_tasks,
|
||||
admin_quality_word_detail,
|
||||
admin_quality_words,
|
||||
admin_revoke_invitation,
|
||||
@@ -31,4 +34,7 @@ urlpatterns = [
|
||||
path("asset-reviews/", admin_asset_reviews, name="admin-asset-reviews"),
|
||||
path("asset-reviews/submit/", admin_asset_reviews_submit, name="admin-asset-reviews-submit"),
|
||||
path("asset-reviews/poll/", admin_asset_reviews_poll, name="admin-asset-reviews-poll"),
|
||||
path("tasks/", admin_tasks, name="admin-tasks"),
|
||||
path("tasks/<uuid:task_id>/", admin_task_detail, name="admin-task-detail"),
|
||||
path("tasks/<uuid:task_id>/retry/", admin_task_retry, name="admin-task-retry"),
|
||||
]
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
"""平台超管后台 · 跨团队端点。所有视图统一挂 IsPlatformAdmin,非超管一律 403,写操作记审计。"""
|
||||
|
||||
from django.db.models import Count, Q
|
||||
from decimal import Decimal
|
||||
|
||||
from django.db.models import Count, F, Q
|
||||
from rest_framework import status
|
||||
from rest_framework.authtoken.models import Token
|
||||
from rest_framework.decorators import api_view, permission_classes
|
||||
@@ -10,13 +12,16 @@ from apps.accounts.audit import log_admin_action
|
||||
from apps.accounts.models import Invitation, Team, User
|
||||
from apps.accounts.permissions import IsPlatformAdmin
|
||||
from apps.accounts.serializers import InvitationSerializer
|
||||
from apps.ai.models import QualityWord
|
||||
from apps.ai.models import AITask, QualityWord
|
||||
from apps.assets.models import Asset
|
||||
from apps.assets.review import poll_asset_review, submit_asset_for_review
|
||||
from apps.common.pagination import DefaultPagination
|
||||
|
||||
from .serializers import (
|
||||
COST_ANOMALY_RATIO,
|
||||
AdminReviewAssetSerializer,
|
||||
AdminTaskDetailSerializer,
|
||||
AdminTaskSerializer,
|
||||
AdminTeamMemberSerializer,
|
||||
AdminTeamSerializer,
|
||||
AdminUserSerializer,
|
||||
@@ -326,3 +331,71 @@ def admin_asset_reviews_poll(request):
|
||||
for asset in qs:
|
||||
statuses[str(asset.id)] = poll_asset_review(asset)
|
||||
return Response({"polled": len(statuses), "statuses": statuses})
|
||||
|
||||
|
||||
# ─────────────────────────── AI 任务监控 + 成本异常 ───────────────────────────
|
||||
|
||||
|
||||
@api_view(["GET"])
|
||||
@permission_classes([IsPlatformAdmin])
|
||||
def admin_tasks(request):
|
||||
"""全局 AITask 列表(?status= / ?task_type= / ?team= / ?anomaly=1 成本异常 筛 + 分页)。"""
|
||||
qs = AITask.objects.select_related("team", "model_config").order_by("-created_at")
|
||||
st = request.query_params.get("status")
|
||||
if st in dict(AITask.Status.choices):
|
||||
qs = qs.filter(status=st)
|
||||
tt = request.query_params.get("task_type")
|
||||
if tt in dict(AITask.Type.choices):
|
||||
qs = qs.filter(task_type=tt)
|
||||
team_id = request.query_params.get("team")
|
||||
if team_id:
|
||||
qs = qs.filter(team_id=team_id)
|
||||
if request.query_params.get("anomaly") in {"1", "true"}:
|
||||
qs = qs.filter(estimated_cost__gt=0, actual_cost__gt=F("estimated_cost") * COST_ANOMALY_RATIO)
|
||||
paginator = DefaultPagination()
|
||||
page = paginator.paginate_queryset(qs, request)
|
||||
return paginator.get_paginated_response(AdminTaskSerializer(page, many=True).data)
|
||||
|
||||
|
||||
@api_view(["GET"])
|
||||
@permission_classes([IsPlatformAdmin])
|
||||
def admin_task_detail(request, task_id):
|
||||
task = AITask.objects.select_related("team", "model_config").filter(id=task_id).first()
|
||||
if task is None:
|
||||
return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND)
|
||||
return Response(AdminTaskDetailSerializer(task).data)
|
||||
|
||||
|
||||
@api_view(["POST"])
|
||||
@permission_classes([IsPlatformAdmin])
|
||||
def admin_task_retry(request, task_id):
|
||||
"""失败任务重投(best-effort):仅 FAILED + 可重投的图像类(基础资产 / 独立生图)。
|
||||
其余类型(脚本 / 故事板 / 视频 / 配音 / 导出)请到对应页面重跑,这里 400 不冒险误投。"""
|
||||
task = AITask.objects.filter(id=task_id).first()
|
||||
if task is None:
|
||||
return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND)
|
||||
if task.status != AITask.Status.FAILED:
|
||||
return Response({"detail": "仅失败任务可重投"}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
image_types = {AITask.Type.PRODUCT_IMAGE, AITask.Type.PERSON_IMAGE, AITask.Type.SCENE_IMAGE}
|
||||
if task.task_type not in image_types:
|
||||
return Response({"detail": "该任务类型暂不支持后台重投,请到对应页面重新生成"}, status=status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
from apps.ai.tasks import generate_base_asset_task, generate_standalone_image_task
|
||||
|
||||
try:
|
||||
if task.project_id:
|
||||
generate_base_asset_task.delay(str(task.id))
|
||||
else:
|
||||
generate_standalone_image_task.delay(str(task.id))
|
||||
except Exception: # noqa: BLE001 — 投递失败如实返回,不静默
|
||||
return Response({"detail": "重投调度失败,请确认 worker 在线"}, status=status.HTTP_502_BAD_GATEWAY)
|
||||
|
||||
log_admin_action(
|
||||
request,
|
||||
"task.retry",
|
||||
target_type="ai_task",
|
||||
target_id=task.id,
|
||||
target_name=task.task_type,
|
||||
)
|
||||
return Response({"retried": True, "task_id": str(task.id)})
|
||||
|
||||
Reference in New Issue
Block a user