Files
yingqing/core/backend/apps/ai/views.py
T

974 lines
49 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import uuid
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.exceptions import ValidationError
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)
if product_id:
try:
normalized_product_id = str(uuid.UUID(product_id))
except (TypeError, ValueError, AttributeError):
return Response({"detail": "商品 ID 无效"}, status=status.HTTP_400_BAD_REQUEST)
if not Product.objects.filter(team=team, id=normalized_product_id).exists():
return Response({"detail": "商品不存在或不属于当前团队"}, status=status.HTTP_400_BAD_REQUEST)
product_id = normalized_product_id
# 对话归属:传了 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()
# 合法重跑/补图必须能在当前对话里找到原批次;归属强制读取原批次,不信任当前路由或客户端值。
original_batch_task = None
if conversation is not None and batch_id:
candidates = AITask.objects.filter(
team=team,
conversation=conversation,
is_deleted=False,
purged_at__isnull=True,
request_payload__batch_id=batch_id,
).order_by("created_at")
original_batch_task = next(
(task for task in candidates if not (task.request_payload or {}).get("batch_append")),
None,
)
if original_batch_task is not None:
product_id = str((original_batch_task.request_payload or {}).get("product_id") or "").strip() or None
else:
# 任意 UUID 不能伪装成可追加批次;降级为普通新批次。
batch_id = None
retry_of_task_id = None
elif batch_id:
batch_id = None
retry_of_task_id = None
# 普通新批次不能写入其他商品/通用范围的旧会话;竞态或旧客户端出现错配时自动开正确范围会话。
if conversation is not None:
same_mode = conversation.mode == mode
same_product = str(conversation.product_id or "") == str(product_id or "")
if not same_mode or (mode == ImageConversation.Mode.IMAGE and original_batch_task is None and not same_product):
conversation = None
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)
# 异常批次缩略图只来自该批任务自己的成功成图。批次删除时这些资产会一起软删,
# 这里批量取回并按任务映射,避免逐批查询;商品封面不再参与缩略图回退。
preview_by_task_id = {}
task_ids = [task.id for task in tasks]
generated_assets = (
Asset.objects.filter(
team=self.get_team(),
origin_task_id__in=task_ids,
asset_type=Asset.Type.IMAGE,
is_deleted=True,
purged_at__isnull=True,
)
.prefetch_related("files")
.order_by("created_at", "id")
)
for asset in generated_assets:
files = sorted(
asset.files.all(),
key=lambda item: (not item.is_primary, item.created_at, str(item.id)),
)
for file in files:
preview_url = AssetFileSerializer(file).data.get("preview_url", "")
if preview_url:
preview_by_task_id.setdefault(asset.origin_task_id, preview_url)
break
# 商品仍只用于异常批次标题文字回退,不再关联或预取商品封面资源。
products = Product.objects.filter(
team=self.get_team(), id__in=product_ids, status=Product.Status.ACTIVE, purged_at__isnull=True
).only("id", "title")
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 = next(
(
preview_by_task_id[task.id]
for task in group
if task.status == AITask.Status.SUCCEEDED and task.id in preview_by_task_id
),
"",
)
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)
if self.action == "list":
scope = self.request.query_params.get("scope", "").strip()
product_id = self.request.query_params.get("product_id", "").strip()
if scope and product_id:
raise ValidationError({"detail": "scope 与 product_id 不能同时传入"})
if scope:
if scope != "unbound":
raise ValidationError({"detail": "scope 仅支持 unbound"})
queryset = queryset.filter(product__isnull=True)
elif product_id:
try:
normalized_product_id = str(uuid.UUID(product_id))
except (TypeError, ValueError, AttributeError) as exc:
raise ValidationError({"detail": "商品 ID 无效"}) from exc
if not Product.objects.filter(team=self.get_team(), id=normalized_product_id).exists():
raise ValidationError({"detail": "商品不存在或不属于当前团队"})
queryset = queryset.filter(product_id=normalized_product_id)
return queryset
def perform_create(self, serializer):
product = serializer.validated_data.get("product")
if product is not None and product.team_id != self.get_team().id:
raise ValidationError({"product": "商品不存在或不属于当前团队"})
super().perform_create(serializer)
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 "",
"product_id": str((t.request_payload or {}).get("product_id") 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/<id>/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/<id>/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/<id>/ —— 软删(在途任务拒删,等终态)。"""
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/<id>/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/<id>/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"]