"""视频复刻:参考视频提炼分镜稿 → Seedance 2.5 按稿出片。

流程很短:
1. 用户上传的参考视频只走「提炼提示词」同一套接口,拆成中文分镜稿;
2. 把这份稿(格式与提炼页一模一样)加上换商品/换角色,交给 Seedance 2.5;
3. 只带商品图/角色图。用户原片不传火山,避免素材审核报错。

商品图/人物图过审后用 asset://;审核失败不扣费。
"""
from __future__ import annotations

import json
import logging
import re
import uuid
from decimal import Decimal
from pathlib import Path

from django.conf import settings
from django.db import transaction
from django.db.models import Q
from django.utils import timezone

from apps.assets.models import Asset, Model
from apps.billing.models import CreditAccount
from apps.billing.pricing import quote_video_estimate, video_reserve_amount
from apps.products.models import Product

from .free_video import (
    FREE_VIDEO_MODELS,
    HIGH_RES_MODEL,
    REPLACE_MODEL,
    IN_FLIGHT_STATUSES,
    RATIOS,
    RESOLUTIONS,
    _reap_stale_free_video_tasks,
    model_duration_range,
    serialize_free_video_task,
    start_pending_free_video,
    submit_free_video,
)
from .models import AITask, ModelConfig

logger = logging.getLogger(__name__)

FEATURE = "video_replace"
REPLACE_MODES = {"product", "character"}
MAX_IMAGES = 9
LEGACY_PROMPT_PREFIX = "[视频复刻]"
# 商品三视图在商品库是 standalone Asset(metadata.view=three_view + product_id),
# 和项目内的 BaseAssetGroup 是两套存储,这里直接取商品库那份。
TRIVIEW_LABEL = "目标商品三视图"
# 复刻走 Seedance 2.5(单次最长 30 秒)。半秒容差同 media_probe(30 秒成片常被探成 30.02)。
REPLACE_REF_DURATION_MAX = 30.5
REVIEW_UNAVAILABLE = "素材审核服务暂不可用，请稍后重试"
REVIEW_FAILED = "参考素材未通过真人合规审核，请更换视频或图片后重试"
REVIEW_SUBMIT_FAILED = "素材提交审核失败，请稍后重试"


class VideoReplaceInProgress(ValueError):
    """本团队已有复刻在跑。视图转 409,并把在跑的那条带回去让前端接上。"""

    def __init__(self, task):
        super().__init__("已有一个视频正在复刻中，完成或取消后才能再提交")
        self.task = task


def get_inflight_video_replace(team):
    """本团队正在跑的复刻任务(含审核中 / 提炼中)。没有返回 None。"""
    _reap_stale_free_video_tasks(team=team)
    return (
        AITask.objects.filter(
            team=team,
            task_type=AITask.Type.FREE_VIDEO,
            status__in=IN_FLIGHT_STATUSES,
            is_deleted=False,
            purged_at__isnull=True,
        )
        .filter(video_replace_q())
        .order_by("-created_at")
        .first()
    )

PRODUCT_PROMPT = (
    "按下面的商品语义重构分镜生成视频。@目标商品只用于锁定外观、材质与包装。"
    "严格遵守每镜的新商品动作、场景与口播；不要恢复参考片原商品、原动作、原卖点或原台词。"
)
CHARACTER_PROMPT = (
    "以@参考视频的镜头、人物动作、场景、节奏、口播和剪辑为唯一基准生成视频。"
    "将画面中所有可见的原人物替换为@目标角色；五官、发型、体态和服装以角色参考图为准。"
    "保留原商品、场景、镜头运动、动作顺序和声音节奏，不添加字幕、花字或额外人物。"
)

# 拆解期间的占位提示词。真正的提示词在 worker 拆完后回写。
DIGEST_PENDING_PROMPT = "正在提炼参考视频的分镜稿…"

# 商品库里有三视图时才追加(没有就不提,免得模型去找一张不存在的图)
PRODUCT_TRIVIEW_NOTE = "商品各面以@目标商品三视图为准。"

# 只在用户主动把成片调得比参考片短很多时才出现。时长一致时绝不出现 ——
# 它和「节奏一模一样」是互相打架的。
PRODUCT_CONDENSE_NOTE = (
    "参考片约 {source} 秒，本次成片 {output} 秒：整镜省略次要镜头，不要快放、不要把两镜压成一镜。"
)

SHOT_HEADER = "【镜头"
DIGEST_TAIL_SECTION = "【拆解存疑】"
# 整片设定里值得留的(顺序即输出顺序)
KEEP_HEADER_FIELDS = ("画面比例", "整体风格", "主线")
# 喂给 Seedance 的镜头原文栏位。字幕/音效/BGM/备注会诱使模型画花字、加戏,一律丢掉。
SHOT_COPY_FIELDS = ("时间", "时长", "景别", "机位", "运镜", "画面", "人物动作", "人物表情", "台词/旁白")
SHOT_INLINE_FIELDS = ("时间", "时长", "景别", "机位", "运镜")
SHOT_DETAIL_FIELDS = (("画面", "画面"), ("人物动作", "动作"), ("人物表情", "表情"), ("台词/旁白", "口播"))
_EMPTY_VALUES = {"", "无", "不可见", "看不出", "看不清", "听不清"}
# 模型会写「不可见。」「无。」这种带句读的空值,比对前先剥掉尾部标点
_TRAILING_PUNCT = "。.,，、;；:：!！?？ "


def _is_empty_value(value: str) -> bool:
    return value.strip().rstrip(_TRAILING_PUNCT) in _EMPTY_VALUES


def _parse_digest(digest_text: str) -> tuple[dict, list[dict]]:
    """分镜稿纯文本 → (整片设定, 镜头列表)。解析不出镜头返回 ([], []),由调用方回落原文。"""
    body = (digest_text or "").strip()
    head, sep, _tail = body.partition(DIGEST_TAIL_SECTION)
    if sep:
        body = head.strip()

    header: dict[str, str] = {}
    shots: list[dict] = []
    current: dict | None = None
    for raw in body.split("\n"):
        line = raw.strip()
        if not line:
            continue
        if line.startswith(SHOT_HEADER):
            current = {}
            shots.append(current)
            continue
        key, sep_char, value = line.partition("：")
        if not sep_char:
            continue
        key, value = key.strip(), value.strip()
        if current is None:
            header[key] = value
        else:
            current[key] = value
    return header, [shot for shot in shots if shot]


def _render_shot_table(header: dict, shots: list[dict]) -> str:
    """镜头表:每镜一行摘要(时间/景别/机位/运镜)+ 画面、动作、表情、口播四行细节。
    空值(无 / 不可见)整行省掉 —— 反复出现的「无」只会稀释真正的指令。"""
    lines: list[str] = ["【整片设定】"]
    for field in KEEP_HEADER_FIELDS:
        value = (header.get(field) or "").strip()
        if not _is_empty_value(value):
            lines.append(f"{field}：{value}")
    lines.append(f"镜头数：{len(shots)} 镜（下面逐镜照拍，不要增删、不要改顺序）")

    lines.append("")
    lines.append("【逐镜还原】")
    for index, shot in enumerate(shots, start=1):
        inline = [shot.get(f, "").strip() for f in SHOT_INLINE_FIELDS]
        inline = [v for v in inline if not _is_empty_value(v)]
        lines.append(f"镜头 {index:02d} | " + " | ".join(inline) if inline else f"镜头 {index:02d}")
        for field, label in SHOT_DETAIL_FIELDS:
            value = (shot.get(field) or "").strip()
            if not _is_empty_value(value):
                lines.append(f"  {label}：{value}")
    return "\n".join(lines)


# —— 逐镜生成 ——
# 一镜一次 Seedance,每段只还原原片对应时间,并把 @参考视频 带上锁切镜。
MIN_SHOT_SECONDS = 4          # 火山单次出片下限
MAX_CLIP_SECONDS = 15         # Seedance 2.0 单次上限;单镜再长也一次出,避免凭空切一刀
MAX_SHOTS = 12                # 再多就是积分黑洞,也超出用户预期
_CLOCK_RE = re.compile(r"(\d{1,2}):(\d{2})")


def _raw_shot_seconds(shot: dict) -> int:
    """分镜稿里这一镜多长,不做上下限裁切。优先「时长」,其次用「时间」区间算。"""
    raw = (shot.get("时长") or "").strip()
    digits = re.sub(r"[^0-9]", "", raw)
    if digits:
        return max(1, int(digits))
    stamps = _CLOCK_RE.findall(shot.get("时间") or "")
    if len(stamps) >= 2:
        start = int(stamps[0][0]) * 60 + int(stamps[0][1])
        end = int(stamps[1][0]) * 60 + int(stamps[1][1])
        if end > start:
            return end - start
    return 5


def _clip_seconds(raw: int) -> int:
    return max(MIN_SHOT_SECONDS, min(MAX_CLIP_SECONDS, int(raw or 0) or MIN_SHOT_SECONDS))


def _pack_shot_groups(shots: list[dict]) -> list[list[dict]]:
    """不足 4 秒的相邻镜合并成一次出片,避免火山拒单,也避免把 2 秒镜硬拉成 4 秒。"""
    groups: list[list[dict]] = []
    current: list[dict] = []
    current_sec = 0
    for shot in shots[:MAX_SHOTS]:
        sec = _raw_shot_seconds(shot)
        if current and current_sec < MIN_SHOT_SECONDS:
            current.append(shot)
            current_sec += sec
            continue
        if current:
            groups.append(current)
        current = [shot]
        current_sec = sec
    if current:
        if groups and current_sec < MIN_SHOT_SECONDS:
            groups[-1].extend(current)
        else:
            groups.append(current)
    return groups


def _trim_groups_to_output(groups: list[list[dict]], output_seconds: int) -> list[list[dict]]:
    """用户把成片要得比参考片短很多时,整组省略尾镜,不要压缩画面。"""
    if not output_seconds or not groups:
        return groups
    total = sum(_clip_seconds(sum(_raw_shot_seconds(s) for s in group)) for group in groups)
    if total - output_seconds < 3:
        return groups
    kept: list[list[dict]] = []
    acc = 0
    for group in groups:
        if acc >= output_seconds:
            break
        kept.append(group)
        acc += _clip_seconds(sum(_raw_shot_seconds(s) for s in group))
    return kept or groups[:1]


def _verbatim_shot_block(shot: dict) -> str:
    """提炼稿这一镜的原文栏位,不改写、不换标签名。"""
    lines = []
    for field in SHOT_COPY_FIELDS:
        value = (shot.get(field) or "").strip()
        if not _is_empty_value(value):
            lines.append(f"{field}：{value}")
    return "\n".join(lines)


def _shot_clock(shot: dict) -> str:
    clock = (shot.get("时间") or "").strip()
    return clock if clock and not _is_empty_value(clock) else ""


def build_shot_plan(
    digest_text: str,
    subject_name: str,
    *,
    has_triview: bool = False,
    replace_mode: str = "product",
    output_seconds: int = 0,
) -> list[dict]:
    """分镜稿 → 逐镜生成计划。解析不出镜头返回 [],调用方回落单次生成。"""
    _header, shots = _parse_digest(digest_text)
    if not shots:
        return []
    groups = _trim_groups_to_output(_pack_shot_groups(shots), output_seconds)
    total = len(groups)
    plan = []
    for index, group in enumerate(groups, start=1):
        seconds = _clip_seconds(sum(_raw_shot_seconds(shot) for shot in group))
        plan.append({
            "index": index,
            "seconds": seconds,
            "prompt": build_shot_prompt(
                group, index=index, total=total, seconds=seconds,
                subject_name=subject_name, has_triview=has_triview, replace_mode=replace_mode,
            ),
        })
    return plan


def build_shot_prompt(
    shots, *, index: int, total: int, seconds: int, setting: str = "",
    subject_name: str, has_triview: bool = False, replace_mode: str = "product",
) -> str:
    """2.0 短指令 + 这一镜的提炼原文。@参考视频 才是切镜来源,不要扩写。"""
    from .services import enforce_no_embedded_captions

    group = shots if isinstance(shots, list) else [shots]
    subject = (subject_name or "").strip() or ("目标角色" if replace_mode == "character" else "目标商品")
    is_character = replace_mode == "character"

    if len(group) == 1:
        clock = _shot_clock(group[0])
        span = f"，对应原片 {clock}" if clock else ""
        scope = f"只生成下面这一镜（第 {index}/{total} 镜），时长 {seconds} 秒{span}。不要拍前后镜头，不要额外切镜。"
        shot_body = _verbatim_shot_block(group[0])
    else:
        scope = (
            f"按顺序只拍下面 {len(group)} 个连续切点（第 {index}/{total} 段），共 {seconds} 秒。"
            "只在这些切点切换，不要加镜。"
        )
        blocks = []
        for offset, shot in enumerate(group, start=1):
            clock = _shot_clock(shot)
            head = f"切点 {offset}" + (f"（原片 {clock}）" if clock else "")
            block = _verbatim_shot_block(shot)
            blocks.append(f"{head}\n{block}" if block else head)
        shot_body = "\n".join(blocks)

    if is_character:
        swap = (
            "将画面中需要替换的原人物完整替换为@目标角色。"
            "保留@参考视频的商品、场景、镜头运动、剪辑节奏与口播。"
        )
    else:
        swap_bits = [
            "将画面中需要替换的原商品完整替换为@目标商品的外观。",
            "保留@参考视频的人物、场景、镜头运动、剪辑节奏与口播氛围。",
        ]
        if has_triview:
            swap_bits.append(PRODUCT_TRIVIEW_NOTE)
        if subject:
            swap_bits.append(f"口播里的旧商品名改说「{subject}」。")
        swap = "".join(swap_bits)

    parts = [
        "使用@参考视频作为镜头、节奏与口播氛围基准。",
        scope,
        swap,
        shot_body,
    ]
    return enforce_no_embedded_captions("\n".join(part for part in parts if part))


def _digest_for_seedance(digest_text: str) -> str:
    """提炼页同款分镜稿。只砍给用户校对的【拆解存疑】,其余原文照传。"""
    body = (digest_text or "").strip()
    if not body:
        raise ValueError("参考视频拆解结果为空，请重试")
    head, sep, _tail = body.partition(DIGEST_TAIL_SECTION)
    return (head if sep else body).strip()


def _semantic_skill_dir() -> Path:
    """语义重构 skill 必须在 backend/skills 中，确保镜像内也能加载。"""
    base = Path(settings.BASE_DIR)
    for candidate in (base / "skills", base.parent.parent / "skills"):
        skill = candidate / "product-semantic-video-remix" / "SKILL.md"
        if skill.is_file():
            return skill
    return base / "skills" / "product-semantic-video-remix" / "SKILL.md"


def load_product_semantic_remix_skill() -> str:
    path = _semantic_skill_dir()
    try:
        return path.read_text(encoding="utf-8")
    except OSError as exc:
        raise ValueError("商品语义重构规则未部署，请联系管理员") from exc


def _product_semantic_facts(product: Product) -> dict:
    """只取商品库中已填写的事实，绝不从商品图片猜用途或功效。"""
    selling_points = [
        {"标题": point.title.strip(), "说明": point.detail.strip()}
        for point in product.selling_points.all()
        if point.title.strip() or point.detail.strip()
    ]
    return {
        "商品名称": product.title.strip(),
        "品牌": product.brand.strip(),
        "商品品类": product.category.strip(),
        "商品描述": product.description.strip(),
        "真实卖点": selling_points,
        "目标用户": product.target_audience.strip(),
        "规格": product.specs or {},
        "资料状态": "完整" if (product.category.strip() and (product.description.strip() or selling_points)) else "不足",
    }


def build_product_semantic_remix_messages(*, digest_text: str, product_facts: dict, duration: int, aspect_ratio: str) -> list[dict]:
    """参考片只作为结构输入；商品事实和参考片信息严格分栏，防止语义串台。"""
    facts = json.dumps(product_facts, ensure_ascii=False, indent=2)
    return [
        {"role": "system", "content": load_product_semantic_remix_skill()},
        {
            "role": "user",
            "content": (
                "请执行商品语义重构，不要直接把参考分镜改几个名词。\n"
                f"成片时长：{duration} 秒；画面比例：{aspect_ratio}。\n\n"
                "【新商品的唯一事实来源】\n"
                f"{facts}\n\n"
                "【参考视频拆解稿，仅可继承角色关系、环境、镜头语言、剪辑节奏和叙事功能】\n"
                f"{_digest_for_seedance(digest_text)}\n\n"
                "输出给视频模型的中文分镜稿：保留【镜头 01】格式和每镜时间、景别、机位、运镜；"
                "每镜的画面、人物动作、台词/旁白必须按新商品真实用途重写。"
                "不要输出分析过程、分类标签或 Markdown。"
            ),
        },
    ]


def validate_product_semantic_remix(text: str) -> str:
    """拒绝模型在事实不足时瞎编，或只给一段泛泛分析而没有可出片分镜。"""
    cleaned = (text or "").strip()
    if "MISSING_PRODUCT_FACTS" in cleaned:
        raise ValueError("商品资料不足：请先在商品库补充品类，以及商品描述或至少一条真实卖点后再复刻")
    if len(cleaned) < 120 or "【镜头" not in cleaned:
        raise ValueError("商品语义重构结果不完整，请重试")
    return cleaned


def rewrite_product_semantic_remix(*, task, digest_text: str, product_facts: dict, duration: int, aspect_ratio: str) -> str:
    """Gemini 负责重写商品语义，Seedance 只负责按这份新分镜出片。"""
    from .video_digest import DIGEST_MAX_TOKENS, resolve_digest_model_config
    from .services import _collect_extract_text, get_text_provider

    model_config = resolve_digest_model_config()
    if model_config is None:
        raise ValueError("商品语义重构模型未配置，请联系管理员")
    provider = get_text_provider(model_config)
    text, _payload = _collect_extract_text(
        provider,
        model_config,
        build_product_semantic_remix_messages(
            digest_text=digest_text,
            product_facts=product_facts,
            duration=duration,
            aspect_ratio=aspect_ratio,
        ),
        temperature=0.2,
        extra_body={"max_tokens": DIGEST_MAX_TOKENS},
    )
    return validate_product_semantic_remix(text)


def build_character_replace_prompt(
    digest_text: str,
    *,
    source_seconds: float = 0,
    output_seconds: int = 0,
) -> str:
    """提炼页分镜稿 + 换角色。原片不传 Seedance。"""
    from .services import enforce_no_embedded_captions

    source = int(round(float(source_seconds or 0)))
    output = int(output_seconds or 0)
    condense = (
        PRODUCT_CONDENSE_NOTE.format(source=source, output=output)
        if source and output and source - output >= 3
        else ""
    )
    return enforce_no_embedded_captions(
        "\n".join(part for part in (CHARACTER_PROMPT, _digest_for_seedance(digest_text), condense) if part)
    )


def compact_digest_for_video(digest_text: str) -> str:
    """分镜稿 → 交给视频模型的镜头表。

    解析失败(模型没按 skill 的格式吐)就回落:只砍掉【拆解存疑】,原文照传 ——
    宁可啰嗦也不能把镜头信息弄丢。
    """
    header, shots = _parse_digest(digest_text)
    if not shots:
        body = (digest_text or "").strip()
        head, sep, _tail = body.partition(DIGEST_TAIL_SECTION)
        return (head if sep else body).strip()
    return _render_shot_table(header, shots)


def build_product_replace_prompt(
    digest_text: str,
    subject_name: str,
    *,
    has_triview: bool = False,
    source_seconds: float = 0,
    output_seconds: int = 0,
) -> str:
    """提炼页分镜稿 + 换商品。原片不传 Seedance。"""
    from .services import enforce_no_embedded_captions

    subject = (subject_name or "").strip()
    source = int(round(float(source_seconds or 0)))
    output = int(output_seconds or 0)
    condense = (
        PRODUCT_CONDENSE_NOTE.format(source=source, output=output)
        if source and output and source - output >= 3
        else ""
    )
    extra = []
    if has_triview:
        extra.append(PRODUCT_TRIVIEW_NOTE)
    if subject:
        extra.append(f"口播里的旧商品名改说「{subject}」。")
    return enforce_no_embedded_captions(
        "\n".join(part for part in (PRODUCT_PROMPT, *extra, _digest_for_seedance(digest_text), condense) if part)
    )


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 "")
    created = task.status == AITask.Status.CREATED
    digesting = created and bool(payload.get("digest_pending"))
    reviewing = created and bool(payload.get("review_pending")) and not digesting
    plan = list(payload.get("shot_plan") or [])
    shot_total = int(payload.get("shot_total") or len(plan) or 0)
    shot_index = 0
    if plan:
        active = next((item for item in plan if item.get("status") in {"submitted", "polling"}), None)
        if active is None:
            done = sum(1 for item in plan if item.get("status") == "succeeded")
            shot_index = done
        else:
            shot_index = int(active.get("index") or 0)
    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 "",
        "review_stage": "reviewing" if reviewing else "",
        "digest_stage": "digesting" if digesting else "",
        "digest_text": str(payload.get("digest_text") or ""),
        "digest_shots": payload.get("digest_shots") or shot_total or 0,
        "digest_source_name": payload.get("digest_source_name") or "",
        "digest_source": payload.get("digest_source_ref") or None,
        "shot_index": shot_index,
        "shot_total": shot_total,
    })
    # 火山编辑接口内部用 adaptive/-1；页面仍展示用户实际上传视频的比例和时长。
    if replace_mode == "character":
        data["duration"] = int(payload.get("billing_duration") or 0)
        data["aspect_ratio"] = str(payload.get("billing_aspect_ratio") or data.get("aspect_ratio") or "")
    return data


def submit_video_replace(*, team, user, params: dict):
    """校验素材 → 送审 → 已过审则直接生成,否则建 CREATED 任务等绿盾。失败抛 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("请选择角色或上传角色参考图")

    # 单飞闸:一个团队同时只允许一条复刻在跑。刷新页面时前端拉状态有空档,
    # 用户容易以为「没任务」再点一次,重复扣费。_reap_stale_free_video_tasks 会先回收
    # 死任务(CREATED 超 16 分钟等),所以不会被永久锁住。
    running = get_inflight_video_replace(team)
    if running is not None:
        raise VideoReplaceInProgress(running)

    video = _team_asset(team, params.get("video_asset_id"), kind=Asset.Type.VIDEO, label="参考视频")
    video_seconds = _asset_duration_seconds(video)
    if video_seconds > REPLACE_REF_DURATION_MAX:
        raise ValueError("参考视频不能超过 30 秒，请剪短后重试")

    if has_product:
        subject_name, image_refs, subject_source = _product_library_refs(team, product_id)
        product_facts = _product_semantic_facts_for_id(team, product_id)
    elif has_model:
        subject_name, image_refs, subject_source = _character_library_refs(team, model_id)
        product_facts = {}
    else:
        noun = "商品" if replace_mode == "product" else "角色"
        subject_name, image_refs, subject_source = _temporary_image_refs(team, image_ids, noun=noun)
        product_facts = {
            "商品名称": subject_name,
            "资料状态": "不足",
            "说明": "临时上传图片只提供外观，未提供品类、用途或真实卖点。",
        }

    # 角色替换是 Seedance 视频编辑:火山要求比例/时长跟随输入视频，不能由前端指定。
    duration = -1 if replace_mode == "character" else _output_duration(params.get("duration"), video_seconds)
    billing_ratio = str(params.get("aspect_ratio") or "9:16")
    if billing_ratio not in RATIOS:
        billing_ratio = "9:16"
    extra = {
        "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 "",
        "product_facts": product_facts,
        # adaptive / -1 只用于火山请求；预估积分仍按参考视频实际时长和上传时检测出的比例计算。
        "billing_duration": max(4, int(round(video_seconds or 4))),
        "billing_aspect_ratio": billing_ratio,
    }
    base_params = {
        "mode": "universal",
        # 固定 Seedance 2.5,不接受前端指定别的档
        "model": REPLACE_MODEL,
        "aspect_ratio": "adaptive" if replace_mode == "character" else billing_ratio,
        "resolution": str(params.get("resolution") or "720p"),
        "duration": duration,
        "seed": params.get("seed", -1),
        "generate_audio": True,
        "feature": FEATURE,
    }

    # 商品复刻需要 Gemini 拆镜后重构商品语义；角色复刻直接交给 Seedance 2.5，
    # 原视频与角色图都先进入同一火山素材库审核，再以 asset:// 作为参考。
    if replace_mode == "product":
        from .video_digest import resolve_digest_model_config

        if resolve_digest_model_config() is None:
            raise ValueError("视频提炼模型未配置，请联系管理员")
    else:
        image_refs = [
            _owned_ref(video, kind="video", role="reference_video", label="参考视频"),
            *image_refs,
        ]
    # 商品图/角色图仍要过审才能当生成参考。提前送审 → 审核和提炼并行跑,明确不过审的直接 400。
    review_state = _ensure_replace_refs_reviewed(team, image_refs)
    if review_state == "failed":
        raise ValueError(REVIEW_FAILED)
    image_refs = _refresh_replace_refs(team, image_refs)
    extra.update({
        "digest_source_asset_id": str(video.id),
        "digest_source_name": video.name or "参考视频",
        "digest_source_duration": round(video_seconds, 2),
        # 原片快照只给历史卡「原视频」和重新生成用,不进 Seedance references。
        "digest_source_ref": _owned_ref(video, kind="video", role="reference_video", label="参考视频"),
        "review_pending": review_state != "ready",
    })
    return _create_reviewing_task(
        team=team,
        user=user,
        params={
            **base_params,
            "prompt": DIGEST_PENDING_PROMPT if replace_mode == "product" else CHARACTER_PROMPT,
            "references": image_refs,
            "extra_payload": extra,
        },
        digest_pending=replace_mode == "product",
    )


def advance_video_replace(task):
    """推进复刻:审核中 → 整份分镜稿一次出片。旧逐镜任务也按单次收尾,不再拼接。"""
    if not is_video_replace_task(task):
        return task
    if task.status != AITask.Status.CREATED:
        from .free_video import finalize_free_video

        return finalize_free_video(task=task)

    payload = task.request_payload or {}
    if payload.get("digest_pending"):
        # worker 还在拆参考视频,提示词都没生成,别当成「审核中」反复送审。
        # worker 挂了也不会永远卡着:CREATED 超 16 分钟由 _reap_stale_free_video_tasks 回收退费。
        return task
    references = (
        _image_refs(payload.get("references") or [])
        if payload.get("replace_mode") == "product"
        else list(payload.get("references") or [])
    )
    try:
        state = _ensure_replace_refs_reviewed(task.team, references)
    except ValueError as exc:
        return _fail_reviewing_task(task, str(exc))
    if state == "failed":
        return _fail_reviewing_task(task, REVIEW_FAILED)
    if state != "ready":
        return task

    refreshed = _refresh_replace_refs(task.team, references)
    try:
        _assert_replace_refs_ready(refreshed)
    except ValueError as exc:
        return _fail_reviewing_task(task, str(exc))
    with transaction.atomic():
        locked = AITask.objects.select_for_update().get(id=task.id)
        if locked.status != AITask.Status.CREATED:
            return locked
        next_payload = dict(locked.request_payload or {})
        next_payload["references"] = refreshed
        next_payload["review_pending"] = False
        locked.request_payload = next_payload
        locked.save(update_fields=["request_payload", "updated_at"])
        task = locked
    return start_replace_generation(task)


def start_replace_generation(task):
    """审核过了:整份分镜稿一次交给 Seedance 2.5。只带商品图/角色图,不传用户原片。"""
    payload = dict(task.request_payload or {})
    if payload.get("replace_mode") == "product":
        payload["references"] = _image_refs(payload.get("references") or [])
        payload.pop("shot_plan", None)
        payload.pop("shot_total", None)
    task.request_payload = payload
    task.save(update_fields=["request_payload", "updated_at"])
    return start_pending_free_video(task)


def run_replace_digest(task) -> None:
    """Worker:参考视频 → 提炼页同款分镜稿 → 一次交给 Seedance。不传原片。

    商品和角色共用。这一步失败 = 整条复刻失败。此时还没预留视频积分,不扣费。
    """
    from apps.ai.video_digest import VideoDigestError, digest_asset_video

    if not is_video_replace_task(task) or task.status != AITask.Status.CREATED:
        return
    payload = task.request_payload or {}
    # 角色复刻没有 Gemini 拆镜环节；审核通过后由 advance_video_replace 直接提交 Seedance。
    if payload.get("replace_mode") == "character":
        return
    if not payload.get("digest_pending"):
        return

    try:
        asset = _load_replace_asset(
            task.team, uuid.UUID(str(payload.get("digest_source_asset_id") or "")), "参考视频"
        )
    except (ValueError, TypeError):
        _fail_reviewing_task(task, "参考视频已失效，请重新上传", error_code="asset_unavailable")
        return

    replace_mode = str(payload.get("replace_mode") or "product")
    subject = str(payload.get("subject_name") or "")
    has_triview = any(
        (ref or {}).get("label") == TRIVIEW_LABEL for ref in (payload.get("references") or [])
    )
    source_seconds = float(payload.get("digest_source_duration") or 0)
    output_seconds = int(payload.get("duration") or 0)

    try:
        digest, meta = digest_asset_video(asset=asset, task=task)
        if replace_mode == "character":
            prompt = build_character_replace_prompt(
                digest, source_seconds=source_seconds, output_seconds=output_seconds,
            )
        else:
            prompt = rewrite_product_semantic_remix(
                task=task,
                digest_text=digest,
                product_facts=payload.get("product_facts") or {"商品名称": subject, "资料状态": "不足"},
                duration=output_seconds,
                aspect_ratio=str(payload.get("aspect_ratio") or "9:16"),
            )
            if has_triview and TRIVIEW_LABEL not in prompt:
                prompt = f"{prompt}\n商品各面与材质细节以@{TRIVIEW_LABEL}为准。"
    except (VideoDigestError, ValueError) as exc:
        _fail_reviewing_task(task, str(exc), error_code="processing_failed")
        return
    except Exception as exc:  # noqa: BLE001 — 拆解任何异常都要把任务收尾,别留 CREATED 僵尸
        logger.exception("video replace digest failed for %s", task.id)
        _fail_reviewing_task(task, f"参考视频拆解失败：{exc}", error_code="processing_failed")
        return

    with transaction.atomic():
        locked = AITask.objects.select_for_update().get(id=task.id)
        if locked.status != AITask.Status.CREATED:
            return
        next_payload = dict(locked.request_payload or {})
        next_payload.update(meta)
        next_payload["digest_pending"] = False
        next_payload["digest_text"] = digest[:32000]
        next_payload["semantic_remix"] = replace_mode == "product"
        next_payload["prompt"] = prompt
        next_payload["references"] = _image_refs(next_payload.get("references") or [])
        next_payload.pop("shot_plan", None)
        next_payload.pop("shot_total", None)
        locked.request_payload = next_payload
        locked.save(update_fields=["request_payload", "updated_at"])
        task = locked

    # 拆完就地推进一次:参考图多半早已过审,能直接提交火山,省掉一轮 8s 轮询。
    try:
        task = advance_video_replace(task)
    except Exception:  # noqa: BLE001 — 推进失败交给轮询重试,任务还在 CREATED
        logger.warning("video replace advance after digest failed for %s", task.id, exc_info=True)
        task.refresh_from_db()
    if task.status == AITask.Status.CREATED:
        _enqueue_replace_review_poll(task)


def start_pending_replace_shots(task):
    """旧逐镜路径已下线。一律整片一次出,且不传用户原片。"""
    return start_replace_generation(task)


def _legacy_start_pending_replace_shots(task):
    """CREATED → 按全部镜头总时长预留积分 → 提交第一镜。"""
    from apps.billing.services.ledger import reserve_credit

    from .free_video import (
        _dispatch_free_video_provider,
        _fail_pending_free_video,
        build_content_items,
        model_duration_range,
    )

    if task.status != AITask.Status.CREATED:
        return task
    payload = task.request_payload or {}
    plan = list(payload.get("shot_plan") or [])
    if not plan:
        return start_pending_free_video(task)
    first = plan[0]
    prompt = str(first.get("prompt") or payload.get("prompt") or "").strip()
    mode = str(payload.get("mode") or "universal")
    aspect_ratio = str(payload.get("aspect_ratio") or "9:16")
    resolution = str(payload.get("resolution") or "720p")
    generate_audio = bool(payload.get("generate_audio", True))
    search_mode = str(payload.get("search_mode") or "off")
    feature = str(payload.get("feature") or FEATURE)
    try:
        seed = int(payload.get("seed") if payload.get("seed") is not None else -1)
    except (TypeError, ValueError):
        seed = -1
    billed_duration = sum(int(item.get("seconds") or 0) for item in plan) or int(payload.get("duration") or 5)
    references = _seedance_references(payload)
    try:
        built = build_content_items(
            team=task.team, prompt=prompt, mode=mode, references=references,
            max_ref_seconds=model_duration_range(task.model_config)[1],
        )
    except ValueError as exc:
        message = str(exc)
        if "正在审核" in message or "已提交审核" in message or "尚未完成合规审核" in message:
            return task
        return _fail_pending_free_video(task, message)

    tokens, quote = quote_video_estimate(
        task.model_config,
        aspect_ratio=aspect_ratio,
        resolution=resolution,
        duration=billed_duration,
        references=built["snapshots"],
        team=task.team,
    )
    reserve_amount = video_reserve_amount(quote.points)

    with transaction.atomic():
        locked = (
            AITask.objects.select_for_update()
            .select_related("model_config", "model_config__provider", "team", "created_by")
            .get(id=task.id)
        )
        if locked.status != AITask.Status.CREATED:
            return locked
        try:
            reserve_credit(team=locked.team, user=locked.created_by, task=locked, amount=reserve_amount)
        except ValueError as exc:
            message = "团队余额不足，请充值后重试" if "insufficient credit" in str(exc) else str(exc)
            locked.status = AITask.Status.FAILED
            locked.error_code = "user_credit_insufficient" if "余额不足" in message else "invalid_input"
            locked.error_message = message[:2000]
            locked.completed_at = timezone.now()
            locked.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"])
            return locked
        next_payload = dict(locked.request_payload or {})
        plan = list(next_payload.get("shot_plan") or plan)
        plan[0] = {**plan[0], "status": "submitted"}
        next_payload["shot_plan"] = plan
        next_payload["prompt"] = prompt
        next_payload["api_prompt"] = built["api_prompt"]
        next_payload["references"] = built["snapshots"]
        next_payload["estimated_tokens"] = tokens
        next_payload["review_pending"] = False
        next_payload["duration"] = billed_duration
        locked.request_payload = next_payload
        locked.estimated_cost = quote.points
        locked.status = AITask.Status.RESERVED
        locked.save(update_fields=["request_payload", "estimated_cost", "status", "updated_at"])

    dispatched = _dispatch_free_video_provider(
        task=locked,
        built=built,
        model_config=locked.model_config,
        aspect_ratio=aspect_ratio,
        duration=int(first.get("seconds") or MIN_SHOT_SECONDS),
        resolution=resolution,
        generate_audio=generate_audio,
        seed=seed,
        search_mode=search_mode,
        feature=feature,
        mode=mode,
        poll_countdown=30,
    )
    return _stamp_current_shot_id(dispatched)


def advance_replace_shots(task):
    """轮询当前这一镜:还在跑就等;失败整单失败;成功则提交下一镜,全部完成再拼接。"""
    from .free_video import (
        _dispatch_free_video_provider,
        build_content_items,
        finalize_free_video,
        model_duration_range,
    )
    from .generation_errors import classify_generation_error
    from .routing_policy import load_model_routing_policy
    from .services import get_video_provider

    payload = dict(task.request_payload or {})
    plan = list(payload.get("shot_plan") or [])
    if not plan:
        return finalize_free_video(task=task)
    if task.status not in (AITask.Status.RESERVED, AITask.Status.SUBMITTED, AITask.Status.POLLING):
        return task

    current = _active_shot(plan)
    if current is None:
        if plan and all(item.get("status") == "succeeded" and item.get("media") for item in plan):
            return complete_replace_shots(task)
        return task

    if not current.get("provider_task_id"):
        if task.provider_task_id and current.get("status") in {"submitted", "polling", "queued"}:
            current = {**current, "provider_task_id": task.provider_task_id, "status": "submitted"}
            _write_shot(task, current)
            plan = _read_shot_plan(task)
            current = _shot_by_index(plan, current["index"]) or current
        else:
            return _dispatch_next_replace_shot(task, current)

    video_policy = load_model_routing_policy().video
    submit_attempt = (
        task.model_attempts.filter(status="succeeded", operation="video_generate")
        .select_related("model_config__provider")
        .order_by("-sequence")
        .first()
    )
    actual_model = submit_attempt.model_config if submit_attempt and submit_attempt.model_config else task.model_config
    provider = get_video_provider(actual_model)
    response = provider.poll_video_task(
        endpoint=actual_model.endpoint,
        provider_task_id=current["provider_task_id"],
        timeout=video_policy.poll_request_timeout,
    )
    remote_status = str(response.get("status") or "")

    if remote_status in {"queued", "running", "processing", "submitted"}:
        if task.status != AITask.Status.POLLING or current.get("status") != "polling":
            current = {**current, "status": "polling"}
            task = _write_shot(task, current)
            if task.status != AITask.Status.POLLING:
                task.status = AITask.Status.POLLING
                task.save(update_fields=["status", "updated_at"])
        return task

    if remote_status in {"failed", "expired", "cancelled"}:
        err = response.get("error") or {}
        code = str(err.get("code") or "")
        raw_message = str(err.get("message") or "video generation failed")
        public_error = classify_generation_error(
            RuntimeError(raw_message),
            operation="video_generate",
            provider_code=code,
            reference_id=str(task.id),
        )
        return _fail_replace_generation(
            task, raw_message, error_code=code[:64] or "CreateTaskError", hint=public_error.fallback_message,
        )

    try:
        media = provider.extract_first_media_url(response)
    except Exception as exc:  # noqa: BLE001
        return _fail_replace_generation(task, str(exc), error_code="PostprocessError")
    usage = response.get("usage") or {}
    try:
        tokens = int(usage.get("total_tokens") or 0)
    except (TypeError, ValueError):
        tokens = 0
    current = {**current, "status": "succeeded", "media": media, "tokens": tokens}
    task = _write_shot(task, current)
    plan = _read_shot_plan(task)
    nxt = next((item for item in plan if item.get("status") in {"queued", "", None}), None)
    if nxt is None:
        if len(plan) == 1:
            return finalize_free_video(task=task)
        return complete_replace_shots(task)
    return _dispatch_next_replace_shot(task, nxt)


def _dispatch_next_replace_shot(task, shot: dict):
    from .free_video import _dispatch_free_video_provider, build_content_items, model_duration_range

    payload = dict(task.request_payload or {})
    prompt = str(shot.get("prompt") or "").strip()
    if not prompt:
        return _fail_replace_generation(task, "分镜提示词缺失，请重试", error_code="processing_failed")
    try:
        built = build_content_items(
            team=task.team, prompt=prompt, mode=str(payload.get("mode") or "universal"),
            references=_seedance_references(payload),
            max_ref_seconds=model_duration_range(task.model_config)[1],
        )
    except ValueError as exc:
        return _fail_replace_generation(task, str(exc), error_code="invalid_input")

    with transaction.atomic():
        locked = AITask.objects.select_for_update().select_related(
            "model_config", "model_config__provider", "team", "created_by",
        ).get(id=task.id)
        if locked.status not in (AITask.Status.RESERVED, AITask.Status.SUBMITTED, AITask.Status.POLLING):
            return locked
        plan = list((locked.request_payload or {}).get("shot_plan") or [])
        target = _shot_by_index(plan, shot["index"])
        if target is None or target.get("status") not in {"queued", "", None, "submitted"}:
            return locked
        if target.get("provider_task_id") and target.get("status") in {"submitted", "polling"}:
            return locked
        next_payload = dict(locked.request_payload or {})
        next_payload["prompt"] = prompt
        next_payload["api_prompt"] = built["api_prompt"]
        plan = [
            {**item, "status": "submitted"} if item.get("index") == shot["index"] else item
            for item in plan
        ]
        next_payload["shot_plan"] = plan
        locked.request_payload = next_payload
        locked.save(update_fields=["request_payload", "updated_at"])

    try:
        seed = int(payload.get("seed") if payload.get("seed") is not None else -1)
    except (TypeError, ValueError):
        seed = -1
    dispatched = _dispatch_free_video_provider(
        task=locked,
        built=built,
        model_config=locked.model_config,
        aspect_ratio=str(payload.get("aspect_ratio") or "9:16"),
        duration=int(shot.get("seconds") or MIN_SHOT_SECONDS),
        resolution=str(payload.get("resolution") or "720p"),
        generate_audio=bool(payload.get("generate_audio", True)),
        seed=seed,
        search_mode=str(payload.get("search_mode") or "off"),
        feature=str(payload.get("feature") or FEATURE),
        mode=str(payload.get("mode") or "universal"),
        poll_countdown=30,
    )
    return _stamp_current_shot_id(dispatched)


def complete_replace_shots(task):
    """全部镜头出完:下载 → ffmpeg 拼接 → 转存 TOS → 按 tokens 合计结算。"""
    from decimal import Decimal

    from apps.billing.pricing import quote_video_actual
    from apps.billing.services.ledger import charge_reserved_credit, release_credit

    from .free_video import _notify_failure, _store_free_video_media
    from .generation_errors import classify_generation_error

    with transaction.atomic():
        locked = AITask.objects.select_for_update().get(id=task.id)
        if locked.status not in (AITask.Status.SUBMITTED, AITask.Status.POLLING):
            return locked
        locked.status = AITask.Status.POSTPROCESSING
        locked.save(update_fields=["status", "updated_at"])

    payload = dict(locked.request_payload or {})
    plan = list(payload.get("shot_plan") or [])
    try:
        if len(plan) == 1 and plan[0].get("media"):
            _store_free_video_media(task=locked, media=plan[0]["media"])
        else:
            video_bytes = _concat_shot_media([item.get("media") or "" for item in plan])
            _store_free_video_media(task=locked, video_bytes=video_bytes)
        total_tokens = 0
        for item in plan:
            try:
                total_tokens += int(item.get("tokens") or 0)
            except (TypeError, ValueError):
                pass
        resolution = payload.get("resolution") or "720p"
        if total_tokens > 0:
            settle = quote_video_actual(
                locked.model_config, tokens=total_tokens, with_video_ref=True,
                resolution=resolution, multiplier=Decimal(str(payload.get("price_multiplier") or "1")),
            )
            actual, base_cost = settle.points, settle.base_cost_yuan
            payload["actual_tokens"] = total_tokens
            if settle.meta.get("rate"):
                payload["points_per_yuan_snapshot"] = settle.meta["rate"]
        else:
            actual, base_cost = locked.estimated_cost, locked.base_cost
        with transaction.atomic():
            locked = AITask.objects.select_for_update().get(id=locked.id)
            if locked.status != AITask.Status.POSTPROCESSING:
                return locked
            reservation = locked.credit_reservation
            if actual > reservation.amount:
                logger.warning(
                    "video replace task %s actual cost %s exceeds reserved %s, clamped",
                    locked.id, actual, reservation.amount,
                )
                actual = reservation.amount
            locked.status = AITask.Status.SUCCEEDED
            locked.actual_cost = actual
            locked.base_cost = base_cost
            locked.request_payload = payload
            locked.completed_at = timezone.now()
            locked.save(
                update_fields=["status", "actual_cost", "base_cost", "request_payload", "completed_at", "updated_at"]
            )
            charge_reserved_credit(reservation=reservation, actual_amount=actual)
        return locked
    except Exception as exc:  # noqa: BLE001
        logger.exception("video replace concat failed for task %s", locked.id)
        public_error = classify_generation_error(
            exc, operation="video_generate", internal_kind="processing_failed",
            reference_id=str(locked.id),
        )
        with transaction.atomic():
            locked = AITask.objects.select_for_update().get(id=locked.id)
            if locked.status != AITask.Status.POSTPROCESSING:
                return locked
            locked.status = AITask.Status.FAILED
            locked.error_code = "PostprocessError"
            locked.error_message = str(exc)[:2000]
            locked.completed_at = timezone.now()
            locked.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"])
            release_credit(reservation=locked.credit_reservation, reason=str(exc)[:200])
        _notify_failure(locked, raw=str(exc), hint=public_error.fallback_message)
        return locked


def _concat_shot_media(urls: list[str]) -> bytes:
    import subprocess
    import tempfile
    from pathlib import Path

    from .providers.volcano import VolcanoArkProvider

    clips = [url for url in urls if url]
    if not clips:
        raise ValueError("没有可拼接的镜头成片")
    if len(clips) == 1:
        fileobj, _ctype = VolcanoArkProvider.media_to_bytes(clips[0])
        return fileobj.getvalue()

    with tempfile.TemporaryDirectory(prefix="airshelf-replace-concat-") as tmp:
        tmp_dir = Path(tmp)
        paths = []
        for index, url in enumerate(clips):
            fileobj, _ctype = VolcanoArkProvider.media_to_bytes(url)
            path = tmp_dir / f"shot{index:02d}.mp4"
            path.write_bytes(fileobj.getvalue())
            paths.append(path)
        output = tmp_dir / "out.mp4"
        inputs: list[str] = []
        for path in paths:
            inputs.extend(["-i", str(path)])
        n = len(paths)
        parts = [f"[{i}:v]fps=24,format=yuv420p,setsar=1[v{i}]" for i in range(n)]
        audio_ok = []
        for path in paths:
            probe = subprocess.run(
                ["ffprobe", "-v", "error", "-select_streams", "a",
                 "-show_entries", "stream=codec_type", "-of", "csv=p=0", str(path)],
                capture_output=True, text=True, timeout=30,
            )
            audio_ok.append(bool((probe.stdout or "").strip()))
        for i, path in enumerate(paths):
            if audio_ok[i]:
                parts.append(f"[{i}:a]aresample=44100,aformat=sample_fmts=fltp:channel_layouts=stereo[a{i}]")
            else:
                dur_proc = subprocess.run(
                    ["ffprobe", "-v", "error", "-show_entries", "format=duration",
                     "-of", "csv=p=0", str(path)],
                    capture_output=True, text=True, timeout=30,
                )
                try:
                    dur = max(0.1, float((dur_proc.stdout or "4").strip() or "4"))
                except ValueError:
                    dur = 4.0
                parts.append(
                    f"anullsrc=channel_layout=stereo:sample_rate=44100,atrim=0:{dur:.3f},"
                    f"asetpts=PTS-STARTPTS,aformat=sample_fmts=fltp:channel_layouts=stereo[a{i}]"
                )
        concat_in = "".join(f"[v{i}][a{i}]" for i in range(n))
        parts.append(f"{concat_in}concat=n={n}:v=1:a=1[v][a]")
        cmd = [
            "ffmpeg", "-y", *inputs, "-filter_complex", ";".join(parts),
            "-map", "[v]", "-map", "[a]",
            "-c:v", "libx264", "-pix_fmt", "yuv420p", "-preset", "veryfast", "-threads", "2",
            "-c:a", "aac", "-b:a", "192k", "-movflags", "+faststart", str(output),
        ]
        proc = subprocess.run(cmd, capture_output=True, timeout=180)
        if proc.returncode != 0 or not output.exists() or not output.stat().st_size:
            err = (proc.stderr or b"").decode("utf-8", errors="replace")[-500:]
            raise RuntimeError(f"镜头拼接失败：{err or 'ffmpeg 未产出文件'}")
        return output.read_bytes()


def _stamp_current_shot_id(task):
    """把刚提交火山拿到的 provider_task_id 写回当前这一镜,下一轮 poll 才找得到。"""
    if task.status not in (AITask.Status.SUBMITTED, AITask.Status.POLLING) or not task.provider_task_id:
        return task
    payload = dict(task.request_payload or {})
    plan = list(payload.get("shot_plan") or [])
    changed = False
    for item in plan:
        if item.get("status") in {"submitted", "polling"} and not item.get("provider_task_id"):
            item["provider_task_id"] = task.provider_task_id
            changed = True
            break
    if not changed:
        return task
    payload["shot_plan"] = plan
    task.request_payload = payload
    task.save(update_fields=["request_payload", "updated_at"])
    return task


def _active_shot(plan: list[dict]) -> dict | None:
    for item in plan:
        if item.get("status") in {"submitted", "polling"}:
            return item
    for item in plan:
        if item.get("status") in {"queued", "", None}:
            return item
    return None


def _shot_by_index(plan: list[dict], index) -> dict | None:
    try:
        index = int(index)
    except (TypeError, ValueError):
        return None
    return next((item for item in plan if int(item.get("index") or 0) == index), None)


def _read_shot_plan(task) -> list[dict]:
    return list((task.request_payload or {}).get("shot_plan") or [])


def _write_shot(task, shot: dict):
    with transaction.atomic():
        locked = AITask.objects.select_for_update().get(id=task.id)
        payload = dict(locked.request_payload or {})
        plan = list(payload.get("shot_plan") or [])
        plan = [
            shot if int(item.get("index") or 0) == int(shot.get("index") or 0) else item
            for item in plan
        ]
        payload["shot_plan"] = plan
        if shot.get("provider_task_id"):
            locked.provider_task_id = shot["provider_task_id"]
        locked.request_payload = payload
        locked.save(update_fields=["request_payload", "provider_task_id", "updated_at"])
        return locked


def _fail_replace_generation(task, message: str, *, error_code: str = "", hint: str = ""):
    from apps.billing.services.ledger import release_credit

    from .free_video import _fail_pending_free_video, _notify_failure

    if task.status in (AITask.Status.CREATED, AITask.Status.RESERVED):
        return _fail_reviewing_task(task, message, error_code=error_code or "processing_failed")
    public_hint = hint or message
    with transaction.atomic():
        locked = AITask.objects.select_for_update().get(id=task.id)
        if locked.status in (AITask.Status.SUCCEEDED, AITask.Status.FAILED, AITask.Status.CANCELLED):
            return locked
        locked.status = AITask.Status.FAILED
        locked.error_code = (error_code or "CreateTaskError")[:64]
        locked.error_message = message[:2000]
        locked.completed_at = timezone.now()
        locked.save(update_fields=["status", "error_code", "error_message", "completed_at", "updated_at"])
        reservation = getattr(locked, "credit_reservation", None)
        if reservation is not None:
            release_credit(reservation=reservation, reason=message[:200])
    _notify_failure(locked, raw=message, hint=public_hint)
    return locked


def _legacy_replace_mode(prompt: str) -> str:
    return "character" if prompt.startswith("[视频复刻·角色]") else "product"


def _fail_reviewing_task(task, message: str, *, error_code: str = ""):
    """收尾一个还没提交火山的任务。

    ``_fail_pending_free_video`` 把错误码写死成 content_rejected —— 那是审核不通过的语义,
    前端会渲染成「内容未通过生成审核」且不可重试。拆解失败/素材丢失不是那回事,
    这里按实际原因改码,否则模型抖一下会被误报成合规问题,用户白白去换素材。
    """
    from .free_video import _fail_pending_free_video

    task = _fail_pending_free_video(task, message)
    if error_code and task.status == AITask.Status.FAILED and task.error_code != error_code:
        task.error_code = error_code
        task.save(update_fields=["error_code", "updated_at"])
    return task


def _enqueue_replace_review_poll(task):
    try:
        from .tasks import poll_free_video_task

        poll_free_video_task.apply_async(args=[str(task.id), 0], countdown=8)
    except Exception:  # noqa: BLE001
        logger.error("video replace review poll enqueue failed; relying on client polling", exc_info=True)


def _enqueue_replace_digest(task):
    try:
        from .tasks import run_video_replace_digest_task

        run_video_replace_digest_task.delay(str(task.id))
    except Exception:  # noqa: BLE001
        logger.error("video replace digest enqueue failed", exc_info=True)


def _create_reviewing_task(*, team, user, params: dict, digest_pending: bool = False):
    """还不能提交火山时:只建 CREATED 任务,不预留积分。

    商品/角色复刻都先拆解参考视频;参考图若还在审核,拆完再等绿盾。
    """
    model_name = str(params.get("model") or REPLACE_MODEL)
    aspect_ratio = str(params.get("aspect_ratio") or "9:16")
    extra = params.get("extra_payload") if isinstance(params.get("extra_payload"), dict) else {}
    is_character_edit = extra.get("replace_mode") == "character"
    resolution = str(params.get("resolution") or "720p")
    try:
        duration = int(params.get("duration") or 5)
    except (TypeError, ValueError):
        raise ValueError("时长参数无效")
    if model_name not in FREE_VIDEO_MODELS:
        raise ValueError("模型无效")
    if aspect_ratio not in RATIOS and not (is_character_edit and aspect_ratio == "adaptive"):
        raise ValueError("画面比例无效")
    if resolution not in RESOLUTIONS:
        raise ValueError("分辨率无效")
    low, high = replace_duration_range()
    if is_character_edit and duration != -1:
        raise ValueError("角色替换时长必须跟随参考视频")
    if not is_character_edit and not low <= duration <= high:
        raise ValueError(f"视频时长需在 {low}-{high} 秒之间")

    model_config = (
        ModelConfig.objects.select_related("provider")
        .filter(name=model_name, capability=ModelConfig.Capability.VIDEO, status=ModelConfig.Status.ACTIVE)
        .first()
    )
    if model_config is None:
        raise ValueError("视频模型未配置，请联系管理员")

    _reap_stale_free_video_tasks(team=team)
    max_concurrent = int(getattr(settings, "FREE_VIDEO_MAX_CONCURRENT", 3))
    in_flight = AITask.objects.filter(
        team=team, task_type=AITask.Type.FREE_VIDEO, status__in=IN_FLIGHT_STATUSES
    ).count()
    if in_flight >= max_concurrent:
        raise ValueError(f"当前有 {in_flight} 个视频任务进行中（上限 {max_concurrent}），请等待完成后再提交")

    references = params.get("references") or []
    billing_duration = int(extra.get("billing_duration") or duration)
    billing_ratio = str(extra.get("billing_aspect_ratio") or aspect_ratio)
    tokens, quote = quote_video_estimate(
        model_config,
        aspect_ratio=billing_ratio,
        resolution=resolution,
        duration=billing_duration,
        references=references,
        team=team,
    )
    reserve_amount = video_reserve_amount(quote.points)
    account = CreditAccount.objects.filter(team=team).first()
    available = (account.balance - account.reserved_balance) if account else Decimal("0")
    if available < reserve_amount:
        raise ValueError("团队余额不足，请充值后重试")

    try:
        seed = int(params.get("seed") if params.get("seed") is not None else -1)
    except (TypeError, ValueError):
        seed = -1
    request_payload = {
        "feature": FEATURE,
        "mode": "universal",
        "model": model_name,
        "endpoint": model_config.endpoint,
        "prompt": params.get("prompt") or "",
        "api_prompt": "",
        "aspect_ratio": aspect_ratio,
        "resolution": resolution,
        "duration": duration,
        "billing_duration": billing_duration,
        "billing_aspect_ratio": billing_ratio,
        "seed": seed,
        "generate_audio": True,
        "search_mode": "off",
        "estimated_tokens": tokens,
        "price_multiplier": quote.meta.get("price_multiplier", "1"),
        "points_per_yuan_snapshot": quote.meta.get("rate", ""),
        "references": references,
        "model_routing_v1": True,
        "review_pending": True,
        "digest_pending": digest_pending,
    }
    for key, value in extra.items():
        if key in request_payload or value in (None, ""):
            continue
        request_payload[key] = value

    task = AITask.objects.create(
        team=team,
        created_by=user,
        project=None,
        task_type=AITask.Type.FREE_VIDEO,
        status=AITask.Status.CREATED,
        model_config=model_config,
        idempotency_key=f"free_video:{team.id}:{uuid.uuid4()}",
        request_payload=request_payload,
        estimated_cost=quote.points,
        base_cost=Decimal("0"),
    )
    if digest_pending:
        _enqueue_replace_digest(task)
    else:
        _enqueue_replace_review_poll(task)
    return task


def _replace_ref_assets(team, references: list) -> list[tuple[dict, Asset]]:
    out = []
    seen = set()
    for ref in references or []:
        raw_id = ref.get("asset_id")
        if not raw_id:
            continue
        try:
            parsed = uuid.UUID(str(raw_id))
        except (TypeError, ValueError):
            continue
        if parsed in seen:
            continue
        seen.add(parsed)
        asset = _load_replace_asset(team, parsed, ref.get("label") or "参考素材")
        out.append((ref, asset))
    return out


def _load_replace_asset(team, asset_id: uuid.UUID, label: str) -> Asset:
    asset = Asset.objects.filter(id=asset_id, is_deleted=False, purged_at__isnull=True).first()
    if asset is None:
        raise ValueError(f"{label}不存在或已被删除")
    if asset.team_id == team.id:
        return asset
    if Model.objects.filter(Q(is_official=True), Q(portrait_asset=asset) | Q(triview_asset=asset)).exists():
        return asset
    raise ValueError(f"{label}不存在或已被删除")


def _ensure_replace_refs_reviewed(team, references: list) -> str:
    """送审/轮询全部参考素材。返回 ready / pending / failed;审核未配置且无 remote_id 抛错。"""
    from apps.assets import assets_client
    from apps.assets.review import poll_asset_review, submit_asset_for_review

    pairs = _replace_ref_assets(team, references)
    if not pairs:
        raise ValueError("请先上传参考素材")
    states = []
    for ref, asset in pairs:
        label = ref.get("label") or asset.name or "参考素材"
        if asset.review_status == "active" and asset.review_remote_id:
            states.append("ready")
            continue
        if not assets_client.is_enabled():
            raise ValueError(REVIEW_UNAVAILABLE)
        if asset.review_status == "processing" and asset.review_remote_id:
            poll_asset_review(asset)
            asset.refresh_from_db(fields=["review_status", "review_remote_id", "review_error"])
        elif asset.review_status != "active" or not asset.review_remote_id:
            ok = submit_asset_for_review(asset, force=True)
            asset.refresh_from_db(fields=["review_status", "review_remote_id", "review_error"])
            if not ok and not asset.review_remote_id:
                # 带上火山给的原因(submit_asset_for_review 会写进 review_error),
                # 否则用户只看到「提交审核失败」,不知道是图不合规、拉不到图还是服务抽风。
                asset.refresh_from_db(fields=["review_error"])
                reason = (asset.review_error or "").strip()
                detail = f"{REVIEW_SUBMIT_FAILED}({reason})" if reason else REVIEW_SUBMIT_FAILED
                raise ValueError(f"「{label}」{detail}")
            if asset.review_status == "processing" and asset.review_remote_id:
                poll_asset_review(asset)
                asset.refresh_from_db(fields=["review_status", "review_remote_id", "review_error"])
        if asset.review_status == "active" and asset.review_remote_id:
            states.append("ready")
        elif asset.review_status == "failed":
            states.append("failed")
        else:
            states.append("pending")
    if any(state == "failed" for state in states):
        return "failed"
    if all(state == "ready" for state in states):
        return "ready"
    return "pending"


def _refresh_replace_refs(team, references: list) -> list:
    """同一团队走 source=asset;官方跨团队素材把过审 id 写成 resolved_url=asset://。"""
    out = []
    for ref in references or []:
        item = dict(ref)
        raw_id = item.get("asset_id")
        if not raw_id:
            out.append(item)
            continue
        try:
            parsed = uuid.UUID(str(raw_id))
        except (TypeError, ValueError):
            out.append(item)
            continue
        try:
            asset = _load_replace_asset(team, parsed, item.get("label") or "参考素材")
        except ValueError:
            out.append(item)
            continue
        if asset.team_id == team.id:
            item["source"] = "asset"
            item.pop("resolved_url", None)
        elif asset.review_remote_id:
            item["source"] = "upload"
            item["resolved_url"] = f"asset://{asset.review_remote_id}"
        out.append(item)
    return out


def _assert_replace_refs_ready(references: list) -> None:
    missing = []
    for ref in references or []:
        raw_id = ref.get("asset_id")
        if not raw_id:
            continue
        try:
            parsed = uuid.UUID(str(raw_id))
        except (TypeError, ValueError):
            continue
        asset = Asset.objects.filter(id=parsed).first()
        if asset is None or not asset.review_remote_id or asset.review_status != "active":
            missing.append(ref.get("label") or "参考素材")
    if missing:
        raise ValueError("素材尚未完成合规审核，请稍后再试")


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 replace_duration_range() -> tuple[int, int]:
    """复刻模型(Seedance 2.5)支持的出片时长区间。模型没配好时回落 4–15,不至于炸。"""
    model = (
        ModelConfig.objects.filter(
            name=REPLACE_MODEL, capability=ModelConfig.Capability.VIDEO, status=ModelConfig.Status.ACTIVE
        ).first()
    )
    return model_duration_range(model)


def _output_duration(requested, video_seconds: float) -> int:
    low, high = replace_duration_range()
    try:
        value = int(requested) if requested not in (None, "") else 0
    except (TypeError, ValueError):
        value = 0
    if value:
        return min(high, max(low, value))
    if video_seconds:
        return min(high, max(low, int(round(video_seconds))))
    return high


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": "asset",
        "asset_id": str(asset.id),
    }
    seconds = _asset_duration_seconds(asset)
    if seconds:
        ref["duration"] = seconds
    return ref


def _image_refs(references) -> list:
    """出片/审核只用图。用户原片哪怕还躺在旧 payload 里,也一律丢掉。"""
    return [
        dict(item)
        for item in (references or [])
        if (item or {}).get("type") != "video"
    ]


def _seedance_references(payload: dict) -> list:
    """出片只带商品图/角色图。用户上传的原片不传 Seedance,避免素材审核报错。"""
    return _image_refs(payload.get("references") or [])


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_triview_asset(team, product_id: uuid.UUID):
    """商品库里这个商品的三视图(白底多角度单张 16:9)。没有就返回 None,不挡生成。"""
    return (
        Asset.objects.filter(
            team=team,
            is_deleted=False,
            purged_at__isnull=True,
            metadata__product_id=str(product_id),
            metadata__view="three_view",
        )
        .order_by("-created_at")
        .first()
    )


def _product_semantic_facts_for_id(team, product_id: uuid.UUID) -> dict:
    product = (
        Product.objects.filter(id=product_id, team=team, purged_at__isnull=True, status=Product.Status.ACTIVE)
        .prefetch_related("selling_points")
        .first()
    )
    if product is None:
        raise ValueError("商品不存在或已被删除")
    return _product_semantic_facts(product)


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("商品不存在或已被删除")
    triview = _product_triview_asset(team, product_id)
    # 三视图也占一张参考图的名额,商品图要给它让位,否则火山那边超 9 张直接拒。
    image_budget = MAX_IMAGES - 1 if triview is not None else MAX_IMAGES
    assets = []
    seen = {triview.id} if triview is not None else 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) >= image_budget:
            break
    if not assets and product.cover_asset_id and product.cover_asset_id not in seen and not product.cover_asset.is_deleted:
        assets.append(product.cover_asset)
    if not assets:
        # 只有三视图、没有任何实拍图时,把三视图顶上来当 @目标商品。
        # 否则提示词里的 @目标商品 找不到同名 label,火山拿到的是一句没有指代的字面量。
        if triview is None:
            raise ValueError("这个商品还没有可用图片")
        assets = [triview]
        triview = None
    refs = [_library_image_ref(asset, team=team, label="目标商品" if index == 0 else f"目标商品{index + 1}") for index, asset in enumerate(assets)]
    if triview is not None:
        # 放最后:@目标商品 仍指向第一张实拍图,三视图作为「各面长什么样」的补充证据。
        refs.append(_library_image_ref(triview, team=team, label=TRIVIEW_LABEL))
    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"