完成视频复刻和优化

core/backend/apps/ai/video_replace.py

@@ -0,0 +1,308 @@
+"""视频复刻:参考视频 + 商品图/人物图 → Seedance 换商品或换角色。
+
+不新建任务类型、不接检测/抠图。提交仍走 submit_free_video,只把
+feature=video_replace 和 replace_mode 写进 payload,提示词由后端写死。
+"""
+from __future__ import annotations
+
+import uuid
+
+from django.db.models import Q
+
+from apps.assets.models import Asset, Model
+from apps.products.models import Product
+
+from .free_video import HIGH_RES_MODEL, serialize_free_video_task, submit_free_video
+from .media_probe import REF_DURATION_MAX
+
+FEATURE = "video_replace"
+REPLACE_MODES = {"product", "character"}
+MAX_IMAGES = 9
+LEGACY_PROMPT_PREFIX = "[视频复刻]"
+
+PRODUCT_PROMPT = (
+    "使用@参考视频作为镜头、节奏与口播氛围基准，"
+    "将画面中需要替换的原商品完整替换为@目标商品的外观。"
+    "保留参考视频的人物、场景、镜头运动、剪辑节奏与口播氛围，"
+    "商品外观、材质、包装必须与参考图一致，不要改变原片构图和人物表演。"
+)
+CHARACTER_PROMPT = (
+    "使用@参考视频作为镜头、节奏与口播氛围基准，"
+    "将画面中需要替换的原人物完整替换为@目标角色。"
+    "保留参考视频的商品、场景、镜头运动、剪辑节奏与口播氛围，"
+    "角色五官、发型、体态必须与参考图一致，不要改变原片构图和商品展示。"
+)
+
+
+def is_video_replace_task(task) -> bool:
+    payload = task.request_payload or {}
+    if payload.get("feature") == FEATURE:
+        return True
+    return str(payload.get("prompt") or "").startswith(LEGACY_PROMPT_PREFIX)
+
+
+def video_replace_q() -> Q:
+    return Q(request_payload__feature=FEATURE) | Q(request_payload__prompt__startswith=LEGACY_PROMPT_PREFIX)
+
+
+def serialize_video_replace_task(task, *, include_deleted_assets: bool = False) -> dict:
+    data = serialize_free_video_task(task, include_deleted_assets=include_deleted_assets)
+    payload = task.request_payload or {}
+    replace_mode = payload.get("replace_mode") or _legacy_replace_mode(payload.get("prompt") or "")
+    data.update({
+        "feature": FEATURE,
+        "replace_mode": replace_mode,
+        "subject_name": payload.get("subject_name") or "",
+        "subject_source": payload.get("subject_source") or "",
+        "product_id": payload.get("product_id") or "",
+        "model_id": payload.get("model_id") or "",
+    })
+    return data
+
+
+def submit_video_replace(*, team, user, params: dict):
+    """校验素材 → 套提示词 → 复用 free_video 提交。失败抛 ValueError。"""
+    replace_mode = str(params.get("replace_mode") or "").strip()
+    if replace_mode not in REPLACE_MODES:
+        raise ValueError("请选择替换商品或替换角色")
+
+    product_id = _optional_uuid(params.get("product_id"), "商品")
+    model_id = _optional_uuid(params.get("model_id"), "角色")
+    image_ids = _uuid_list(params.get("image_asset_ids"), "参考图")
+    has_product = product_id is not None
+    has_model = model_id is not None
+    has_temp = bool(image_ids)
+
+    if replace_mode == "product":
+        if has_model:
+            raise ValueError("商品复刻请选择商品，不要同时选择角色")
+        if has_product and has_temp:
+            raise ValueError("请从商品库选择，或临时上传商品图，不要混用")
+        if not has_product and not has_temp:
+            raise ValueError("请选择商品或上传商品参考图")
+    else:
+        if has_product:
+            raise ValueError("角色复刻请选择角色，不要同时选择商品")
+        if has_model and has_temp:
+            raise ValueError("请从人物库选择，或临时上传角色图，不要混用")
+        if not has_model and not has_temp:
+            raise ValueError("请选择角色或上传角色参考图")
+
+    video = _team_asset(team, params.get("video_asset_id"), kind=Asset.Type.VIDEO, label="参考视频")
+    video_seconds = _asset_duration_seconds(video)
+    if video_seconds > REF_DURATION_MAX:
+        raise ValueError("参考视频不能超过 15 秒，请剪短后重试")
+
+    if has_product:
+        subject_name, image_refs, subject_source = _product_library_refs(team, product_id)
+    elif has_model:
+        subject_name, image_refs, subject_source = _character_library_refs(team, model_id)
+    else:
+        noun = "商品" if replace_mode == "product" else "角色"
+        subject_name, image_refs, subject_source = _temporary_image_refs(team, image_ids, noun=noun)
+
+    prompt = PRODUCT_PROMPT if replace_mode == "product" else CHARACTER_PROMPT
+    duration = _output_duration(params.get("duration"), video_seconds)
+    references = [
+        _owned_ref(video, kind="video", role="reference_video", label="参考视频"),
+        *image_refs,
+    ]
+    return submit_free_video(
+        team=team,
+        user=user,
+        params={
+            "prompt": prompt,
+            "mode": "universal",
+            "model": str(params.get("model") or HIGH_RES_MODEL),
+            "aspect_ratio": str(params.get("aspect_ratio") or "9:16"),
+            "resolution": str(params.get("resolution") or "720p"),
+            "duration": duration,
+            "seed": params.get("seed", -1),
+            "generate_audio": True,
+            "references": references,
+            "feature": FEATURE,
+            "extra_payload": {
+                "replace_mode": replace_mode,
+                "subject_name": subject_name,
+                "subject_source": subject_source,
+                "product_id": str(product_id) if product_id else "",
+                "model_id": str(model_id) if model_id else "",
+            },
+        },
+    )
+
+
+def _legacy_replace_mode(prompt: str) -> str:
+    return "character" if prompt.startswith("[视频复刻·角色]") else "product"
+
+
+def _optional_uuid(value, label: str):
+    text = str(value or "").strip()
+    if not text:
+        return None
+    try:
+        return uuid.UUID(text)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"{label}无效") from exc
+
+
+def _uuid_list(value, label: str) -> list:
+    if value in (None, ""):
+        return []
+    if not isinstance(value, (list, tuple)):
+        raise ValueError(f"{label}格式无效")
+    if len(value) > MAX_IMAGES:
+        raise ValueError(f"{label}最多 {MAX_IMAGES} 张")
+    seen = set()
+    out = []
+    for item in value:
+        parsed = _optional_uuid(item, label)
+        if parsed is None or parsed in seen:
+            continue
+        seen.add(parsed)
+        out.append(parsed)
+    return out
+
+
+def _team_asset(team, asset_id, *, kind: str, label: str) -> Asset:
+    parsed = _optional_uuid(asset_id, label)
+    if parsed is None:
+        raise ValueError(f"请先上传{label}")
+    asset = Asset.objects.filter(id=parsed, team=team, is_deleted=False, purged_at__isnull=True).first()
+    if asset is None:
+        raise ValueError(f"{label}不存在或已被删除")
+    if asset.asset_type != kind:
+        raise ValueError(f"{label}类型不正确")
+    return asset
+
+
+def _asset_duration_seconds(asset: Asset) -> float:
+    primary = asset.files.filter(is_primary=True).first() or asset.files.first()
+    if primary is None or not primary.duration_ms:
+        return 0.0
+    return primary.duration_ms / 1000.0
+
+
+def _output_duration(requested, video_seconds: float) -> int:
+    try:
+        value = int(requested) if requested not in (None, "") else 0
+    except (TypeError, ValueError):
+        value = 0
+    if value:
+        return min(15, max(4, value))
+    if video_seconds:
+        return min(15, max(4, int(round(video_seconds))))
+    return 15
+
+
+def _owned_ref(asset: Asset, *, kind: str, role: str, label: str) -> dict:
+    from .services import _asset_preview_url
+
+    url = _asset_preview_url(asset)
+    if not url:
+        raise ValueError(f"「{label}」没有可用文件")
+    ref = {
+        "url": url,
+        "type": kind,
+        "role": role,
+        "label": label,
+        "source": "upload",
+        "asset_id": str(asset.id),
+    }
+    seconds = _asset_duration_seconds(asset)
+    if seconds:
+        ref["duration"] = seconds
+    return ref
+
+
+def _library_image_ref(asset: Asset, *, team, label: str) -> dict:
+    from .services import _asset_preview_url, _seedance_ref_url
+
+    if asset.team_id == team.id and not asset.is_deleted:
+        return {
+            "url": _asset_preview_url(asset) or "",
+            "type": "image",
+            "role": "reference_image",
+            "label": label,
+            "source": "asset",
+            "asset_id": str(asset.id),
+        }
+    raw = _asset_preview_url(asset)
+    url = _seedance_ref_url(raw, asset.review_status, asset.review_remote_id)
+    if not url:
+        raise ValueError(f"「{label}」没有可用文件")
+    return {
+        "url": url,
+        "type": "image",
+        "role": "reference_image",
+        "label": label,
+        "source": "upload",
+        "asset_id": str(asset.id),
+    }
+
+
+def _product_library_refs(team, product_id: uuid.UUID) -> tuple[str, list, str]:
+    product = (
+        Product.objects.filter(id=product_id, team=team, purged_at__isnull=True, status=Product.Status.ACTIVE)
+        .select_related("cover_asset")
+        .prefetch_related("images__asset")
+        .first()
+    )
+    if product is None:
+        raise ValueError("商品不存在或已被删除")
+    assets = []
+    seen = set()
+    for image in product.images.all():
+        asset = image.asset
+        if asset is None or asset.id in seen or asset.is_deleted:
+            continue
+        seen.add(asset.id)
+        assets.append(asset)
+        if len(assets) >= MAX_IMAGES:
+            break
+    if not assets and product.cover_asset_id and not product.cover_asset.is_deleted:
+        assets.append(product.cover_asset)
+    if not assets:
+        raise ValueError("这个商品还没有可用图片")
+    refs = [_library_image_ref(asset, team=team, label="目标商品" if index == 0 else f"目标商品{index + 1}") for index, asset in enumerate(assets)]
+    return product.title, refs, "library"
+
+
+def _character_library_refs(team, model_id: uuid.UUID) -> tuple[str, list, str]:
+    model = (
+        Model.objects.filter(Q(team=team) | Q(is_official=True), id=model_id, is_deleted=False, purged_at__isnull=True)
+        .select_related("portrait_asset", "triview_asset")
+        .first()
+    )
+    if model is None:
+        raise ValueError("角色不存在或已被删除")
+    assets = []
+    seen = set()
+    for asset in (model.portrait_asset, model.triview_asset):
+        if asset is None or asset.id in seen or asset.is_deleted:
+            continue
+        seen.add(asset.id)
+        assets.append(asset)
+        if len(assets) >= MAX_IMAGES:
+            break
+    if not assets:
+        raise ValueError("这个角色还没有可用图片")
+    labels = ["目标角色", "目标角色三视图"]
+    refs = [_library_image_ref(asset, team=team, label=labels[index] if index < len(labels) else f"目标角色{index + 1}") for index, asset in enumerate(assets)]
+    return model.name, refs, "library"
+
+
+def _temporary_image_refs(team, image_ids: list, *, noun: str) -> tuple[str, list, str]:
+    refs = []
+    for index, asset_id in enumerate(image_ids):
+        asset = _team_asset(team, asset_id, kind=Asset.Type.IMAGE, label=f"{noun}参考图")
+        label = "目标商品" if noun == "商品" else "目标角色"
+        if index > 0:
+            label = f"{label}{index + 1}"
+        refs.append(_owned_ref(asset, kind="image", role="reference_image", label=label))
+    fallback = "临时商品素材" if noun == "商品" else "临时角色素材"
+    name = Asset.objects.filter(id=image_ids[0]).values_list("name", flat=True).first() or fallback
+    subject = name.rsplit(".", 1)[0] if name else fallback
+    if len(refs) > 1:
+        subject = f"{subject}（{len(refs)}张参考图）"
+    return subject, refs, "temporary"