from django.db import transaction from django.db.models import Count, Exists, OuterRef, Q from django.utils import timezone from rest_framework import status from rest_framework.decorators import action from rest_framework.parsers import FormParser, MultiPartParser from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet from apps.assets.models import Asset from apps.assets.serializers import AssetFileSerializer, AssetSerializer from apps.common.api import TeamScopedViewSetMixin, get_current_team from apps.common.celery_health import require_worker from apps.products.models import Product from .generation_errors import classify_generation_error, public_error_for_task from .models import AITask, ImageConversation, ModelConfig from .serializers import AITaskSerializer, ImageConversationSerializer, ModelConfigSerializer from .services import enqueue_standalone_images class GenerateImageView(APIView): """独立生图(不绑项目)· 图片创作/模特图/平台套图共用 —— **异步**。 POST /api/ai/generate-image/ 提交生成,秒级返回 RESERVED 任务列表(慢出图交给 worker)。 GET /api/ai/generate-image/?ids=… 轮询这些任务的状态;成功的任务带回成图 asset。 """ def post(self, request): require_worker() # 异步出图依赖 worker 兜底执行,没 worker 直接拒绝(否则任务永远 RESERVED) prompt = str(request.data.get("prompt") or "").strip() if not prompt: return Response({"detail": "prompt 不能为空"}, status=status.HTTP_400_BAD_REQUEST) mode = str(request.data.get("mode") or "image") try: count = int(request.data.get("count") or 1) except (TypeError, ValueError): count = 1 product_id = str(request.data.get("product_id") or "").strip() or None reference_product = bool(request.data.get("reference_product")) model_id = str(request.data.get("model_id") or "").strip() or None model_entity_id = str(request.data.get("model_entity_id") or "").strip() or None ratio = str(request.data.get("ratio") or "").strip() or None image_model = str(request.data.get("image_model") or "").strip() or None # 平台套图:前端传规范化平台 id(taobao/douyin/…),用于后端注入平台版式块(优化版) platform_id = str(request.data.get("platform_id") or "").strip() or None conversation_id = str(request.data.get("conversation_id") or "").strip() or None # 重跑/补图:前端带原批次 batch_id → enqueue 沿用(UUID 校验),记录归回原批次不裂新卡 batch_id = str(request.data.get("batch_id") or "").strip() or None retry_of_task_id = str(request.data.get("retry_of_task_id") or "").strip() or None # 用户在图片创作里上传的参考图(已先传成 Asset),按 id 列表带入 → 生成时作多图参考(image_edit) raw_refs = request.data.get("reference_image_ids") or [] if isinstance(raw_refs, str): raw_refs = [s for s in raw_refs.split(",") if s.strip()] reference_image_ids = [str(r).strip() for r in raw_refs if str(r).strip()] team = get_current_team(request.user) # 对话归属:传了 id 就用现成对话(限本团队);没传则自动开一条新对话,标题取 prompt 前 24 字。 conversation = None if conversation_id: conversation = ImageConversation.objects.filter( team=team, id=conversation_id, is_deleted=False, purged_at__isnull=True ).first() if conversation is None: conversation = ImageConversation.objects.create( team=team, created_by=request.user, mode=mode if mode in dict(ImageConversation.Mode.choices) else ImageConversation.Mode.IMAGE, product_id=product_id, title=(prompt[:24] or "默认创作"), ) try: tasks = enqueue_standalone_images(team=team, user=request.user, prompt=prompt, mode=mode, count=count, product_id=product_id, reference_product=reference_product, model_id=model_id, model_entity_id=model_entity_id, ratio=ratio, image_model=image_model, conversation=conversation, reference_image_ids=reference_image_ids, platform_id=platform_id, batch_id=batch_id, retry_of_task_id=retry_of_task_id) except ValueError as exc: # 无可用模型 / 余额不足等,立即反馈 internal_kind = "user_credit_insufficient" if str(exc).strip().lower() == "insufficient credit" else "" public_error = classify_generation_error( exc, operation="image_generate", internal_kind=internal_kind ) return Response( {"detail": public_error.fallback_message, "error": public_error.as_dict()}, status=status.HTTP_400_BAD_REQUEST, ) # 本次提交即刷新对话活跃时间,左栏「最近」据此置顶 ImageConversation.objects.filter(id=conversation.id).update(last_active_at=timezone.now()) # batch_id 回传给前端存进批次卡:后续「重跑这张 / 重跑整批」带它回来即可归回原批次 return Response( { "conversation_id": str(conversation.id), "batch_id": (tasks[0].request_payload or {}).get("batch_id") if tasks else None, "tasks": [{"id": str(t.id), "status": t.status} for t in tasks], }, status=status.HTTP_202_ACCEPTED, ) def get(self, request): team = get_current_team(request.user) ids = [s for s in str(request.query_params.get("ids") or "").split(",") if s] if not ids: return Response({"tasks": []}) tasks = AITask.objects.filter(team=team, id__in=ids).prefetch_related( "generated_assets", "generated_assets__files" ) data = [] for task in tasks: public_error = public_error_for_task(task) data.append({ "id": str(task.id), "status": task.status, "error": public_error.as_dict() if public_error else None, "error_message": public_error.fallback_message if public_error else "", "assets": AssetSerializer( [a for a in task.generated_assets.all() if not a.is_deleted and a.purged_at is None], many=True ).data, }) return Response({"tasks": data}) class AITaskViewSet(TeamScopedViewSetMixin, ReadOnlyModelViewSet): # 序列化器不含 request_payload/response_payload(单条可达 3MB+ base64 图),defer 掉: # 否则只为序列化 14 个小字段也会把几十 MB blob 从库里拉回(远程库实测 40 条要 30s+)。 # 默认按创建时间倒序:任务中心 = 历史流水,最新的(多为成功)排最前。 # 缺省排序时 MySQL 按主键(UUID)乱序返回,会把一批旧失败记录顶到首页, # 前端只取首页 → 误判「全部失败」。order_by 保证稳定且新任务优先。 queryset = AITask.objects.select_related("team", "project", "model_config", "model_config__provider").defer("request_payload", "response_payload").order_by("-created_at") serializer_class = AITaskSerializer search_fields = ["idempotency_key", "provider_task_id", "project__name"] ordering_fields = ["created_at", "updated_at", "completed_at"] def get_queryset(self): # 可选 ?task_type=a,b,c 过滤:生图工作室的任务中心只想看生图任务(模特上身图/平台套图/ # 图片创作 = person_image / product_image),不掺脚本/实体抽取/故事板等流水线内部任务。 queryset = super().get_queryset().filter(is_deleted=False, purged_at__isnull=True) # 从 request_payload(已 defer)里只抽 batch_id / mode 两个 JSON 标量供前端「按批分组 + 标签」用: # KeyTextTransform 在 SQL 层 JSON_EXTRACT,不会把几 MB 的 payload 整列拉回(避开 payload 性能坑)。 from django.db.models.fields.json import KeyTextTransform queryset = queryset.annotate( rp_batch_id=KeyTextTransform("batch_id", "request_payload"), rp_mode=KeyTextTransform("mode", "request_payload"), rp_product_id=KeyTextTransform("product_id", "request_payload"), ) raw = self.request.query_params.get("task_type", "").strip() if raw: types = [t.strip() for t in raw.split(",") if t.strip()] if types: queryset = queryset.filter(task_type__in=types) return queryset # YYX#row22:只有「工作台图片生成」(mode∈model/cover/image)才计入未读; # 脚本/实体抽取/故事板等流水线内部任务不进这块未读统计。 _GEN_MODES = ("model", "cover", "image") _WORKBENCH_IMAGE_TYPES = (AITask.Type.PERSON_IMAGE, AITask.Type.PRODUCT_IMAGE) _DELETABLE_IMAGE_BATCH_STATUSES = (AITask.Status.SUCCEEDED, AITask.Status.FAILED, AITask.Status.CANCELLED) def _image_batch_tasks(self, pk, *, deleted: bool): """Resolve one standalone image batch from a user-visible task anchor. The browser only supplies an anchor task id. Team, mode and batch membership are always derived server-side so one team cannot delete or restore another team's task group. """ base = AITask.objects.filter( team=self.get_team(), project__isnull=True, task_type__in=self._WORKBENCH_IMAGE_TYPES, is_deleted=deleted, purged_at__isnull=True, ) anchor = base.filter(pk=pk).first() if anchor is None: return None, [] payload = anchor.request_payload or {} mode = str(payload.get("mode") or "").strip() if mode not in self._GEN_MODES: return None, [] batch_id = str(payload.get("batch_id") or "").strip() batch_qs = base.filter(request_payload__mode=mode) if batch_id: batch_qs = batch_qs.filter(request_payload__batch_id=batch_id) else: batch_qs = batch_qs.filter(pk=anchor.pk) return anchor, list(batch_qs.order_by("created_at").select_for_update()) @staticmethod def _is_exception_batch(tasks): return any(task.status in (AITask.Status.FAILED, AITask.Status.CANCELLED) for task in tasks) @action(detail=True, methods=["delete"], url_path="workbench-batch") def delete_workbench_batch(self, request, pk=None): """Delete a completed image batch. Fully successful batches preserve the existing asset-trash behavior. A batch containing a failed/cancelled task is soft-deleted as one recoverable exception batch instead. """ with transaction.atomic(): anchor, tasks = self._image_batch_tasks(pk, deleted=False) if anchor is None or not tasks: return Response({"detail": "图片批次不存在或不可删除"}, status=status.HTTP_404_NOT_FOUND) if any(task.status not in self._DELETABLE_IMAGE_BATCH_STATUSES for task in tasks): return Response({"detail": "批次仍在处理中，暂不能删除"}, status=status.HTTP_409_CONFLICT) now = timezone.now() task_ids = [task.id for task in tasks] Asset.objects.filter(origin_task_id__in=task_ids, purged_at__isnull=True).update(is_deleted=True, updated_at=now) if self._is_exception_batch(tasks): AITask.objects.filter(id__in=task_ids).update(is_deleted=True, updated_at=now) return Response({"storage": "exception_batch", "deleted_task_count": len(task_ids)}) return Response({"storage": "asset", "deleted_task_count": 0}) @action(detail=True, methods=["post"], url_path="restore-workbench-batch") def restore_workbench_batch(self, request, pk=None): """Restore one previously deleted exception batch, including its generated images.""" with transaction.atomic(): anchor, tasks = self._image_batch_tasks(pk, deleted=True) if anchor is None or not tasks: return Response({"detail": "图片异常批次不存在或无法恢复"}, status=status.HTTP_404_NOT_FOUND) now = timezone.now() task_ids = [task.id for task in tasks] AITask.objects.filter(id__in=task_ids).update(is_deleted=False, purged_at=None, updated_at=now) Asset.objects.filter(origin_task_id__in=task_ids, purged_at__isnull=True).update(is_deleted=False, updated_at=now) return Response({"restored_task_count": len(task_ids)}) @action(detail=True, methods=["delete"], url_path="purge-workbench-batch") def purge_workbench_batch(self, request, pk=None): """Permanently hide one deleted exception batch and all of its generated images.""" with transaction.atomic(): anchor, tasks = self._image_batch_tasks(pk, deleted=True) if anchor is None or not tasks: return Response({"detail": "图片异常批次不存在或已彻底删除"}, status=status.HTTP_404_NOT_FOUND) now = timezone.now() task_ids = [task.id for task in tasks] AITask.objects.filter(id__in=task_ids).update(is_deleted=True, purged_at=now, updated_at=now) Asset.objects.filter(origin_task_id__in=task_ids, purged_at__isnull=True).update(is_deleted=True, purged_at=now, updated_at=now) return Response(status=status.HTTP_204_NO_CONTENT) @action(detail=False, methods=["get"], url_path="workbench-exception-batches-trash") def workbench_exception_batches_trash(self, request): """List recoverable failed/cancelled image batches for the existing global trash page.""" tasks = list( AITask.objects.filter( team=self.get_team(), project__isnull=True, task_type__in=self._WORKBENCH_IMAGE_TYPES, is_deleted=True, purged_at__isnull=True, request_payload__mode__in=self._GEN_MODES, ).filter(Q(conversation__isnull=True) | Q(conversation__is_deleted=False)).order_by("created_at") ) groups = {} product_ids = set() for task in tasks: payload = task.request_payload or {} key = str(payload.get("batch_id") or task.id) groups.setdefault(key, []).append(task) product_id = str(payload.get("product_id") or "").strip() if product_id: product_ids.add(product_id) products = Product.objects.filter( team=self.get_team(), id__in=product_ids, status=Product.Status.ACTIVE, purged_at__isnull=True ).select_related("cover_asset").prefetch_related("cover_asset__files") product_by_id = {str(product.id): product for product in products} rows = [] for key, group in groups.items(): if not self._is_exception_batch(group): continue first = group[0] payload = first.request_payload or {} product = product_by_id.get(str(payload.get("product_id") or "")) cover = "" if product and product.cover_asset and not product.cover_asset.is_deleted and product.cover_asset.purged_at is None: files = list(product.cover_asset.files.all()) primary = next((item for item in files if item.is_primary), files[0] if files else None) if primary: cover = AssetFileSerializer(primary).data.get("preview_url", "") intended = sum(1 for task in group if not (task.request_payload or {}).get("batch_append")) rows.append({ "id": str(first.id), "batch_id": str(payload.get("batch_id") or ""), "mode": str(payload.get("mode") or "image"), "prompt": str(payload.get("prompt") or ""), "count": max(1, intended), "product_id": str(payload.get("product_id") or ""), "product_title": product.title if product else "", "cover_preview_url": cover, "updated_at": max(task.updated_at for task in group), }) rows.sort(key=lambda row: row["updated_at"], reverse=True) return Response({"results": rows}) def _unread_base(self): """本团队、属于图片生成、且未读(read_at is null)的任务集合(已 annotate rp_mode/rp_product_id)。 过滤必须走 request_payload__mode 路径查找而非 rp_mode 注解:KeyTextTransform 注解上的 exact/in 查找在 MySQL 会把裸字符串塞进 JSON_EXTRACT 当文档解析 → 3141 全查询炸 (sqlite 编译路径不同,单测测不出来);路径查找 RHS 会被正确 JSON 编码。""" return self.get_queryset().filter(request_payload__mode__in=self._GEN_MODES, read_at__isnull=True) @action(detail=False, methods=["get"], url_path="unread") def unread(self, request): """未读生成任务汇总:总数(导航胶囊) + 按商品分组(商品预览角标)。""" rows = self._unread_base().values_list("rp_product_id", flat=True) total = 0 by_product: dict[str, int] = {} for pid in rows: total += 1 if pid: by_product[str(pid)] = by_product.get(str(pid), 0) + 1 return Response({"total": total, "by_product": by_product}) @action(detail=False, methods=["get"], url_path="workbench") def workbench(self, request): """R100:工作台(模特上身图/平台套图/图片创作)的生成记录改从后端持久数据拉取 —— 原先只存前端 localStorage(1 小时过期、换浏览器/清缓存即空),任务中心有记录而工作台丢。 与任务中心同源 = 同一批 AITask;按 request_payload.mode(白名单)+ product 过滤, 附带每个任务的存活成图资产,前端按 batch_id 归批还原批次流。 性能:仍只用 KeyTextTransform 在 SQL 层抽 payload 标量、defer 两个巨型 payload 列,不整列拖出。""" from django.db.models.fields.json import KeyTextTransform mode = str(request.query_params.get("mode") or "").strip() if mode not in self._GEN_MODES: return Response({"detail": "mode 仅支持 model / cover / image"}, status=status.HTTP_400_BAD_REQUEST) try: limit = min(500, max(1, int(request.query_params.get("limit") or 200))) except (TypeError, ValueError): limit = 200 qs = ( AITask.objects.filter( team=self.get_team(), project__isnull=True, # 工作台独立生图不挂项目;排除流水线内部任务 task_type__in=[AITask.Type.PERSON_IMAGE, AITask.Type.PRODUCT_IMAGE], is_deleted=False, purged_at__isnull=True, ) .defer("request_payload", "response_payload") .annotate( rp_mode=KeyTextTransform("mode", "request_payload"), rp_batch_id=KeyTextTransform("batch_id", "request_payload"), rp_product_id=KeyTextTransform("product_id", "request_payload"), rp_prompt=KeyTextTransform("prompt", "request_payload"), rp_ratio=KeyTextTransform("ratio", "request_payload"), rp_platform_id=KeyTextTransform("platform_id", "request_payload"), rp_model_id=KeyTextTransform("model_id", "request_payload"), rp_model_entity_id=KeyTextTransform("model_entity_id", "request_payload"), # 只在重跑任务里落此键(值恒为 True);键不存在 → NULL → 假值,存在 → "true"/"1" → 真值 rp_batch_append=KeyTextTransform("batch_append", "request_payload"), ) .filter(request_payload__mode=mode) # 路径查找,不能用 rp_mode 注解比较(MySQL 3141,见 _unread_base) ) product_id = str(request.query_params.get("product") or "").strip() if product_id: qs = qs.filter(request_payload__product_id=product_id) tasks = list(qs.order_by("-created_at").prefetch_related("generated_assets", "generated_assets__files")[:limit]) tasks.reverse() # 旧 → 新,与对话流/工作台批次流的时间序一致 data = [] for t in tasks: public_error = public_error_for_task(t) data.append({ "id": str(t.id), "status": t.status, "error": public_error.as_dict() if public_error else None, "error_message": public_error.fallback_message if public_error else "", "prompt": t.rp_prompt or "", "batch_id": t.rp_batch_id or "", "ratio": t.rp_ratio or "", "product_id": t.rp_product_id or "", "model_id": t.rp_model_id or "", "model_entity_id": t.rp_model_entity_id or "", "platform_id": t.rp_platform_id or "", "rerun": bool(t.rp_batch_append), "retry_of_task_id": str((t.request_payload or {}).get("retry_of_task_id") or ""), "created_at": t.created_at, # 软删的图不再出现在工作台记录里(R109:删除资产库图片 → 任务记录联动) "assets": AssetSerializer( [a for a in t.generated_assets.all() if not a.is_deleted and a.purged_at is None], many=True ).data, }) return Response({"tasks": data}) @action(detail=False, methods=["post"], url_path="mark-read") def mark_read(self, request): """标记已读 → 清零未读胶囊。 body 可选:product_id(只清该商品) / batch_id(只清该批) / ids(指定任务); 都不传 = 把当前团队所有未读生成任务标记已读(进任务中心时调用)。""" qs = self._unread_base() product_id = str(request.data.get("product_id") or "").strip() batch_id = str(request.data.get("batch_id") or "").strip() ids = request.data.get("ids") or [] if isinstance(ids, str): ids = [s for s in ids.split(",") if s.strip()] if product_id: qs = qs.filter(request_payload__product_id=product_id) if batch_id: qs = qs.filter(request_payload__batch_id=batch_id) if ids: qs = qs.filter(id__in=[str(i).strip() for i in ids if str(i).strip()]) updated = qs.update(read_at=timezone.now()) return Response({"updated": updated}) class ImageConversationViewSet(TeamScopedViewSetMixin, ModelViewSet): """图片创作工作室的「对话」CRUD。 list 按 ?mode= 过滤、排除软删、按 last_active_at 倒序(左栏「最近」); create 开新对话;partial_update 重命名;destroy 软删(不连带删图)。 detail action `tasks` 返回该对话下的生图任务 + 成图 asset,供切换对话时回填批次流。 """ serializer_class = ImageConversationSerializer queryset = ImageConversation.objects.select_related("product").order_by("-last_active_at") def get_queryset(self): queryset = super().get_queryset().annotate(_task_count=Count("tasks")) if self.action in ("trash", "restore", "purge"): queryset = queryset.filter(is_deleted=True, purged_at__isnull=True) else: queryset = queryset.filter(is_deleted=False, purged_at__isnull=True) mode = self.request.query_params.get("mode", "").strip() if mode: queryset = queryset.filter(mode=mode) return queryset def perform_destroy(self, instance): # 只联动该会话生成任务产出的 Asset,不碰用户上传的参考素材。 with transaction.atomic(): Asset.objects.filter(team=instance.team, origin_task__conversation=instance, purged_at__isnull=True).update(is_deleted=True) AITask.objects.filter(team=instance.team, conversation=instance, purged_at__isnull=True).update(is_deleted=True) instance.is_deleted = True instance.save(update_fields=["is_deleted", "updated_at"]) @action(detail=False, methods=["get"], url_path="trash") def trash(self, request): """垃圾桶:列出本团队已软删且未彻底隐藏的自由创作图片会话。""" qs = self.filter_queryset(self.get_queryset()) page = self.paginate_queryset(qs) if page is not None: return self.get_paginated_response(self.get_serializer(page, many=True).data) return Response(self.get_serializer(qs, many=True).data) @action(detail=True, methods=["post"], url_path="restore") def restore(self, request, pk=None): """从垃圾桶恢复自由创作图片会话。""" conversation = self.get_object() with transaction.atomic(): conversation.is_deleted = False conversation.purged_at = None conversation.save(update_fields=["is_deleted", "purged_at", "updated_at"]) AITask.objects.filter(team=conversation.team, conversation=conversation, purged_at__isnull=True).update(is_deleted=False) Asset.objects.filter(team=conversation.team, origin_task__conversation=conversation, purged_at__isnull=True).update(is_deleted=False) return Response(self.get_serializer(conversation).data, status=status.HTTP_200_OK) @action(detail=True, methods=["delete"], url_path="purge") def purge(self, request, pk=None): """彻底删除 = 二级软删除:从垃圾桶隐藏,DB 记录保留。""" conversation = self.get_object() now = timezone.now() with transaction.atomic(): conversation.purged_at = now conversation.save(update_fields=["purged_at", "updated_at"]) AITask.objects.filter(team=conversation.team, conversation=conversation, purged_at__isnull=True).update( is_deleted=True, purged_at=now ) Asset.objects.filter(team=conversation.team, origin_task__conversation=conversation, purged_at__isnull=True).update( is_deleted=True, purged_at=now ) return Response(status=status.HTTP_204_NO_CONTENT) @action(detail=True, methods=["get"]) def tasks(self, request, pk=None): from apps.assets.models import Asset from apps.assets.serializers import _asset_preview conversation = self.get_object() tasks = ( AITask.objects.filter(conversation=conversation, is_deleted=False, purged_at__isnull=True) .prefetch_related("generated_assets", "generated_assets__files") .order_by("created_at") ) # 参考图 id → {name,url}:跨任务可能重复,缓存一次解析,供切换/刷新后批次头回显「参考了哪些图」 ref_cache: dict[str, dict] = {} def resolve_refs(ids): out = [] for rid in ids or []: rid = str(rid) if rid not in ref_cache: a = Asset.objects.filter(id=rid, is_deleted=False, purged_at__isnull=True).prefetch_related("files").first() # 带上 id:前端重跑时凭它原样复用参考图(否则刷新恢复的批次只有 {name,url},重跑丢参考) ref_cache[rid] = {"id": rid, "name": a.name, "url": _asset_preview(a)} if a else None if ref_cache[rid]: out.append(ref_cache[rid]) return out data = [] for t in tasks: public_error = public_error_for_task(t) data.append({ "id": str(t.id), "status": t.status, "error": public_error.as_dict() if public_error else None, "error_message": public_error.fallback_message if public_error else "", "prompt": (t.request_payload or {}).get("prompt", ""), "batch_id": (t.request_payload or {}).get("batch_id", ""), "ratio": (t.request_payload or {}).get("ratio") or "", # 重跑/补图任务:不计入批次「应出张数」,前端据此正确渲染失败格数量 "rerun": bool((t.request_payload or {}).get("batch_append")), "retry_of_task_id": str((t.request_payload or {}).get("retry_of_task_id") or ""), "reference_images": resolve_refs((t.request_payload or {}).get("reference_image_ids")), "created_at": t.created_at, "assets": AssetSerializer( [a for a in t.generated_assets.all() if not a.is_deleted and a.purged_at is None], many=True ).data, }) return Response({"conversation_id": str(conversation.id), "tasks": data}) def _free_video_task_queryset(team): return ( AITask.objects.filter(team=team, task_type=AITask.Type.FREE_VIDEO, is_deleted=False, purged_at__isnull=True) .select_related("model_config") .prefetch_related("generated_assets", "generated_assets__files") ) def _free_video_list_queryset(team): """正常任务流隐藏已从资产库删除的成品,但保留生成中/失败及无落库资产的历史任务。""" video_assets = Asset.objects.filter(origin_task_id=OuterRef("pk"), asset_type=Asset.Type.VIDEO) active_video_assets = video_assets.filter(is_deleted=False, purged_at__isnull=True) return ( _free_video_task_queryset(team) .annotate( _has_video_asset=Exists(video_assets), _has_active_video_asset=Exists(active_video_assets), ) .exclude( status=AITask.Status.SUCCEEDED, _has_video_asset=True, _has_active_video_asset=False, ) ) def _free_video_trash_queryset(team): return ( AITask.objects.filter(team=team, task_type=AITask.Type.FREE_VIDEO, is_deleted=True, purged_at__isnull=True) .select_related("model_config") .prefetch_related("generated_assets", "generated_assets__files") ) def _set_free_video_generated_assets_deleted(task, deleted): Asset.objects.filter(team=task.team, origin_task=task, purged_at__isnull=True).update(is_deleted=deleted) class FreeVideoView(APIView): """自由创作·视频生成(不绑项目,universal 全能参考 / keyframe 首尾帧)。 POST /api/ai/free-video/ 提交任务,秒回(火山 create 同步调、轮询交给 worker 兜底 + 前端主动 poll) GET /api/ai/free-video/ 任务流分页(offset/page_size,新→旧) """ def post(self, request): require_worker() # 生成闸:无 worker 时任务提交火山后无人兜底轮询(额度冻结、结果丢失) from .free_video import serialize_free_video_task, submit_free_video team = get_current_team(request.user) try: task = submit_free_video(team=team, user=request.user, params=request.data or {}) except ValueError as exc: message = str(exc) internal_kind = ( "user_credit_insufficient" if "余额不足" in message else "model_unavailable" if "模型未配置" in message else "provider_rate_limited" if "任务进行中" in message else "invalid_input" ) public_error = classify_generation_error( exc, operation="video_generate", internal_kind=internal_kind ) return Response( {"detail": public_error.fallback_message, "error": public_error.as_dict()}, status=status.HTTP_400_BAD_REQUEST, ) # 重取带 prefetch 的实例,序列化统一走同一条路 task = _free_video_task_queryset(team).get(id=task.id) return Response({"task": serialize_free_video_task(task)}, status=status.HTTP_202_ACCEPTED) def get(self, request): from .free_video import serialize_free_video_task team = get_current_team(request.user) try: offset = max(0, int(request.query_params.get("offset") or 0)) except (TypeError, ValueError): offset = 0 try: page_size = min(50, max(1, int(request.query_params.get("page_size") or 20))) except (TypeError, ValueError): page_size = 20 qs = _free_video_list_queryset(team).order_by("-created_at") total = qs.count() tasks = list(qs[offset : offset + page_size]) return Response( { "results": [serialize_free_video_task(t) for t in tasks], "total": total, "has_more": offset + page_size < total, } ) class FreeVideoPollView(APIView): """POST /api/ai/free-video//poll/ —— web 进程内单次轮询+终态化(幂等)。 前端渐进轮询打这里;本地无 worker 也能全程收尾(与 pipeline poll-video-segment 同模式)。""" def post(self, request, task_id): from .free_video import finalize_free_video, serialize_free_video_task team = get_current_team(request.user) task = _free_video_task_queryset(team).filter(id=task_id).first() if task is None: return Response({"detail": "任务不存在"}, status=status.HTTP_404_NOT_FOUND) if task.status in (AITask.Status.SUBMITTED, AITask.Status.POLLING): try: task = finalize_free_video(task=task) except Exception: # noqa: BLE001 — 单次轮询失败(网络抖动)不终结任务,返回现状继续轮 import logging logging.getLogger(__name__).warning("free video poll failed for %s", task_id, exc_info=True) # 终态后重取(finalize 里可能新建了资产) task = _free_video_task_queryset(team).get(id=task.id) return Response({"task": serialize_free_video_task(task)}) class FreeVideoFavoriteView(APIView): """POST /api/ai/free-video//favorite/ —— 收藏开关。""" def post(self, request, task_id): team = get_current_team(request.user) task = AITask.objects.filter( team=team, task_type=AITask.Type.FREE_VIDEO, id=task_id, is_deleted=False, purged_at__isnull=True ).first() if task is None: return Response({"detail": "任务不存在"}, status=status.HTTP_404_NOT_FOUND) task.is_favorited = not task.is_favorited task.save(update_fields=["is_favorited", "updated_at"]) return Response({"is_favorited": task.is_favorited}) class FreeVideoDetailView(APIView): """DELETE /api/ai/free-video// —— 软删(在途任务拒删,等终态)。""" def delete(self, request, task_id): team = get_current_team(request.user) task = AITask.objects.filter( team=team, task_type=AITask.Type.FREE_VIDEO, id=task_id, is_deleted=False, purged_at__isnull=True ).first() if task is None: return Response({"detail": "任务不存在"}, status=status.HTTP_404_NOT_FOUND) if task.status in (AITask.Status.SUBMITTED, AITask.Status.POLLING, AITask.Status.POSTPROCESSING): return Response({"detail": "任务生成中，请等待完成后再删除"}, status=status.HTTP_400_BAD_REQUEST) with transaction.atomic(): _set_free_video_generated_assets_deleted(task, True) task.is_deleted = True task.save(update_fields=["is_deleted", "updated_at"]) return Response(status=status.HTTP_204_NO_CONTENT) class FreeVideoTrashView(APIView): """GET /api/ai/free-video/trash/ —— 自由创作视频垃圾桶。""" def get(self, request): from .free_video import serialize_free_video_task team = get_current_team(request.user) try: offset = max(0, int(request.query_params.get("offset") or 0)) except (TypeError, ValueError): offset = 0 try: page_size = min(50, max(1, int(request.query_params.get("page_size") or 20))) except (TypeError, ValueError): page_size = 20 qs = _free_video_trash_queryset(team).order_by("-updated_at") total = qs.count() tasks = list(qs[offset : offset + page_size]) return Response( { "results": [serialize_free_video_task(t, include_deleted_assets=True) for t in tasks], "total": total, "has_more": offset + page_size < total, } ) class FreeVideoRestoreView(APIView): """POST /api/ai/free-video//restore/ —— 从垃圾桶恢复自由创作视频。""" def post(self, request, task_id): from .free_video import serialize_free_video_task team = get_current_team(request.user) task = _free_video_trash_queryset(team).filter(id=task_id).first() if task is None: return Response({"detail": "任务不存在"}, status=status.HTTP_404_NOT_FOUND) with transaction.atomic(): task.is_deleted = False task.purged_at = None task.save(update_fields=["is_deleted", "purged_at", "updated_at"]) _set_free_video_generated_assets_deleted(task, False) task = _free_video_task_queryset(team).get(id=task.id) return Response({"task": serialize_free_video_task(task)}, status=status.HTTP_200_OK) class FreeVideoPurgeView(APIView): """DELETE /api/ai/free-video//purge/ —— 二级软删除,从垃圾桶隐藏。""" def delete(self, request, task_id): team = get_current_team(request.user) task = _free_video_trash_queryset(team).filter(id=task_id).first() if task is None: return Response({"detail": "任务不存在"}, status=status.HTTP_404_NOT_FOUND) now = timezone.now() with transaction.atomic(): task.purged_at = now task.save(update_fields=["purged_at", "updated_at"]) Asset.objects.filter(team=task.team, origin_task=task, purged_at__isnull=True).update(is_deleted=True, purged_at=now) return Response(status=status.HTTP_204_NO_CONTENT) # 上传参考素材的格式/尺寸限制(与 jimeng inputBar 校验对齐;后端兜底,前端也拦) _FREE_REF_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp"} _FREE_REF_VIDEO_TYPES = {"video/mp4", "video/quicktime"} _FREE_REF_AUDIO_TYPES = {"audio/mpeg", "audio/wav", "audio/x-wav", "audio/wave"} _FREE_REF_IMAGE_MAX = 30 * 1024 * 1024 _FREE_REF_VIDEO_MAX = 50 * 1024 * 1024 _FREE_REF_AUDIO_MAX = 15 * 1024 * 1024 class FreeVideoUploadView(APIView): """POST /api/ai/free-video/upload/ —— 参考素材上传(图/视频/音频)。 校验(图 300-6000px、比例(0.4,2.5)、≤30MB;视频 mp4/mov ≤50MB、2-15s;音频 mp3/wav ≤15MB、2-15s) → TOS → Asset(source=UPLOAD, in_library=False) → {asset_id,url,type,duration,thumb_url}。 视频顺带 ffmpeg 抽首帧缩略图。""" parser_classes = [MultiPartParser, FormParser] def post(self, request): import tempfile import uuid as _uuid from io import BytesIO from pathlib import Path from apps.assets.models import Asset, AssetFile from apps.assets.storage import TosStorage from .media_probe import extract_video_poster, probe_duration upload = request.FILES.get("file") if upload is None: return Response({"detail": "缺少文件"}, status=status.HTTP_400_BAD_REQUEST) team = get_current_team(request.user) content_type = (upload.content_type or "").lower() size = upload.size or 0 if content_type in _FREE_REF_IMAGE_TYPES: kind, asset_type, suffix = "image", Asset.Type.IMAGE, {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"}[content_type] if size > _FREE_REF_IMAGE_MAX: return Response({"detail": "图片大小不能超过 30MB"}, status=status.HTTP_400_BAD_REQUEST) elif content_type in _FREE_REF_VIDEO_TYPES: kind, asset_type, suffix = "video", Asset.Type.VIDEO, ".mp4" if content_type == "video/mp4" else ".mov" if size > _FREE_REF_VIDEO_MAX: return Response({"detail": "视频大小不能超过 50MB"}, status=status.HTTP_400_BAD_REQUEST) elif content_type in _FREE_REF_AUDIO_TYPES: kind, asset_type, suffix = "audio", Asset.Type.AUDIO, ".mp3" if content_type == "audio/mpeg" else ".wav" if size > _FREE_REF_AUDIO_MAX: return Response({"detail": "音频大小不能超过 15MB"}, status=status.HTTP_400_BAD_REQUEST) else: return Response( {"detail": "不支持的文件格式（图片 JPG/PNG/WebP，视频 MP4/MOV，音频 MP3/WAV）"}, status=status.HTTP_400_BAD_REQUEST, ) raw = upload.read() width = height = None duration = None poster_bytes = None if kind == "image": try: from PIL import Image with Image.open(BytesIO(raw)) as im: width, height = im.size except Exception: # noqa: BLE001 return Response({"detail": "图片解析失败，请更换文件"}, status=status.HTTP_400_BAD_REQUEST) if not (300 <= width <= 6000 and 300 <= height <= 6000): return Response({"detail": "图片边长需在 300-6000 像素之间"}, status=status.HTTP_400_BAD_REQUEST) ratio = width / height if not (0.4 <= ratio <= 2.5): return Response({"detail": "图片宽高比需在 0.4-2.5 之间"}, status=status.HTTP_400_BAD_REQUEST) else: with tempfile.TemporaryDirectory(prefix="airshelf-fc-upload-") as tmp: tmp_path = Path(tmp) / f"in{suffix}" tmp_path.write_bytes(raw) duration = probe_duration(str(tmp_path)) if duration is None: return Response({"detail": "媒体文件解析失败，请更换文件"}, status=status.HTTP_400_BAD_REQUEST) if not (2 <= duration <= 15): label = "视频" if kind == "video" else "音频" return Response({"detail": f"{label}时长需在 2-15 秒之间"}, status=status.HTTP_400_BAD_REQUEST) if kind == "video": poster_bytes = extract_video_poster(str(tmp_path)) asset_id = _uuid.uuid4() storage = TosStorage() object_key = f"teams/{team.id}/free-create/uploads/{asset_id}{suffix}" stored = storage.upload_fileobj(fileobj=BytesIO(raw), object_key=object_key, content_type=content_type) name = (upload.name or f"素材{suffix}")[:255] asset = Asset.objects.create( id=asset_id, team=team, created_by=request.user, name=name, asset_type=asset_type, source=Asset.Source.UPLOAD, category=Asset.Category.UPLOAD, in_library=False, # 仅作生成参考,不进资产库列表 metadata={"feature": "free_video_reference"}, ) AssetFile.objects.create( asset=asset, object_key=stored.object_key, bucket=stored.bucket, content_type=stored.content_type, size_bytes=stored.size_bytes, width=width, height=height, duration_ms=int(duration * 1000) if duration else None, is_primary=True, ) url = storage.public_url(object_key=stored.object_key) thumb_url = "" if poster_bytes: poster_key = f"teams/{team.id}/free-create/uploads/{asset_id}-poster.jpg" poster_stored = storage.upload_fileobj( fileobj=BytesIO(poster_bytes), object_key=poster_key, content_type="image/jpeg" ) AssetFile.objects.create( asset=asset, object_key=poster_stored.object_key, bucket=poster_stored.bucket, content_type=poster_stored.content_type, size_bytes=poster_stored.size_bytes, is_primary=False, ) thumb_url = storage.public_url(object_key=poster_key) return Response( { "asset_id": str(asset.id), "url": url, "type": kind, "name": name, "duration": duration, "width": width, "height": height, "thumb_url": thumb_url or (url if kind == "image" else ""), }, status=status.HTTP_201_CREATED, ) class ModelConfigViewSet(ReadOnlyModelViewSet): # 按创建序固定排序:最早创建的 active 模型排第一 = 前端选择器默认项,与 get_default_model 口径一致 # (否则 DB 默认序不稳定,可能默认选到 Gemini 等;用户要默认 = 豆包 2.0 Pro,它最早创建) queryset = ModelConfig.objects.select_related("provider").filter(status=ModelConfig.Status.ACTIVE).order_by("created_at") serializer_class = ModelConfigSerializer search_fields = ["name", "display_name", "capability"] ordering_fields = ["created_at", "display_name"]