"""平台超管后台 · 跨团队端点。所有视图统一挂 IsPlatformAdmin,非超管一律 403,写操作记审计。""" import logging from decimal import Decimal, ROUND_HALF_UP 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 from rest_framework.response import Response 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.model_catalog import invalidate_model_catalog_cache from apps.ai.models import AITask, ModelConfig, ModelProvider, PromptTemplate, QualityWord from apps.assets.models import Asset from apps.assets.review import poll_asset_review, submit_asset_for_review from apps.billing.models import BillingConfig, CreditLedger, QuotaPolicy from apps.billing.pricing import get_billing_config, invalidate_billing_config_cache from apps.billing.services.ledger import adjust_credit from apps.common.pagination import DefaultPagination from apps.products.models import Product from apps.projects.models import Project from .serializers import ( COST_ANOMALY_RATIO, AdminLedgerSerializer, AdminModelConfigSerializer, AdminModelProviderSerializer, AdminProjectSerializer, AdminQuotaPolicySerializer, AdminReviewAssetSerializer, AdminTaskDetailSerializer, AdminTaskSerializer, AdminTeamMemberSerializer, AdminTeamSerializer, AdminUserSerializer, PromptTemplateSerializer, QualityWordSerializer, ) logger = logging.getLogger(__name__) # 后台「刷新状态」只向供应商拉取已提交的异步视频任务;单次上限避免请求拖死。 _ADMIN_TASK_POLL_LIMIT = 20 _ADMIN_TASK_POLL_STATUSES = (AITask.Status.SUBMITTED, AITask.Status.POLLING) # 任务监控「生成中」Tab:含已创建/已预留/已提交/轮询/后处理(与自由创作 IN_FLIGHT 对齐) _ADMIN_TASK_INFLIGHT_STATUSES = ( AITask.Status.CREATED, AITask.Status.RESERVED, AITask.Status.SUBMITTED, AITask.Status.POLLING, AITask.Status.POSTPROCESSING, ) def _team_qs(): return ( Team.objects.select_related("owner", "credit_account") .annotate(member_count_anno=Count("members", distinct=True)) ) @api_view(["GET", "POST"]) @permission_classes([IsPlatformAdmin]) def admin_invitations(request): """GET 列所有邀请码(跨团队,可按 kind/status/search 过滤,分页); POST 平台超管发「开团队码」(create_team,不绑团队,新用户凭码开新团队当 owner)。""" if request.method == "GET": qs = Invitation.objects.select_related("team", "used_by").order_by("-created_at") kind = request.query_params.get("kind") if kind in {Invitation.Kind.JOIN_TEAM, Invitation.Kind.CREATE_TEAM}: qs = qs.filter(kind=kind) st = request.query_params.get("status") if st in dict(Invitation.Status.choices): qs = qs.filter(status=st) search = (request.query_params.get("search") or "").strip() if search: qs = qs.filter(Q(code__icontains=search) | Q(team__name__icontains=search)) paginator = DefaultPagination() page = paginator.paginate_queryset(qs, request) return paginator.get_paginated_response(InvitationSerializer(page, many=True).data) email = str(request.data.get("email") or "").strip() invite = Invitation.objects.create( kind=Invitation.Kind.CREATE_TEAM, team=None, email=email, created_by=request.user, ) log_admin_action( request, "invite.issue_create_team", target_type="invitation", target_id=invite.id, target_name=invite.code, after={"kind": invite.kind, "email": email}, ) return Response(InvitationSerializer(invite).data, status=status.HTTP_201_CREATED) @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_revoke_invitation(request, invite_id): """撤销一个待用邀请码(任意团队)。已用/已撤销/已过期则原样返回(幂等)。""" invite = Invitation.objects.select_related("team").filter(id=invite_id).first() if invite is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) if invite.status == Invitation.Status.PENDING: invite.status = Invitation.Status.REVOKED invite.save(update_fields=["status", "updated_at"]) log_admin_action( request, "invite.revoke", target_type="invitation", target_id=invite.id, target_name=invite.code, ) return Response(InvitationSerializer(invite).data) # ─────────────────────────── 团队管理 ─────────────────────────── @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_teams(request): """列所有团队(跨团队,可按 status/search 过滤,分页)。""" qs = _team_qs().order_by("-created_at") st = request.query_params.get("status") if st in dict(Team.Status.choices): qs = qs.filter(status=st) search = (request.query_params.get("search") or "").strip() if search: qs = qs.filter(name__icontains=search) paginator = DefaultPagination() page = paginator.paginate_queryset(qs, request) return paginator.get_paginated_response(AdminTeamSerializer(page, many=True).data) @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_team_detail(request, team_id): team = _team_qs().filter(id=team_id).first() if team is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) members = team.members.select_related("user").order_by("created_at") data = AdminTeamSerializer(team).data data["members"] = AdminTeamMemberSerializer(members, many=True).data return Response(data) @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_team_toggle(request, team_id): """启停团队。团队停用后其成员登录会被拒(login 校验团队状态)。""" team = Team.objects.filter(id=team_id).first() if team is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) before = team.status team.status = Team.Status.ACTIVE if team.status == Team.Status.DISABLED else Team.Status.DISABLED team.save(update_fields=["status", "updated_at"]) log_admin_action( request, "team.toggle_status", target_type="team", target_id=team.id, target_name=team.name, before={"status": before}, after={"status": team.status}, ) return Response(AdminTeamSerializer(_team_qs().get(id=team.id)).data) # ─────────────────────────── 用户管理 ─────────────────────────── @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_team_pricing(request, team_id): """团队差异化调价(jimeng 同款诉求):设置团队价格系数(0.10~10.00)。 最终积分价 = 挂牌价 × 系数;全部计费类型统一生效;视频在途任务用下单快照不受影响。""" from decimal import InvalidOperation team = Team.objects.filter(id=team_id).first() if team is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) try: value = Decimal(str(request.data.get("price_multiplier"))) except (InvalidOperation, TypeError, ValueError): return Response({"price_multiplier": ["数值格式不正确"]}, status=status.HTTP_400_BAD_REQUEST) if not value.is_finite() or value < Decimal("0.10") or value > Decimal("10"): return Response({"price_multiplier": ["系数需在 0.10 ~ 10.00 之间"]}, status=status.HTTP_400_BAD_REQUEST) before = str(team.price_multiplier) # 显式 HALF_UP:quantize 默认银行家舍入(0.125→0.12),与全仓取整纪律不一致(review 确认) team.price_multiplier = value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) team.save(update_fields=["price_multiplier", "updated_at"]) log_admin_action( request, "team.pricing_update", target_type="team", target_id=team.id, target_name=team.name, before={"price_multiplier": before}, after={"price_multiplier": str(team.price_multiplier)}, ) return Response(AdminTeamSerializer(_team_qs().get(id=team.id)).data) @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_users(request): """列所有用户(跨团队,可按 status/search 过滤,分页)。""" qs = User.objects.prefetch_related("team_memberships__team").order_by("-date_joined") st = request.query_params.get("status") if st in dict(User.Status.choices): qs = qs.filter(status=st) search = (request.query_params.get("search") or "").strip() if search: qs = qs.filter(username__icontains=search) paginator = DefaultPagination() page = paginator.paginate_queryset(qs, request) return paginator.get_paginated_response(AdminUserSerializer(page, many=True).data) @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_user_toggle(request, user_id): """启停用户。停用即清 token 强制下线;不允许停用平台超管(防自锁)。""" user = User.objects.filter(id=user_id).first() if user is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) if user.is_platform_admin: return Response({"detail": "不能停用平台超管"}, status=status.HTTP_400_BAD_REQUEST) before = user.status user.status = User.Status.ACTIVE if user.status == User.Status.DISABLED else User.Status.DISABLED user.save(update_fields=["status"]) if user.status == User.Status.DISABLED: Token.objects.filter(user=user).delete() log_admin_action( request, "user.toggle_status", target_type="user", target_id=user.id, target_name=user.username, before={"status": before}, after={"status": user.status}, ) return Response(AdminUserSerializer(user).data) @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_user_reset_password(request, user_id): """平台超管强制改用户密码(改后清 token 强制重登)。""" user = User.objects.filter(id=user_id).first() if user is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) password = str(request.data.get("password") or "").strip() if len(password) < 8: return Response({"password": ["新密码至少 8 位"]}, status=status.HTTP_400_BAD_REQUEST) user.set_password(password) user.save(update_fields=["password"]) Token.objects.filter(user=user).delete() log_admin_action( request, "user.reset_password", target_type="user", target_id=user.id, target_name=user.username, ) return Response(status=status.HTTP_204_NO_CONTENT) # ─────────────────────────── 质量词(平台单层)─────────────────────────── @api_view(["GET", "POST"]) @permission_classes([IsPlatformAdmin]) def admin_quality_words(request): """GET 列全部质量词(配置量小,不分页,前端按 stage 分组);POST 新增一条。""" if request.method == "GET": qs = QualityWord.objects.all().order_by("stage", "sort", "created_at") return Response(QualityWordSerializer(qs, many=True).data) serializer = QualityWordSerializer(data=request.data) serializer.is_valid(raise_exception=True) obj = serializer.save() log_admin_action( request, "quality_word.create", target_type="quality_word", target_id=obj.id, target_name=f"{obj.stage}:{obj.text}", after={"stage": obj.stage, "text": obj.text}, ) return Response(QualityWordSerializer(obj).data, status=status.HTTP_201_CREATED) @api_view(["PATCH", "DELETE"]) @permission_classes([IsPlatformAdmin]) def admin_quality_word_detail(request, word_id): obj = QualityWord.objects.filter(id=word_id).first() if obj is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) if request.method == "DELETE": log_admin_action( request, "quality_word.delete", target_type="quality_word", target_id=obj.id, target_name=f"{obj.stage}:{obj.text}", before={"stage": obj.stage, "text": obj.text}, ) obj.delete() return Response(status=status.HTTP_204_NO_CONTENT) before_text = obj.text serializer = QualityWordSerializer(obj, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() log_admin_action( request, "quality_word.update", target_type="quality_word", target_id=obj.id, target_name=f"{obj.stage}:{obj.text}", before={"text": before_text}, after={"text": obj.text}, ) return Response(QualityWordSerializer(obj).data) # ─────────────────────────── 生图/视频 提示词模板 ─────────────────────────── @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_prompt_templates(request): """列视频线 6 条提示词模板(固定 key,数量小不分页,按 key 排序)。""" # 故事板已下线,分镜图提示词不再出现在后台可编辑列表里(历史行留在库里不动) qs = PromptTemplate.objects.exclude(key=PromptTemplate.Key.STORYBOARD_FRAME).order_by("key") return Response(PromptTemplateSerializer(qs, many=True).data) @api_view(["PATCH"]) @permission_classes([IsPlatformAdmin]) def admin_prompt_template_detail(request, template_id): """改一条提示词模板的正文 / 比例 / 启用(key 固定不可改,不增不删)。""" obj = PromptTemplate.objects.filter(id=template_id).first() if obj is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) before = {"template": obj.template, "ratio": obj.ratio, "enabled": obj.enabled} serializer = PromptTemplateSerializer(obj, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() log_admin_action( request, "prompt_template.update", target_type="prompt_template", target_id=obj.id, target_name=obj.key, before=before, after={"template": obj.template, "ratio": obj.ratio, "enabled": obj.enabled}, ) return Response(PromptTemplateSerializer(obj).data) # ─────────────────────────── 火山人像审核队列 ─────────────────────────── _REVIEW_STATUSES = {"", "processing", "active", "failed"} @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_asset_reviews(request): """跨团队送审资产审核队列(角色定妆照/三视图/分镜图)。?review_status=none|processing|active|failed 过滤(none=未送审)。""" qs = ( Asset.objects.filter(category__in=Asset.REVIEW_CATEGORIES, is_deleted=False) .select_related("team") .prefetch_related("files") .order_by("-created_at") ) rs = request.query_params.get("review_status") if rs == "none": qs = qs.filter(review_status="") elif rs in {"processing", "active", "failed"}: qs = qs.filter(review_status=rs) search = (request.query_params.get("search") or "").strip() if search: qs = qs.filter(team__name__icontains=search) paginator = DefaultPagination() page = paginator.paginate_queryset(qs, request) return paginator.get_paginated_response(AdminReviewAssetSerializer(page, many=True).data) @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_asset_reviews_submit(request): """批量送审(也用于失败重试):对给定 person 资产逐个 submit_asset_for_review。""" ids = request.data.get("asset_ids") or [] assets = list(Asset.objects.filter(id__in=ids, category__in=Asset.REVIEW_CATEGORIES, is_deleted=False)) for asset in assets: submit_asset_for_review(asset) statuses = {str(a.id): a.review_status for a in Asset.objects.filter(id__in=ids)} log_admin_action( request, "asset_review.submit", target_type="asset", target_name=f"{len(assets)} assets", after={"count": len(assets)}, ) return Response({"submitted": len(assets), "statuses": statuses}) @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_asset_reviews_poll(request): """自动送审未送审资产 + 轮询审核中状态。给 asset_ids 则只轮询这些,否则全平台兜底。""" from apps.assets.review import drain_platform_reviews ids = request.data.get("asset_ids") if ids: qs = Asset.objects.filter( category__in=Asset.REVIEW_CATEGORIES, review_status="processing", is_deleted=False, id__in=ids ) statuses = {} for asset in qs: statuses[str(asset.id)] = poll_asset_review(asset) return Response({"polled": len(statuses), "submitted": 0, "statuses": statuses}) result = drain_platform_reviews() return Response(result) def _refresh_inflight_task(task: AITask, *, operator) -> AITask: """对单条在途任务向供应商拉一次最新态。无远端 ID / 非视频异步任务则原样返回。""" if not task.provider_task_id: return task if task.task_type == AITask.Type.FREE_VIDEO: from apps.ai.free_video import finalize_free_video return finalize_free_video(task=task) if task.task_type == AITask.Type.VIDEO_SEGMENT: from apps.ai.services import poll_video_segment from apps.projects.models import VideoSegment segment_id = (task.request_payload or {}).get("video_segment_id") if not segment_id: return task segment = VideoSegment.objects.filter(id=segment_id).first() if segment is None: return task user = task.created_by or getattr(task.team, "owner", None) or operator poll_video_segment(video_segment=segment, user=user) task.refresh_from_db() return task return task # ─────────────────────────── AI 任务监控 + 成本异常 ─────────────────────────── @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_tasks(request): """全局 AITask 列表(?status= / ?task_type= / ?team= / ?anomaly=1 成本异常 筛 + 分页)。""" # 列表不返回请求/响应/完整错误正文。部分图片、视频任务的 JSON 可达数 MB，若随列表页 # 从远程 MySQL 读取，会让只有 10 行的分页请求也长时间卡在“加载中”；详情接口仍完整读取。 qs = ( AITask.objects.select_related("team", "model_config") .defer("request_payload", "response_payload", "error_message") .order_by("-created_at") ) st = request.query_params.get("status") if st in {"generating", "running", "in_flight"}: qs = qs.filter(status__in=_ADMIN_TASK_INFLIGHT_STATUSES) elif 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(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_tasks_poll(request): """向供应商刷新在途视频任务状态。给 task_ids 则只拉这些,否则拉全平台 submitted/polling。 图片/脚本等同步任务没有远端 task id,点刷新后前端会再拉一次列表,worker 已落库的终态会一并更新。 """ ids = request.data.get("task_ids") qs = ( AITask.objects.select_related("team", "model_config", "model_config__provider", "project") .filter(status__in=_ADMIN_TASK_POLL_STATUSES) .exclude(provider_task_id="") .order_by("-updated_at") ) if ids: qs = qs.filter(id__in=ids) statuses = {} for task in qs[:_ADMIN_TASK_POLL_LIMIT]: try: refreshed = _refresh_inflight_task(task, operator=request.user) statuses[str(refreshed.id)] = refreshed.status except Exception: # noqa: BLE001 — 单条失败不阻断整页刷新 logger.warning("admin poll task %s failed", task.id, exc_info=True) statuses[str(task.id)] = task.status return Response({"polled": len(statuses), "statuses": statuses}) @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_task_detail(request, task_id): # 尝试链只在详情请求加载，列表仍保持原查询与一任务一行。 task = ( AITask.objects.select_related("team", "model_config") .prefetch_related("model_attempts") .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)}) @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_task_reap(request, task_id): """手动回收僵尸任务:标失败 + 退还预留积分(与自动回收 _reap_stale_standalone_image_tasks 同款账务路径)。 自动回收只在「同团队下次提交」时顺手触发——团队从此不再用该功能,冻结积分就永远躺着; 这里给超管一个兜底出口。范围收紧到 RESERVED(worker 从未认领):SUBMITTED/POLLING 的视频 任务可能正在 ARK 生成(合法耗时 5-10 分钟+),后台强杀会与轮询结算竞态,不冒险。""" from datetime import timedelta from django.db import transaction from django.core.exceptions import ObjectDoesNotExist from django.utils import timezone from apps.billing.services.ledger import release_credit with transaction.atomic(): # 锁行后再判状态:worker 可能恰好并发认领,凭陈旧快照回收会把在途任务误杀 task = AITask.objects.select_for_update().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.RESERVED: return Response({"detail": "仅能回收卡在「已预留」状态的任务"}, status=status.HTTP_400_BAD_REQUEST) if task.updated_at >= timezone.now() - timedelta(minutes=10): return Response({"detail": "任务仍在 10 分钟活跃窗口内,可能只是在排队,稍后再试"}, status=status.HTTP_400_BAD_REQUEST) task.status = AITask.Status.FAILED task.error_message = "管理员手动回收(预留积分已退还)" task.completed_at = timezone.now() task.save(update_fields=["status", "error_message", "completed_at", "updated_at"]) try: reservation = task.credit_reservation except ObjectDoesNotExist: reservation = None if reservation is not None: release_credit(reservation=reservation, reason="管理员手动回收僵尸任务") log_admin_action( request, "task.reap", target_type="ai_task", target_id=task.id, target_name=task.task_type, ) return Response(AdminTaskSerializer(task).data) # ─────────────────────────── 计费审计 + 额度策略 ─────────────────────────── @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_ledgers(request): """全局信用流水浏览(?ledger_type= / ?team= 筛 + 分页)。""" qs = CreditLedger.objects.select_related("team", "user").order_by("-created_at") lt = request.query_params.get("ledger_type") if lt in dict(CreditLedger.Type.choices): qs = qs.filter(ledger_type=lt) team_id = request.query_params.get("team") if team_id: qs = qs.filter(team_id=team_id) paginator = DefaultPagination() page = paginator.paginate_queryset(qs, request) return paginator.get_paginated_response(AdminLedgerSerializer(page, many=True).data) @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_ledger_adjust(request): """手动调额(争议补偿):{team, amount, reason}。amount 可正可负,落 ADJUSTMENT 流水。""" from decimal import InvalidOperation team_id = request.data.get("team") or request.data.get("team_id") team = Team.objects.filter(id=team_id).first() if team is None: return Response({"detail": "团队不存在"}, status=status.HTTP_404_NOT_FOUND) try: amount = Decimal(str(request.data.get("amount"))) except (InvalidOperation, TypeError, ValueError): return Response({"amount": ["金额格式不正确"]}, status=status.HTTP_400_BAD_REQUEST) if amount == 0: return Response({"amount": ["调额金额不能为 0"]}, status=status.HTTP_400_BAD_REQUEST) reason = str(request.data.get("reason") or "").strip() try: ledger = adjust_credit(team=team, amount=amount, reason=reason, operator=request.user) except ValueError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) log_admin_action( request, "credit.adjust", target_type="team", target_id=team.id, target_name=team.name, after={"amount": str(amount), "balance_after": str(ledger.balance_after), "reason": reason}, ) return Response(AdminLedgerSerializer(ledger).data, status=status.HTTP_201_CREATED) def _billing_config_payload(cfg: BillingConfig) -> dict: return { "points_per_yuan": str(cfg.points_per_yuan), "video_margin_multiplier": str(cfg.video_margin_multiplier), "video_reserve_buffer": str(cfg.video_reserve_buffer), "updated_at": cfg.updated_at, } @api_view(["GET", "PATCH"]) @permission_classes([IsPlatformAdmin]) def admin_billing_config(request): """平台计费配置(积分汇率/视频毛利系数/预留 buffer)。PATCH 即刻生效于下一次估价/结算。""" from decimal import InvalidOperation cfg = get_billing_config() if request.method == "GET": return Response(_billing_config_payload(cfg)) before = _billing_config_payload(cfg) updates: dict = {} for field, minimum in (("points_per_yuan", Decimal("0.01")), ("video_margin_multiplier", Decimal("0.01")), ("video_reserve_buffer", Decimal("1"))): if request.data.get(field) is None: continue try: value = Decimal(str(request.data[field])) except (InvalidOperation, TypeError, ValueError): return Response({field: ["数值格式不正确"]}, status=status.HTTP_400_BAD_REQUEST) # Decimal("NaN") 构造不抛,比较才抛 InvalidOperation → 500(review 确认),显式拦 if not value.is_finite(): return Response({field: ["数值格式不正确"]}, status=status.HTTP_400_BAD_REQUEST) if value < minimum: return Response({field: [f"不能小于 {minimum}"]}, status=status.HTTP_400_BAD_REQUEST) updates[field] = value if not updates: return Response({"detail": "没有可更新的字段"}, status=status.HTTP_400_BAD_REQUEST) for field, value in updates.items(): setattr(cfg, field, value) cfg.save(update_fields=[*updates.keys(), "updated_at"]) invalidate_billing_config_cache() after = _billing_config_payload(cfg) log_admin_action(request, "billing_config.update", target_type="billing_config", target_id=cfg.id, before=before, after=after) return Response(after) @api_view(["GET", "POST"]) @permission_classes([IsPlatformAdmin]) def admin_quota_policies(request): """GET 列额度策略(?team= 筛);POST 新建。team 必填;user/project 留空=团队级。""" if request.method == "GET": qs = QuotaPolicy.objects.select_related("team").order_by("-created_at") team_id = request.query_params.get("team") if team_id: qs = qs.filter(team_id=team_id) paginator = DefaultPagination() page = paginator.paginate_queryset(qs, request) return paginator.get_paginated_response(AdminQuotaPolicySerializer(page, many=True).data) serializer = AdminQuotaPolicySerializer(data=request.data) serializer.is_valid(raise_exception=True) obj = serializer.save() log_admin_action( request, "quota_policy.create", target_type="quota_policy", target_id=obj.id, target_name=str(obj.team_id), after=serializer.data, ) return Response(AdminQuotaPolicySerializer(obj).data, status=status.HTTP_201_CREATED) @api_view(["PATCH", "DELETE"]) @permission_classes([IsPlatformAdmin]) def admin_quota_policy_detail(request, policy_id): obj = QuotaPolicy.objects.filter(id=policy_id).first() if obj is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) if request.method == "DELETE": log_admin_action(request, "quota_policy.delete", target_type="quota_policy", target_id=obj.id, target_name=str(obj.team_id)) obj.delete() return Response(status=status.HTTP_204_NO_CONTENT) serializer = AdminQuotaPolicySerializer(obj, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() log_admin_action(request, "quota_policy.update", target_type="quota_policy", target_id=obj.id, target_name=str(obj.team_id), after=serializer.data) return Response(AdminQuotaPolicySerializer(obj).data) # ─────────────────────────── 模型供应商 / 模型 ─────────────────────────── @api_view(["GET", "POST"]) @permission_classes([IsPlatformAdmin]) def admin_providers(request): if request.method == "GET": qs = ModelProvider.objects.annotate(model_count_anno=Count("models", distinct=True)).order_by("created_at") return Response(AdminModelProviderSerializer(qs, many=True).data) serializer = AdminModelProviderSerializer(data=request.data) serializer.is_valid(raise_exception=True) obj = serializer.save() log_admin_action(request, "provider.create", target_type="model_provider", target_id=obj.id, target_name=obj.name) return Response(AdminModelProviderSerializer(obj).data, status=status.HTTP_201_CREATED) @api_view(["PATCH", "DELETE"]) @permission_classes([IsPlatformAdmin]) def admin_provider_detail(request, provider_id): obj = ModelProvider.objects.filter(id=provider_id).first() if obj is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) if request.method == "DELETE": log_admin_action(request, "provider.delete", target_type="model_provider", target_id=obj.id, target_name=obj.name) obj.delete() return Response(status=status.HTTP_204_NO_CONTENT) serializer = AdminModelProviderSerializer(obj, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() log_admin_action(request, "provider.update", target_type="model_provider", target_id=obj.id, target_name=obj.name) return Response(AdminModelProviderSerializer(ModelProvider.objects.annotate(model_count_anno=Count("models", distinct=True)).get(id=obj.id)).data) @api_view(["GET", "POST"]) @permission_classes([IsPlatformAdmin]) def admin_models(request): if request.method == "GET": qs = ModelConfig.objects.select_related("provider").order_by("-is_default", "status", "capability", "provider__name", "created_at") prov = request.query_params.get("provider") if prov: qs = qs.filter(provider_id=prov) cap = request.query_params.get("capability") if cap in dict(ModelConfig.Capability.choices): qs = qs.filter(capability=cap) return Response(AdminModelConfigSerializer(qs, many=True).data) serializer = AdminModelConfigSerializer(data=request.data) serializer.is_valid(raise_exception=True) obj = serializer.save() invalidate_model_catalog_cache() log_admin_action(request, "model.create", target_type="model_config", target_id=obj.id, target_name=f"{obj.provider_id}:{obj.name}") return Response(AdminModelConfigSerializer(obj).data, status=status.HTTP_201_CREATED) @api_view(["PATCH", "DELETE"]) @permission_classes([IsPlatformAdmin]) def admin_model_detail(request, model_id): obj = ModelConfig.objects.select_related("provider").filter(id=model_id).first() if obj is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) if request.method == "DELETE": log_admin_action(request, "model.delete", target_type="model_config", target_id=obj.id, target_name=obj.name) obj.delete() invalidate_model_catalog_cache() return Response(status=status.HTTP_204_NO_CONTENT) serializer = AdminModelConfigSerializer(obj, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() invalidate_model_catalog_cache() log_admin_action(request, "model.update", target_type="model_config", target_id=obj.id, target_name=obj.name) return Response(AdminModelConfigSerializer(ModelConfig.objects.select_related("provider").get(id=obj.id)).data) @api_view(["POST"]) @permission_classes([IsPlatformAdmin]) def admin_model_set_default(request, model_id): """把某模型设为其 capability 的默认模型(同 capability 其余清默认)。""" from django.db import transaction obj = ModelConfig.objects.filter(id=model_id).first() if obj is None: return Response({"detail": "not found"}, status=status.HTTP_404_NOT_FOUND) with transaction.atomic(): ModelConfig.objects.filter(capability=obj.capability).exclude(id=obj.id).update(is_default=False) obj.is_default = True obj.save(update_fields=["is_default", "updated_at"]) invalidate_model_catalog_cache() log_admin_action(request, "model.set_default", target_type="model_config", target_id=obj.id, target_name=f"{obj.capability}:{obj.name}") return Response(AdminModelConfigSerializer(ModelConfig.objects.select_related("provider").get(id=obj.id)).data) # ─────────────────────────── 收尾治理(项目监控 + 数据完整性)─────────────────────────── @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_projects(request): """全局项目流水线监控(?status= / ?search= 筛 + 分页)。""" qs = Project.objects.select_related("team", "product").order_by("-created_at") st = request.query_params.get("status") if st in dict(Project.Status.choices): qs = qs.filter(status=st) search = (request.query_params.get("search") or "").strip() if search: qs = qs.filter(Q(name__icontains=search) | Q(team__name__icontains=search)) paginator = DefaultPagination() page = paginator.paginate_queryset(qs, request) return paginator.get_paginated_response(AdminProjectSerializer(page, many=True).data) @api_view(["GET"]) @permission_classes([IsPlatformAdmin]) def admin_integrity(request): """数据完整性 / 治理体检(只读):统计潜在孤儿 / 异常记录数,供平台超管巡检。""" from django.utils import timezone now = timezone.now() checks = [ {"key": "teams_without_account", "label": "无信用账户的团队", "count": Team.objects.filter(credit_account__isnull=True).count()}, {"key": "products_without_image", "label": "无图商品(无主图且无图册)", "count": Product.objects.filter(cover_asset__isnull=True, images__isnull=True).distinct().count()}, {"key": "expired_pending_invites", "label": "已过期但未标记的邀请码", "count": Invitation.objects.filter(status=Invitation.Status.PENDING, expires_at__lt=now).count()}, {"key": "assets_without_files", "label": "无文件的资产(未删除)", "count": Asset.objects.filter(is_deleted=False, files__isnull=True).distinct().count()}, {"key": "failed_projects", "label": "失败的项目", "count": Project.objects.filter(status=Project.Status.FAILED).count()}, {"key": "stale_reserved_tasks", "label": "卡在 reserved 的任务", "count": AITask.objects.filter(status=AITask.Status.RESERVED).count()}, ] return Response({"checks": checks, "totals": {"teams": Team.objects.count(), "users": User.objects.count(), "projects": Project.objects.count(), "products": Product.objects.count()}})