yingqing/core/backend/apps/ai/services.py

import json
import re
import subprocess
import tempfile
import uuid
from datetime import timedelta
from decimal import Decimal
from io import BytesIO
from pathlib import Path
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist
from django.db import transaction
from django.utils import timezone

from apps.ai.models import AITask, ModelConfig
from apps.ai.providers import (
    OpenAICompatibleProvider,
    TtsNotConfigured,
    VolcanoArkProvider,
    VolcanoTtsProvider,
)
from apps.assets.models import Asset, AssetFile
from apps.assets.storage import TosStorage
from apps.billing.services.ledger import charge_reserved_credit, release_credit, reserve_credit
from apps.projects.models import (
    BaseAssetGroup,
    ExportJob,
    ProjectStage,
    ScriptSegment,
    ScriptVersion,
    StoryboardFrame,
    StoryboardVersion,
    Timeline,
    VideoSegment,
    VideoSegmentVersion,
)


def get_default_model(capability: str) -> ModelConfig:
    qs = (
        ModelConfig.objects.select_related("provider")
        .filter(capability=capability, status=ModelConfig.Status.ACTIVE, provider__status="active")
    )
    # 优先平台超管钦定的默认模型;未钦定则回落「最早创建的 active」(原行为,零回归)
    return qs.filter(is_default=True).order_by("created_at").first() or qs.order_by("created_at").first()


# 火山官方直连(SeeDream 生图 / Seedance 视频 / 豆包文本)走 ARK SDK;其余 provider 一律
# 视为「OpenAI 兼容中转站」走通用适配器。加/换中转站 = DB 加一行 ModelProvider,零改代码。
# 注意:DB 里火山 provider 实际命名为 "volcengine"(豆包),必须包含,否则会被错路由到中转站。
OFFICIAL_DIRECT_PROVIDERS = {"volcengine", "volcano", "ark", "volcano_ark"}


def resolve_provider_credentials(provider) -> tuple[str | None, str | None]:
    """解析中转站凭证。可插拔顺序:DB(ModelProvider.base_url/api_key)优先 → settings(.env)回退。
    两者都不写死;换站只改 DB 这一行,或改 .env 对应项。"""
    base_url = (provider.base_url or "").strip() or settings.PROVIDER_BASE_URLS.get(provider.name)
    api_key = (getattr(provider, "api_key", "") or "").strip() or settings.PROVIDER_KEYS.get(provider.name)
    return (base_url or None), (api_key or None)


def build_provider(model_config: ModelConfig):
    """按 provider.name 分流:火山官方直连 → VolcanoArkProvider;其余 → 通用 OpenAICompatibleProvider。
    两条路都走 resolve_provider_credentials,统一 DB→.env 优先级(官方直连的 None 再由 __post_init__ 回退 settings.VOLCANO)。"""
    provider = model_config.provider
    base_url, api_key = resolve_provider_credentials(provider)
    if provider.name in OFFICIAL_DIRECT_PROVIDERS:
        return VolcanoArkProvider(base_url=base_url, api_key=api_key)
    # api_version:某些中转站(yunqi)的 images/edits 需 Azure 风格 ?api-version=...。
    # DB(ModelProvider.metadata.api_version)优先 → settings(.env)回退,与凭证同款可插拔。
    api_version = (provider.metadata or {}).get("api_version") or settings.PROVIDER_API_VERSIONS.get(provider.name)
    return OpenAICompatibleProvider(base_url=base_url, api_key=api_key, api_version=api_version or None)


def get_image_provider(model_config: ModelConfig):
    return build_provider(model_config)


def get_text_provider(model_config: ModelConfig):
    return build_provider(model_config)


def get_video_provider(model_config: ModelConfig):
    return build_provider(model_config)


def estimate_cost(model_config: ModelConfig) -> Decimal:
    return model_config.unit_price if model_config.unit_price > 0 else Decimal("1.0000")


def parse_segment_fields(block: str) -> tuple[str, str]:
    """从一镜文本里拆出(旁白, 画面)。

    模型按 build_script_prompt 的格式输出「旁白：…/画面：…」标签行时精确拆分;
    自带脚本/旧格式没有标签则两个字段都用整段(保持旧行为),字幕/故事板各自兜底。
    """
    narration_lines: list[str] = []
    visual_lines: list[str] = []
    current: list[str] | None = None
    for raw in (block or "").splitlines():
        line = raw.strip()
        if not line:
            continue
        matched = re.match(r"^(旁白|口播|台词|文案)\s*[:：]\s*(.*)$", line)
        if matched:
            current = narration_lines
            if matched.group(2):
                current.append(matched.group(2))
            continue
        matched = re.match(r"^(画面|镜头描述|视觉|画面描述)\s*[:：]\s*(.*)$", line)
        if matched:
            current = visual_lines
            if matched.group(2):
                current.append(matched.group(2))
            continue
        if re.match(r"^(镜头|分镜|场)\s*\d+", line):
            continue  # 「镜头N」标题行不计入任何字段
        if current is not None:
            current.append(line)
    narration = " ".join(narration_lines).strip()
    visual = " ".join(visual_lines).strip()
    if not narration and not visual:
        return block.strip(), block.strip()
    return narration or visual, visual or narration


def build_cast_scene_extract_prompt(content: str) -> list[dict[str, str]]:
    """轻量抽取提示词:从镜头脚本里提炼人物 / 场景标签,并给每个标签一句可直接生图的画面提示词。"""
    system = (
        "你是短视频脚本分析助手。请从给定的镜头脚本中提取出现的『人物』和『场景』,"
        "并为每个人物 / 场景写一句可直接用于文生图的画面提示词(中文,30 字内,描述外形 / 着装 / 环境 / 光线)。"
        "人物提示词用 9:16 竖屏(出镜角色全身);场景提示词用 16:9 横屏(环境 / 背景空镜)。"
        "人物指出镜的角色(例:女主、同事、闺蜜);场景指画面发生的地点或环境(例:卫生间、地铁、办公室)。"
        "去重,人物与场景各最多 6 个。只输出一个 JSON 对象,不要 markdown 代码块,不要任何额外文字,格式如下：\n"
        '{"cast":[{"name":"女主","prompt":"26岁都市女性,自然妆容,米色针织衫,柔和室内光,9:16竖屏"}],'
        '"scenes":[{"name":"卫生间","prompt":"现代简约浴室,暖色灯光,干净台面,16:9横屏"}]}'
    )
    return [{"role": "system", "content": system}, {"role": "user", "content": f"镜头脚本如下：\n{content}".strip()}]


def _coerce_tag_entries(items: object, limit: int = 6) -> tuple[list[str], dict[str, str]]:
    """把模型回的 [{"name","prompt"}] 列表整理成 (标签列表, {标签: 提示词}),去重保序、容错。"""
    names: list[str] = []
    prompts: dict[str, str] = {}
    if not isinstance(items, list):
        return names, prompts
    for item in items:
        if isinstance(item, dict):
            name = str(item.get("name") or "").strip()
            prompt = str(item.get("prompt") or "").strip()
        else:
            name, prompt = str(item or "").strip(), ""
        if not name or name in ("无", "暂无", "未提及") or name in names:
            continue
        names.append(name)
        if prompt:
            prompts[name] = prompt
        if len(names) >= limit:
            break
    return names, prompts


def extract_cast_and_scenes(*, project, user, content: str) -> dict:
    """轻量调一次文本模型,从脚本里抽取人物 / 场景标签及每个标签的建议生图提示词。

    出稿后的增益步骤(对齐流程文档「自动从脚本提取信息」),走完整的 AITask + 计费闭环
    (reserve→charge/release,任务类型记为 script_optimization)。但全程 best-effort:
    无可用模型 / 预扣失败 / 调用失败 / 解析失败都吞掉返回空,绝不阻断脚本生成主流程。
    返回 {cast, scenes, cast_prompts, scene_prompts}。
    """
    empty = {"cast": [], "scenes": [], "cast_prompts": {}, "scene_prompts": {}}
    model_config = get_default_model(ModelConfig.Capability.TEXT)
    if model_config is None or not (content or "").strip():
        return empty

    messages = build_cast_scene_extract_prompt(content)
    try:
        task = create_ai_task(
            project=project,
            user=user,
            task_type=AITask.Type.SCRIPT_OPTIMIZATION,
            model_config=model_config,
            request_payload={"model": model_config.name, "endpoint": model_config.endpoint, "messages": messages},
        )
    except Exception:
        return empty  # 余额不足等预扣失败:跳过提取,不挡出稿
    reservation = task.credit_reservation

    try:
        task.status = AITask.Status.SUBMITTED
        task.submitted_at = timezone.now()
        task.save(update_fields=["status", "submitted_at", "updated_at"])

        provider = build_provider(model_config)
        response = provider.chat_completion(model=model_config.name, endpoint=model_config.endpoint, messages=messages)
        text = provider.extract_text(response)

        # LLM 调用已真实消耗 token → 按成功计费(无论后续能否解析出标签)
        with transaction.atomic():
            task.status = AITask.Status.SUCCEEDED
            task.response_payload = response
            task.actual_cost = task.estimated_cost
            task.completed_at = timezone.now()
            task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
            charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)

        match = re.search(r"\{.*\}", text, re.DOTALL)  # 容忍模型多裹了 markdown / 解释文字
        if not match:
            return empty
        data = json.loads(match.group(0))
        cast, cast_prompts = _coerce_tag_entries(data.get("cast"))
        scenes, scene_prompts = _coerce_tag_entries(data.get("scenes"))
        return {"cast": cast, "scenes": scenes, "cast_prompts": cast_prompts, "scene_prompts": scene_prompts}
    except Exception as exc:
        with transaction.atomic():
            task.status = AITask.Status.FAILED
            task.error_message = str(exc)
            task.completed_at = timezone.now()
            task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
            release_credit(reservation=reservation, reason=str(exc))
        return empty


def _skills_root() -> "Path":
    """skills 根目录。优先 BASE_DIR/skills(= core/backend/skills,随后端打进 Docker 镜像);
    回落仓库根 BASE_DIR.parent.parent/skills(本地/旧布局)。

    ⚠️ 历史坑:镜像由 `./core/backend` 构建,skills 旧在仓库根 → 不在构建上下文 → 镜像里没有 →
    `_load_skill_system_prompt` 返回空串 → 提取拿不到「只输出 JSON」铁律 → 模型吐散文 → 解析失败。
    现 skills 已挪进 core/backend 随镜像打包;此处仍保留双路径兜底。"""
    from pathlib import Path

    from django.conf import settings

    base = Path(settings.BASE_DIR)
    for cand in (base / "skills", base.parent.parent / "skills"):
        if cand.is_dir():
            return cand
    return base / "skills"


def _load_skill_system_prompt(name: str) -> str:
    """读取 skills/<name>/SKILL.md + references/*.md 拼成系统提示词(领域知识)。缺文件返回空串,不致命。"""
    skill_dir = _skills_root() / name
    parts: list[str] = []
    main = skill_dir / "SKILL.md"
    if main.exists():
        parts.append(main.read_text(encoding="utf-8"))
    ref_dir = skill_dir / "references"
    if ref_dir.exists():
        for ref in sorted(ref_dir.glob("*.md")):
            parts.append(f"\n\n===== references/{ref.name} =====\n\n{ref.read_text(encoding='utf-8')}")
    return "\n".join(parts)


def _normalize_extracted_entities(items: object) -> list[dict]:
    """整理提取出的 entities:只留 character/scene(绝不收 product),保留模型给的 id(供 segment refs 对应),
    按 id 去重,角色/场景各最多 6 个,补 ref_index。"""
    out: list[dict] = []
    if not isinstance(items, list):
        return out
    seen_ids: set[str] = set()
    n_char = n_scene = 0
    for it in items:
        if not isinstance(it, dict):
            continue
        typ = str(it.get("type") or "").strip()
        if typ not in ("character", "scene"):  # 丢弃 product / 未知类型
            continue
        name = str(it.get("name") or "").strip()
        eid = str(it.get("id") or "").strip()
        if not name or not eid or eid in seen_ids:
            continue
        if typ == "character":
            n_char += 1
            if n_char > 6:
                continue
        else:
            n_scene += 1
            if n_scene > 6:
                continue
        seen_ids.add(eid)
        out.append(
            {
                "id": eid,
                "type": typ,
                "name": name,
                "visual_prompt": str(it.get("visual_prompt") or "").strip(),
                "ref_index": len(out) + 1,
            }
        )
    return out


def _normalize_extracted_segment_refs(items: object, valid_ids: set[str]) -> list[dict]:
    """整理每镜 refs:只留指向存活实体(角色/场景)的 id,丢掉被剔除的(如商品/无效 id)。"""
    out: list[dict] = []
    if not isinstance(items, list):
        return out
    for it in items:
        if not isinstance(it, dict):
            continue
        idx = it.get("index")
        if not isinstance(idx, int):
            continue
        raw = it.get("entity_refs") or it.get("refs") or []
        refs = [r for r in raw if isinstance(r, str) and r in valid_ids]
        out.append({"index": idx, "entity_refs": refs})
    return out


# 提取步固定锁定豆包 2.0 Pro(与脚本生成同款),不靠 get_default_model 的「最早创建」排序——
# 避免不同环境 DB 创建序漂移把提取路由到别的(中转站)推理模型。取不到再回落默认文本模型。
EXTRACT_TEXT_MODEL_NAME = "doubao-seed-2-0-pro-260215"


def _resolve_extract_model_config():
    """提取实体用的文本模型:优先豆包 2.0 Pro(active 且 provider active),否则回落默认文本模型。"""
    pinned = (
        ModelConfig.objects.select_related("provider")
        .filter(
            name=EXTRACT_TEXT_MODEL_NAME,
            capability=ModelConfig.Capability.TEXT,
            status=ModelConfig.Status.ACTIVE,
            provider__status="active",
        )
        .order_by("created_at")
        .first()
    )
    return pinned or get_default_model(ModelConfig.Capability.TEXT)


def _collect_extract_text(provider, model_config, messages) -> tuple[str, dict]:
    """走与「脚本生成」同一条已在生产验证稳定的流式通道把模型输出收全。

    豆包 seed-pro / GPT / Gemini 等思考模型,思考期只发 reasoning_content、正文期才发 content,
    流式分支(chat_completion_stream)已分别转发为 `reasoning` / `delta` 事件。这里**只收正文 delta**;
    仅当正文为空(整轮被思考占满的极端兜底)才回退已收到的 reasoning。如此从根上避免非流式只读
    content 拿到空串 → 解析不到 JSON 的老问题。返回 (text, 存档用精简 payload)。
    """
    content_parts: list[str] = []
    reasoning_parts: list[str] = []
    for ev in provider.chat_completion_stream(
        model=model_config.name,
        endpoint=model_config.endpoint,
        messages=messages,
        temperature=0.3,  # 结构化抽取要稳:低温降低 JSON 漂移
    ):
        etype = ev.get("type")
        if etype == "delta":
            content_parts.append(ev.get("text") or "")
        elif etype == "reasoning":
            reasoning_parts.append(ev.get("text") or "")
    content = "".join(content_parts).strip()
    reasoning = "".join(reasoning_parts).strip()
    text = content or reasoning
    payload = {
        "streamed": True,
        "model": model_config.name,
        "content": content,
        "reasoning_chars": len(reasoning),
        "reasoning_preview": reasoning[:500],  # 失败时看一眼模型在想啥/有没有跑题
        "system_chars": sum(len(m.get("content") or "") for m in messages if m.get("role") == "system"),
    }
    return text, payload


# 在途状态:据此判「已有提取在跑」(提交侧防重复扣费 + 前端刷新后重建 loading)
_EXTRACT_INFLIGHT = (
    AITask.Status.CREATED,
    AITask.Status.RESERVED,
    AITask.Status.SUBMITTED,
    AITask.Status.POLLING,
)


def get_inflight_extraction(project):
    """本项目正在跑的实体提取任务(最近一条优先);无则 None。"""
    return (
        AITask.objects.filter(
            project=project, task_type=AITask.Type.ENTITY_EXTRACTION, status__in=_EXTRACT_INFLIGHT
        )
        .order_by("-created_at")
        .first()
    )


def submit_extract_entities(*, project, user) -> AITask:
    """提交独立实体提取步(**异步**)。读已定稿(或最新)脚本 → 校验 + 建 RESERVED 任务 + 预留额度(秒级),
    慢活(豆包思考模型流式抽取,可达数十秒)交给 Celery worker(run_extract_entities_task)跑。

    这样 Web 层(gunicorn/nginx)不被数十秒的模型请求占住 → 不再 502。
    **防重复扣费**:本项目已有在途提取任务时直接复用它,不新建/不二次预扣
    (用户刷新后手痒重点、或并发点击都安全 —— 提取是实体唯一权威来源,本就只需跑一次)。
    校验失败(无脚本/无分镜/无模型/额度不足)抛 ValueError 由端点转 400;运行期失败由 worker 记进
    task.error_message,前端轮询 extract-status 读取。返回 AITask。"""
    from apps.projects.models import ScriptVersion
    from apps.ai.tasks import extract_entities_task

    script = (
        ScriptVersion.objects.filter(project=project, is_adopted=True).order_by("-created_at").first()
        or ScriptVersion.objects.filter(project=project).order_by("-created_at").first()
    )
    if script is None:
        raise ValueError("请先生成并定稿脚本,再提取角色 / 场景")
    segments = list(script.segments.order_by("sort_order"))
    if not segments:
        raise ValueError("脚本没有分镜,无法提取")

    inflight = get_inflight_extraction(project)
    if inflight is not None:
        return inflight  # 已有提取在跑:复用,绝不二次预扣 / 重复出活

    model_config = _resolve_extract_model_config()
    if model_config is None:
        raise ValueError("没有可用的文本模型")

    product = project.product
    if product is not None:
        sp = "、".join([p.title for p in product.selling_points.all()[:5]])
        prod_line = f"名称：{product.title}\n品类：{product.category or '未填'}\n卖点：{sp or '未填'}"
    else:
        prod_line = "（无商品信息）"
    seg_lines = [
        f"镜{i}  role={s.role or ''}  narration={(s.narration or '').strip()}  visual={(s.visual_prompt or '').strip()}"
        for i, s in enumerate(segments)
    ]
    user_msg = (
        f"商品信息：\n{prod_line}\n\n分镜脚本（共 {len(segments)} 镜，index 从 0 开始）：\n" + "\n".join(seg_lines)
    )
    system = _load_skill_system_prompt("ecommerce-entity-extract")
    messages = [{"role": "system", "content": system}, {"role": "user", "content": user_msg}]

    try:
        task = create_ai_task(
            project=project,
            user=user,
            task_type=AITask.Type.ENTITY_EXTRACTION,
            model_config=model_config,
            request_payload={
                "model": model_config.name,
                "endpoint": model_config.endpoint,
                "messages": messages,
                "script_id": str(script.id),
            },
        )
    except Exception as exc:  # 余额不足等预扣失败
        raise ValueError("额度不足,无法提取(请先充值)") from exc

    extract_entities_task.delay(str(task.id))
    return task


def run_extract_entities_task(*, task_id: str) -> None:
    """Celery worker 内执行实体提取慢活:流式调模型 → 解析 JSON → 落库(覆盖 project.metadata 的
    cast/scenes/*_prompts/script_entities + 回填每镜 entity_refs + entities_extracted 标记)并扣费;
    失败退费并把可读错误记进 task.error_message(前端轮询 extract-status 读取)。
    幂等:只处理 RESERVED 任务,重复投递不会二次出活、二次扣费。"""
    from apps.projects.models import ScriptVersion
    from apps.ai.script_agent import _map_entities_to_project_metadata

    task = AITask.objects.select_related("team", "created_by", "project", "model_config").filter(id=task_id).first()
    if task is None or task.status != AITask.Status.RESERVED:
        return
    project = task.project
    payload = task.request_payload or {}
    messages = payload.get("messages") or []
    model_config = task.model_config
    reservation = task.credit_reservation
    response: dict = {}  # 模型返回的精简存档;失败时也落库(便于事后定位"模型到底吐了啥")
    try:
        task.status = AITask.Status.SUBMITTED
        task.submitted_at = timezone.now()
        task.save(update_fields=["status", "submitted_at", "updated_at"])

        provider = build_provider(model_config)
        text, response = _collect_extract_text(provider, model_config, messages)

        match = re.search(r"\{.*\}", text, re.DOTALL)
        if not match:
            raise ValueError("提取结果解析失败(模型未返回有效 JSON),请重试")
        try:
            data = json.loads(match.group(0))
        except json.JSONDecodeError as exc:  # 抓出来给用户可读话术,别把裸解析报错塞进 error_message
            raise ValueError("提取结果解析失败,请重试") from exc
        entities = _normalize_extracted_entities(data.get("entities"))
        if not entities:
            raise ValueError("没有从脚本里识别到角色 / 场景,可调整脚本后重试")
        seg_refs = _normalize_extracted_segment_refs(data.get("segments"), {e["id"] for e in entities})

        script_id = payload.get("script_id")
        script = ScriptVersion.objects.filter(id=script_id).first() if script_id else None
        if script is None:  # 极端兜底:payload 里 script 没了就回落项目当前定稿稿
            script = (
                ScriptVersion.objects.filter(project=project, is_adopted=True).order_by("-created_at").first()
                or ScriptVersion.objects.filter(project=project).order_by("-created_at").first()
            )
        segments = list(script.segments.order_by("sort_order")) if script is not None else []

        with transaction.atomic():
            task.status = AITask.Status.SUCCEEDED
            task.response_payload = response
            task.actual_cost = task.estimated_cost
            task.completed_at = timezone.now()
            task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
            charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)

            # 落库:覆盖 project.metadata + 回填每镜 entity_refs(提取为实体唯一权威来源)。
            # entities_extracted 标记:前端资产页提取闸门据此显隐(没走过 → 盖蒙版露按钮;走过 → 露卡片)。
            _map_entities_to_project_metadata(project, entities)
            md_flag = dict(project.metadata or {})
            md_flag["entities_extracted"] = True
            project.metadata = md_flag
            project.save(update_fields=["metadata", "updated_at"])
            refs_by_index = {r["index"]: r["entity_refs"] for r in seg_refs}
            for i, seg in enumerate(segments):
                new_refs = refs_by_index.get(i, [])
                if (seg.entity_refs or []) != new_refs:
                    seg.entity_refs = new_refs
                    seg.save(update_fields=["entity_refs", "updated_at"])
    except Exception as exc:  # noqa: BLE001 — 失败退费并把错误记进 AITask 供前端轮询;不向上抛(避免 celery 重试二次扣费)
        # ValueError 是我们给用户写好的可读话术(解析失败 / 没识别到角色等);其余(网络/模型异常)给通用话术。
        msg = str(exc) if isinstance(exc, ValueError) else "提取调用失败,请重试"
        with transaction.atomic():
            task.status = AITask.Status.FAILED
            task.error_message = msg
            task.response_payload = response  # 即便失败也存下模型输出片段(content/reasoning 长度等),供事后定位
            task.completed_at = timezone.now()
            task.save(update_fields=["status", "response_payload", "error_message", "completed_at", "updated_at"])
            release_credit(reservation=reservation, reason=msg)


def split_script_into_segments(content: str, count: int = 4) -> list[str]:
    """把一段脚本稳健地拆成 `count` 个分镜文本,保证每镜都非空、且所有内容都被分配到某一镜。

    原实现按行 `[:4]`,ARK 返回整段散文时常变成「第1镜有词、2/3/4镜全空」,
    导致后续故事板帧 / 视频段拿到空提示词,前后内容断裂。这里改为:
    优先按空行/标号块切,块数够就把全部块均匀分桶;块不够再按句子切;仍不够则补齐。
    """

    def _bucketize(items: list[str], joiner: str) -> list[str]:
        buckets: list[list[str]] = [[] for _ in range(count)]
        per = len(items) / count
        for index, item in enumerate(items):
            buckets[min(count - 1, int(index / per))].append(item)
        return [joiner.join(bucket).strip() for bucket in buckets]

    text = (content or "").strip()
    if not text:
        return [""] * count

    # 1) 优先按空行分段;只有一段时退回按行分
    blocks = [block.strip() for block in re.split(r"\n\s*\n", text) if block.strip()]
    if len(blocks) < 2:
        blocks = [line.strip() for line in text.splitlines() if line.strip()]
    if len(blocks) >= count:
        return _bucketize(blocks, "\n")

    # 2) 段落不足:按中英文句末标点切句,再均匀分桶
    sentences = [s.strip() for s in re.split(r"(?<=[。！？!?.;；\n])", text) if s.strip()]
    if len(sentences) >= count:
        return _bucketize(sentences, " ")

    # 3) 仍不足:用已有块/句补齐到 count,绝不留空镜
    base = blocks or sentences or [text]
    filled = list(base)
    while len(filled) < count:
        filled.append(base[-1])
    return filled[:count]


@transaction.atomic
def create_ai_task(*, project, user, task_type: str, model_config: ModelConfig, request_payload: dict) -> AITask:
    cost = estimate_cost(model_config)
    task = AITask.objects.create(
        team=project.team,
        created_by=user,
        project=project,
        task_type=task_type,
        status=AITask.Status.CREATED,
        model_config=model_config,
        idempotency_key=f"{task_type}:{project.id}:{uuid.uuid4()}",
        request_payload=request_payload,
        estimated_cost=cost,
    )
    reserve_credit(team=project.team, user=user, task=task, amount=cost)
    task.status = AITask.Status.RESERVED
    task.save(update_fields=["status", "updated_at"])
    return task


def regenerate_script_segment(*, project, user, segment, instruction: str = "") -> ScriptVersion:
    """单镜重跑(「场次刷新」按钮):复用脚本 agent 的精准改一镜——读全脚本上下文、只动该镜、保留其余镜,落新 ScriptVersion。
    旧的「散文 prompt + 正则解析」整条已废,统一走 agent(结构化 + entity_refs 不丢)。"""
    from apps.ai.script_agent import regenerate_segment_via_agent

    model_config = get_default_model(ModelConfig.Capability.TEXT)
    if model_config is None:
        raise ValueError("no active text model configured")
    return regenerate_segment_via_agent(
        project=project, user=user, model_config=model_config, segment=segment, instruction=instruction
    )


def _generate_video_poster(*, video_bytes: bytes, team, project, asset_id) -> "StoredObject | None":
    """用 ffmpeg 抽视频首帧作为封面(poster)并上传 TOS。best-effort:任何失败都返回 None,不影响视频资产落地。"""
    if not video_bytes:
        return None
    try:
        with tempfile.TemporaryDirectory(prefix="airshelf-poster-") as tmp:
            tmp_dir = Path(tmp)
            video_path = tmp_dir / "in.mp4"
            poster_path = tmp_dir / "poster.jpg"
            video_path.write_bytes(video_bytes)
            proc = subprocess.run(
                ["ffmpeg", "-y", "-ss", "0", "-i", str(video_path), "-frames:v", "1", "-q:v", "3", str(poster_path)],
                capture_output=True,
                timeout=60,
            )
            if proc.returncode != 0 or not poster_path.exists():
                return None
            poster_bytes = poster_path.read_bytes()
            if not poster_bytes:
                return None
            object_key = f"teams/{team.id}/projects/{project.id}/generated/{asset_id}-poster.jpg"
            return TosStorage().upload_fileobj(
                fileobj=BytesIO(poster_bytes), object_key=object_key, content_type="image/jpeg"
            )
    except Exception:  # noqa: BLE001 — poster 仅用于展示,失败不阻断
        return None


def _store_generated_media(*, team, user, project, task, media: str, name: str, category: str, asset_type: str) -> Asset:
    fileobj, content_type = VolcanoArkProvider.media_to_bytes(media)
    suffix = ".png"
    if "video" in content_type:
        suffix = ".mp4"
    elif "jpeg" in content_type:
        suffix = ".jpg"
    elif "webp" in content_type:
        suffix = ".webp"
    asset_id = uuid.uuid4()
    object_key = f"teams/{team.id}/projects/{project.id}/generated/{asset_id}{suffix}"
    stored = TosStorage().upload_fileobj(fileobj=fileobj, object_key=object_key, content_type=content_type)
    asset = Asset.objects.create(
        id=asset_id,
        team=team,
        created_by=user,
        name=name,
        asset_type=asset_type,
        source=Asset.Source.AI_GENERATED,
        category=category,
        origin_task=task,
    )
    AssetFile.objects.create(
        asset=asset,
        object_key=stored.object_key,
        bucket=stored.bucket,
        content_type=stored.content_type,
        size_bytes=stored.size_bytes,
        is_primary=True,
    )
    # 视频资产:额外抽首帧作为封面图,挂成同一 Asset 下的 image 文件,供任务中心/列表显示缩略图
    if "video" in content_type:
        try:
            video_bytes = fileobj.getvalue() if isinstance(fileobj, BytesIO) else b""
        except Exception:  # noqa: BLE001
            video_bytes = b""
        poster = _generate_video_poster(video_bytes=video_bytes, team=team, project=project, asset_id=asset_id)
        if poster:
            AssetFile.objects.create(
                asset=asset,
                object_key=poster.object_key,
                bucket=poster.bucket,
                content_type=poster.content_type,
                size_bytes=poster.size_bytes,
                is_primary=False,
            )
    return asset


def _find_entity_group(project, kind: str, label: str, group_id: str | None):
    """复用同一实体的基础资产组(版本=candidate_assets,采用=adopted_asset)。
    优先 group_id;商品=该项目唯一商品组;人物/场景按 label 命中非三视图组;都没有则 None(新建)。"""
    if group_id:
        return project.base_asset_groups.filter(id=group_id).first()
    candidates = list(project.base_asset_groups.filter(kind=kind).order_by("created_at"))
    candidates = [g for g in candidates if not (g.metadata or {}).get("triview_of")]  # 三视图组不算实体本体
    if kind == BaseAssetGroup.Kind.PRODUCT:
        return candidates[0] if candidates else None
    lbl = (label or "").strip()
    if lbl:
        return next((g for g in candidates if (g.metadata or {}).get("label") == lbl), None)
    return None


def _ratio_to_image_size(ratio: str) -> str:
    """前端比例(1:1 / 3:4 / 9:16)→ gpt-image 支持的尺寸。竖屏统一 1024x1536,横屏 1536x1024。"""
    return {
        "1:1": "1024x1024",
        "3:4": "1024x1536",
        "9:16": "1024x1536",
        "4:3": "1536x1024",
        "16:9": "1536x1024",
    }.get((ratio or "").strip(), "1024x1024")


def _product_cover_url(product) -> str:
    """商品主图 URL:优先 cover_asset,其次标记为主图的商品图,再次首张商品图。无图返回 ''。"""
    if product is None:
        return ""
    if product.cover_asset_id:
        url = _asset_preview_url(product.cover_asset)
        if url:
            return url
    image = product.images.filter(is_primary=True).first() or product.images.order_by("sort_order", "created_at").first()
    if image is not None:
        return _asset_preview_url(image.asset)
    return ""


def quality_words(stage: str, slot: str = "quality") -> list[str]:
    """平台单层质量词配置(QualityWord)。无配置/表不存在 → 返回 [],调用方回落写死值,保证零回归。"""
    try:
        from apps.ai.models import QualityWord

        return list(
            QualityWord.objects.filter(stage=stage, slot=slot, enabled=True)
            .order_by("sort", "created_at")
            .values_list("text", flat=True)
        )
    except Exception:  # noqa: BLE001 — 配置读取失败绝不阻断生成
        return []


def quality_suffix(stage: str, fallback: str, slot: str = "quality", sep: str = "，") -> str:
    """取该阶段配置的质量词拼成尾串;无配置回落到 fallback(各 builder 的原写死值)。"""
    words = quality_words(stage, slot)
    return sep.join(words) if words else fallback


def build_product_triview_prompt_refs(product, base_prompt: str = "") -> str:
    """商品三视图 image_edit 提示词(refs 版):参考图1=商品真实主图,锁包装一致性。"""
    name = (getattr(product, "title", "") or "商品").strip()
    lines = [
        f"参考图1是「{name}」的真实商品主图。",
        "请严格参照该图的包装外形、品牌文字、配色、Logo 与材质,生成同一件商品的三视图:",
        "从左到右依次为正面、侧面、背面,统一光照,纯白背景,16:9 构图。",
        "三个视图必须是同一件商品,品牌字样/配色/外形高度一致,不要改动或重新设计包装。",
    ]
    if base_prompt and base_prompt.strip():
        lines.append(base_prompt.strip())
    return " ".join(lines)


def build_model_tryon_prompt_refs(product, has_model: bool, base_prompt: str = "") -> str:
    """模特上身图 image_edit 提示词(refs 版):
    参考图1=商品真实主图(锁商品外形/品牌/配色),参考图2=选中模特(锁人脸/身形/气质)。
    生成「该模特自然展示/使用该商品」的电商效果图。"""
    name = (getattr(product, "title", "") or "商品").strip()
    lines = [f"参考图1是「{name}」的真实商品。"]
    if has_model:
        lines += [
            "参考图2是出镜模特。请生成参考图2中的这位模特自然地展示/佩戴/使用参考图1中商品的电商效果图。",
            "模特的五官、发型、肤色、身形与气质必须与参考图2高度一致,不要换人;",
            "商品的外形、品牌文字、配色、Logo 必须与参考图1高度一致,不要改动或重新设计。",
        ]
    else:
        lines += [
            "请生成一位真人模特自然地展示/佩戴/使用参考图1中商品的电商效果图。",
            "商品的外形、品牌文字、配色、Logo 必须与参考图1高度一致,不要改动或重新设计。",
        ]
    lines.append(quality_suffix("model_tryon", "自然光、真实质感、干净背景、电商主图构图,人物与商品比例真实协调。"))
    if base_prompt and base_prompt.strip():
        lines.append(base_prompt.strip())
    return " ".join(lines)


def build_platform_cover_prompt_refs(product, has_model: bool, base_prompt: str = "") -> str:
    """平台套图 image_edit 提示词(refs 版):参考图1=商品真实主图(锁外形/品牌/配色/Logo),
    有模特时参考图2=出镜模特(锁人脸/身形)。生成电商平台主图 / 封面套图,商品须还原真实包装。"""
    name = (getattr(product, "title", "") or "商品").strip()
    lines = [f"参考图1是「{name}」的真实商品主图。"]
    if has_model:
        lines += [
            "参考图2是出镜模特。请生成参考图2中的这位模特展示参考图1中商品的电商平台套图(主图 / 封面 / 详情);",
            "模特的五官、发型、肤色、身形必须与参考图2高度一致,不要换人;",
        ]
    else:
        lines.append("请基于该商品生成电商平台套图(主图 / 封面 / 详情排版),统一视觉风格;")
    lines.append(
        "商品的外形、品牌文字、配色、Logo、材质必须与参考图1高度一致,严禁改动或重新设计包装;"
        "干净背景、电商主图构图、真实质感。"
    )
    if base_prompt and base_prompt.strip():
        lines.append(base_prompt.strip())
    return " ".join(lines)


def build_person_frontal_prompt(description: str = "") -> str:
    """人物正面氛围图提示词:把脚本提取(或用户输入)的人物描述包成统一模板。
    用户钦定格式:电商真人模特，氛围正面全身照，<描述>，自然妆容，柔和影棚光，真实质感，单人，纯色背景。"""
    desc = (description or "").strip()
    inner = f"{desc}，" if desc else ""
    tail = quality_suffix("person", "自然妆容，柔和影棚光，真实质感，单人，纯色背景")
    return f"电商真人模特，氛围正面全身照，{inner}{tail}"


def generate_base_asset(*, project, user, kind: str, prompt: str, label: str = "", group_id: str | None = None) -> AITask:
    """提交基础资产生成(**异步**):Web 请求只建 RESERVED 任务 + 预留额度(秒级),
    慢出图(文生图 / 商品 image_edit)交给 Celery worker(run_base_asset_task)跑。

    这样 Web 层(gunicorn)不被 ~30s+ 的出图请求占住 → 健康探针不饿死 → 不再"生成几张就整站 502/卡死"。
    返回 RESERVED 的 AITask,前端拿 id 轮询 /api/ai/generate-image/?ids=… 取结果;出图后刷新项目即见新组。"""
    from apps.ai.tasks import generate_base_asset_task

    model_config = get_default_model(ModelConfig.Capability.IMAGE)
    if model_config is None:
        raise ValueError("no active image model configured")
    provider = get_image_provider(model_config)
    # 商品三视图:有真实商品主图 → 走 image_edit 以主图为参考,锁定包装(品牌字/配色/外形/Logo)一致;
    # 无主图或当前模型不支持 image_edit → 回落纯文生图(仅凭商品名脑补,不保证还原真实包装)。
    product_ref_url = _product_cover_url(project.product) if kind == BaseAssetGroup.Kind.PRODUCT else ""
    use_edit = bool(product_ref_url) and hasattr(provider, "image_edit")
    if use_edit:
        gen_prompt = build_product_triview_prompt_refs(project.product, prompt)
    elif kind == BaseAssetGroup.Kind.PERSON:
        gen_prompt = build_person_frontal_prompt(prompt)  # 人物立绘:包成「电商真人模特/正面全身/纯色背景」统一模板
    else:
        gen_prompt = prompt
    payload = {
        "model": model_config.name, "endpoint": model_config.endpoint, "prompt": gen_prompt,
        "kind": kind, "label": label or "", "group_id": str(group_id) if group_id else "",
        "use_edit": use_edit, "reference_image": product_ref_url,
    }
    task = create_ai_task(
        project=project,
        user=user,
        task_type={
            BaseAssetGroup.Kind.PRODUCT: AITask.Type.PRODUCT_IMAGE,
            BaseAssetGroup.Kind.PERSON: AITask.Type.PERSON_IMAGE,
            BaseAssetGroup.Kind.SCENE: AITask.Type.SCENE_IMAGE,
        }[kind],
        model_config=model_config,
        request_payload=payload,
    )
    generate_base_asset_task.delay(str(task.id))
    return task


def run_base_asset_task(*, task_id: str) -> None:
    """Celery worker 内执行基础资产的慢出图:调模型 → 成功落库扣费并归组 / 失败退费。
    幂等:只处理 RESERVED 任务,重复投递不会二次出图、二次扣费。"""
    task = AITask.objects.select_related("team", "created_by", "project", "model_config").filter(id=task_id).first()
    if task is None or task.status != AITask.Status.RESERVED:
        return
    project = task.project
    user = task.created_by
    payload = task.request_payload or {}
    kind = payload.get("kind")
    prompt = str(payload.get("prompt") or "")
    label = str(payload.get("label") or "")
    group_id = payload.get("group_id") or None
    use_edit = bool(payload.get("use_edit"))
    ref_url = str(payload.get("reference_image") or "")
    model_config = task.model_config
    provider = get_image_provider(model_config)
    reservation = task.credit_reservation
    try:
        if use_edit and ref_url:
            response = provider.image_edit(model=model_config.name, prompt=prompt, images=[ref_url], size="1536x1024")
        else:
            # 场景 = 16:9 横构图(环境/背景空镜);人物立绘 = 9:16 竖全身。默认竖。
            gen_size = "1536x1024" if kind == BaseAssetGroup.Kind.SCENE else "1024x1536"
            response = provider.image_generation(model=model_config.name, endpoint=model_config.endpoint, prompt=prompt, size=gen_size)
        media = provider.extract_first_media_url(response)
        with transaction.atomic():
            task.status = AITask.Status.SUCCEEDED
            task.response_payload = response
            task.actual_cost = task.estimated_cost
            task.completed_at = timezone.now()
            task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
            charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
            category = {
                BaseAssetGroup.Kind.PRODUCT: Asset.Category.PRODUCT_IMAGE,
                BaseAssetGroup.Kind.PERSON: Asset.Category.PERSON,
                BaseAssetGroup.Kind.SCENE: Asset.Category.SCENE,
            }[kind]
            asset = _store_generated_media(
                team=project.team,
                user=user,
                project=project,
                task=task,
                media=media,
                name=f"{project.name}-{kind}",
                category=category,
                asset_type=Asset.Type.IMAGE,
            )
            # 复用同实体的组:追加候选 + 采用最新(版本=candidate_assets,采用=adopted_asset);无则新建
            group = _find_entity_group(project, kind, label, group_id)
            if group is None:
                group_meta = {"label": label.strip()} if label and label.strip() else {}
                group = BaseAssetGroup.objects.create(project=project, kind=kind, task=task, prompt=prompt, metadata=group_meta)
            elif prompt:
                group.prompt = prompt
            group.candidate_assets.add(asset)
            group.adopted_asset = asset
            group.save(update_fields=["adopted_asset", "prompt", "updated_at"])
            # 真人资产:事务提交后静默送火山审核(best-effort,网络调用放 on_commit 避免占着事务)
            if kind == BaseAssetGroup.Kind.PERSON:
                from apps.assets.review import submit_asset_for_review

                transaction.on_commit(lambda a=asset: submit_asset_for_review(a))
    except Exception as exc:  # noqa: BLE001 — 失败要退费并把错误记进 AITask 供前端轮询读取;不向上抛(避免 celery 重试二次扣费)
        task.status = AITask.Status.FAILED
        task.error_message = str(exc)
        task.completed_at = timezone.now()
        task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
        release_credit(reservation=reservation, reason=str(exc))


def generate_person_triview(*, project, user, portrait_asset) -> AITask:
    """流程步骤4 · 据「某一版立绘资产」生成它配套的三视图(**异步**:image_edit 慢,交给 worker)。
    Web 请求只建 RESERVED 任务 + 预留额度后秒回;worker 内跑 image_edit 并把三视图归组(run_triview_task)。
    三视图与立绘 1:1 绑定:metadata.triview_of=<立绘 asset id>;同一立绘多次=同组追加候选(版本)。"""
    from apps.ai.tasks import generate_triview_task
    from apps.ai.model_library import THREE_VIEW_PROMPT

    if portrait_asset is None:
        raise ValueError("该立绘尚未生成,无法据它生成三视图")
    asset_key = str(portrait_asset.id)
    model_config = get_default_model(ModelConfig.Capability.IMAGE)
    if model_config is None:
        raise ValueError("no active image model configured")
    provider = get_image_provider(model_config)
    if not hasattr(provider, "image_edit"):
        raise ValueError(f"当前图像模型 {model_config.provider.name}:{model_config.name} 不支持参考图三视图(image_edit)")
    ref_url = _asset_preview_url(portrait_asset)
    payload = {"model": model_config.name, "prompt": THREE_VIEW_PROMPT, "kind": "person", "triview_of": asset_key, "reference_image": ref_url}
    task = create_ai_task(project=project, user=user, task_type=AITask.Type.PERSON_IMAGE, model_config=model_config, request_payload=payload)
    generate_triview_task.delay(str(task.id))
    return task


def run_triview_task(*, task_id: str) -> None:
    """Celery worker 内执行三视图慢出图(image_edit 以立绘为参考):成功落库扣费并归到立绘的三视图组 / 失败退费。
    幂等:只处理 RESERVED 任务,重复投递不会二次出图、二次扣费。"""
    from apps.ai.model_library import THREE_VIEW_PROMPT

    task = AITask.objects.select_related("team", "created_by", "project", "model_config").filter(id=task_id).first()
    if task is None or task.status != AITask.Status.RESERVED:
        return
    project = task.project
    user = task.created_by
    payload = task.request_payload or {}
    asset_key = str(payload.get("triview_of") or "")
    ref_url = str(payload.get("reference_image") or "")
    prompt = str(payload.get("prompt") or THREE_VIEW_PROMPT)
    model_config = task.model_config
    provider = get_image_provider(model_config)
    reservation = task.credit_reservation
    try:
        response = provider.image_edit(model=model_config.name, prompt=prompt, images=[ref_url], size="1536x1024")
        media = provider.extract_first_media_url(response)
        with transaction.atomic():
            task.status = AITask.Status.SUCCEEDED
            task.response_payload = response
            task.actual_cost = task.estimated_cost
            task.completed_at = timezone.now()
            task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
            charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
            asset = _store_generated_media(
                team=project.team, user=user, project=project, task=task, media=media,
                name=f"{project.name}-三视图", category=Asset.Category.TRI_VIEW, asset_type=Asset.Type.IMAGE,
            )
            # 复用该立绘的三视图组(triview_of==立绘asset id):追加候选 + 采用最新
            group = next((g for g in project.base_asset_groups.filter(kind=BaseAssetGroup.Kind.PERSON).order_by("created_at")
                          if (g.metadata or {}).get("triview_of") == asset_key), None)
            if group is None:
                group = BaseAssetGroup.objects.create(
                    project=project, kind=BaseAssetGroup.Kind.PERSON, task=task, prompt=prompt,
                    metadata={"label": "·三视图", "triview_of": asset_key},
                )
            group.candidate_assets.add(asset)
            group.adopted_asset = asset
            group.save(update_fields=["adopted_asset", "updated_at"])
            # B 路(期3):视频流程新生成的「角色」立绘 + 三视图成套 → 自动加入模特库(团队级可复用)。
            # 幂等:同一立绘只建一次 Model;best-effort,不影响主链路。
            from apps.assets.models import Model as ModelEntity

            portrait = Asset.objects.filter(id=asset_key).first()
            if portrait is not None and not ModelEntity.objects.filter(portrait_asset=portrait).exists():
                ModelEntity.objects.create(
                    team=project.team, created_by=user,
                    name=(portrait.name or f"{project.name}-角色").split("·")[0].strip()[:255] or "角色",
                    source=ModelEntity.Source.AI,
                    portrait_asset=portrait, triview_asset=asset,
                    metadata={"from_project": str(project.id), "auto_enrolled": True},
                )
            # 合规(期3):三视图含人脸,视频生成前必须过火山审核 → 事务提交后静默送审(best-effort)
            from apps.assets.review import submit_asset_for_review

            transaction.on_commit(lambda a=asset: submit_asset_for_review(a))
    except Exception as exc:  # noqa: BLE001 — 失败退费 + 错误记进 AITask 供前端轮询读取;不向上抛(避免 celery 重试二次扣费)
        task.status = AITask.Status.FAILED
        task.error_message = str(exc)
        task.completed_at = timezone.now()
        task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
        release_credit(reservation=reservation, reason=str(exc))


def _scene_context(project) -> str:
    """从商品 + 已采用基础资产提炼一句「风格锚点」,贯穿故事板 / 视频,保证各镜内容一致。"""
    product = project.product
    parts = [f"商品:{product.title}"]
    if product.brand:
        parts.append(f"品牌:{product.brand}")
    if product.category:
        parts.append(f"类目:{product.category}")
    if getattr(product, "target_audience", ""):
        parts.append(f"人群:{product.target_audience}")
    adopted_kinds = set(
        project.base_asset_groups.filter(adopted_asset__isnull=False).values_list("kind", flat=True)
    )
    if BaseAssetGroup.Kind.PERSON in adopted_kinds:
        parts.append("真人出镜,保持人物一致")
    if BaseAssetGroup.Kind.SCENE in adopted_kinds:
        parts.append("统一场景与色调")
    return " · ".join(parts)


def _segment_script_text(segment) -> str:
    """本镜脚本文本(画面 + 口播/旁白 + 商品露出),拼进故事板/视频提示词的【分镜脚本】。"""
    parts = []
    visual = (segment.visual_prompt or "").strip()
    if visual:
        parts.append(f"画面:{visual}")
    narration = (segment.narration or "").strip()
    if narration:
        parts.append(f"口播/旁白:{narration}")
    if getattr(segment, "product_exposure", ""):
        parts.append(f"商品露出:{segment.product_exposure.strip()}")
    return "\n".join(parts)


def build_storyboard_frame_prompt(project, version, segment) -> str:
    """单镜导演故事板提示词(一镜 = 一张导演故事板,用户钦定结构)。无参考图时的文本版。"""
    dur = segment.duration_seconds or 15
    lines = [
        "根据以下脚本生成一个导演故事板,用于指导 seedance 的视频生成。",
        _scene_context(project),
        f"【分镜脚本】(本段时长约 {dur} 秒)",
        _segment_script_text(segment) or f"第 {segment.sort_order + 1} 镜",
    ]
    if version.prompt:
        lines.append(version.prompt.strip())
    lines.append(quality_suffix("storyboard", "电商竖屏 9:16 导演故事板,真实清新、生活化,画面清晰,可直接指导视频生成。"))
    return "\n".join(line for line in lines if line)


def build_video_segment_prompt(project, video_segment, scene, refs, user_prompt: str = "") -> str:
    """单段视频提示词(用户钦定 @图N 格式):
    【设定】@图N 点名 角色/场景/商品;【分镜】根据 @图(分镜图) 生成;【脚本】本镜脚本。
    refs 顺序与传给 seedance 的 reference_images 一致(@图N 对齐不错位)。"""
    product_name = (getattr(project.product, "title", "") or "商品").strip()
    setup_parts = []
    storyboard_idx = None
    for i, r in enumerate(refs or []):
        n = i + 1
        if r.get("type") == "storyboard":
            storyboard_idx = n
        else:
            setup_parts.append(f"@图{n}是{r.get('label') or ''}({_ENTITY_TYPE_CN.get(r.get('type'), '参考')})")
    lines = []
    if setup_parts:
        lines.append("【设定】" + "，".join(setup_parts) + "。")
    if storyboard_idx is not None:
        lines.append(f"【分镜】根据@图{storyboard_idx}分镜图生成「{product_name}」短视频。")
    script_text = _segment_script_text(scene) if scene is not None else ""
    extra = (user_prompt or "").strip()
    if extra:
        script_text = (script_text + "\n" + extra).strip() if script_text else extra
    lines.append("【脚本】" + (script_text or f"第 {video_segment.sort_order + 1} 段"))
    lines.append(
        f"{video_segment.target_duration_seconds}s · "
        + quality_suffix("video", "9:16 竖屏电商带货短视频,镜头稳定,商品露出清晰,节奏有转化感。")
    )
    return "\n".join(line for line in lines if line)


def submit_storyboard(*, project, user, prompt: str = "") -> StoryboardVersion:
    """异步故事板·提交:快速创建(或复用)一个未采用的版本,不在此处生图。逐帧生成交给 generate_storyboard_frame(轮询)。"""
    adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
    if adopted_script is None:
        raise ValueError("script must be adopted before generating storyboard")
    if get_default_model(ModelConfig.Capability.IMAGE) is None:
        raise ValueError("no active image model configured")
    # 复用尚未完成(未采用)的版本,避免重复提交产生多版本;否则新建
    version = project.storyboard_versions.filter(is_adopted=False).order_by("-created_at").first()
    if version is None:
        version = StoryboardVersion.objects.create(project=project, prompt=prompt)
    elif prompt and version.prompt != prompt:
        version.prompt = prompt
        version.save(update_fields=["prompt", "updated_at"])
    return version


_ENTITY_TYPE_CN = {"character": "角色", "scene": "场景", "product": "商品"}


def _product_reference_image(project, groups: list | None = None) -> dict | None:
    """商品参考图:优先已采用的商品三视图(product 基础资产组)→ 否则商品真实主图。无图返回 None。
    (商品是预创建的真实商品,不从脚本提取;每镜参考图都无条件带上它。)"""
    if groups is None:
        groups = list(
            project.base_asset_groups.filter(adopted_asset__isnull=False).select_related("adopted_asset")
        )
    pg = next(
        (g for g in groups if g.kind == BaseAssetGroup.Kind.PRODUCT and g.adopted_asset_id),
        None,
    )
    if pg is not None:
        url = _asset_preview_url(pg.adopted_asset)
        if url:
            return {"url": url, "label": "商品", "type": "product"}
    cover = _product_cover_url(project.product)
    if cover:
        return {"url": cover, "label": "商品", "type": "product"}
    return None


def _storyboard_reference_images(project, segment) -> list[dict]:
    """按本镜 entity_refs 取参考图(角色 / 场景 已采用基础资产)+ **无条件带上商品参考图**,
    供 gpt-image-2 多图合成 @图N。返回 [{url,label,type}],最多 4 张(角色/场景 ≤3 + 商品 1)。
    商品不靠 entity_refs(预创建真实商品、不从脚本提),统一用商品三视图 / 主图带上,从根上保证
    商品参考永不缺失。依赖提取步落进 metadata 的 script_entities。"""
    entities = {
        e.get("id"): e
        for e in (project.metadata or {}).get("script_entities", [])
        if isinstance(e, dict)
    }
    kind_by_type = {
        "character": BaseAssetGroup.Kind.PERSON,
        "scene": BaseAssetGroup.Kind.SCENE,
    }
    groups = list(project.base_asset_groups.filter(adopted_asset__isnull=False).select_related("adopted_asset"))
    # 下游只取「已采用」的角色/场景组(C/D):metadata.adopt 显性优先,否则按是否在脚本里推导;
    # 商品组永远保留(不受采用态影响)。顺带把不在脚本里的(含三视图组 label「·三视图」)排除,
    # 避免兜底误抓未采用/三视图组。
    _entity_names = {(e.get("name") or "").strip() for e in entities.values() if (e.get("name") or "").strip()}

    def _group_adopted(g) -> bool:
        if g.kind == BaseAssetGroup.Kind.PRODUCT:
            return True
        a = (g.metadata or {}).get("adopt")
        if a == "adopted":
            return True
        if a == "unadopted":
            return False
        return (g.metadata or {}).get("label", "").strip() in _entity_names

    groups = [g for g in groups if _group_adopted(g)]
    out: list[dict] = []
    used: set = set()
    for rid in (segment.entity_refs or []):
        ent = entities.get(rid)
        if not ent:
            continue
        kind = kind_by_type.get(ent.get("type"))
        if kind is None:  # 商品 / 未知类型不在此处理,商品统一在末尾无条件带上
            continue
        name = (ent.get("name") or "").strip()
        match = next(
            (g for g in groups if g.kind == kind and (g.metadata or {}).get("label", "").strip() == name and g.id not in used),
            None,
        ) or next((g for g in groups if g.kind == kind and g.id not in used), None)
        if match:
            used.add(match.id)
            url = _asset_preview_url(match.adopted_asset)
            if url:
                out.append({"url": url, "label": name or _ENTITY_TYPE_CN.get(ent.get("type"), "参考"), "type": ent.get("type")})
        if len(out) >= 3:  # 给商品留一个位置(总计最多 4 张)
            break
    # 商品参考图永远带上:优先已采用商品三视图组 → 否则真实主图
    product_ref = _product_reference_image(project, groups)
    if product_ref is not None and product_ref["url"] not in {r["url"] for r in out}:
        out.append(product_ref)
    # 规范 @图N 顺序:角色 → 商品 → 场景(与下游 image_edit 传图顺序一致,标注不错位)
    _ord = {"character": 0, "product": 1, "scene": 2}
    out.sort(key=lambda r: _ord.get(r.get("type"), 9))
    return out[:4]


def build_storyboard_frame_prompt_refs(project, version, segment, refs: list[dict]) -> str:
    """参考图合成版(用户钦定格式):顶部 @图N 点名每张参考图(角色/商品/场景),再给导演故事板指令 + 分镜脚本。
    @图N 顺序与传给 gpt-image-2 的参考图顺序严格一致(refs 即 image_edit 的 images 顺序)。"""
    if not refs:
        return build_storyboard_frame_prompt(project, version, segment)
    setup = "，".join(
        f"@图{i + 1}是{r['label']}({_ENTITY_TYPE_CN.get(r.get('type'), '参考')})" for i, r in enumerate(refs)
    )
    dur = segment.duration_seconds or 15
    lines = [
        f"【设定】{setup}。",
        "根据以下脚本生成一个导演故事板,用于指导 seedance 的视频生成。",
        _scene_context(project),
        f"【分镜脚本】(本段时长约 {dur} 秒)",
        _segment_script_text(segment) or f"第 {segment.sort_order + 1} 镜",
        "请严格保持各参考图中角色的同一张脸、同一商品的外观与配色;电商竖屏 9:16 导演故事板,一镜一图,画面清晰,可直接指导视频生成。",
    ]
    if version.prompt:
        lines.append(version.prompt.strip())
    return "\n".join(line for line in lines if line)


def _is_transient_error(exc: Exception) -> bool:
    """网络抖动/超时类瞬时错误(可重试),区别于内容审核拦截、参数非法等确定性失败。
    中转站(tokenssr 等)偶发 Read timeout / 连接重置会无谓掐掉单帧,这类才重试。"""
    msg = str(exc).lower()
    return any(
        k in msg
        for k in ("timed out", "timeout", "connection", "reset by peer", "temporarily",
                  "bad gateway", "502", "503", "504", "remotedisconnected", "max retries")
    )


def _call_image_with_retry(fn, *, attempts: int = 2, base_delay: float = 2.0):
    """对一次出图网络调用做有界重试:仅瞬时错误重试(指数退避),确定性失败立即抛出。
    出图无副作用(失败=没拿到图),重试安全;成功一次即返回。
    attempts 默认 2(一次重试):单次 HTTP 超时上限 300s,2 次≈10min,须 < poll 的「在途锁过期窗口」
    (STORYBOARD_INFLIGHT_STALE_MINUTES)否则 worker 重试期间任务被判僵尸 → 重复起线程 + 重复扣费。"""
    import time

    last: Exception | None = None
    for i in range(attempts):
        try:
            return fn()
        except Exception as exc:  # noqa: BLE001
            last = exc
            if i == attempts - 1 or not _is_transient_error(exc):
                raise
            time.sleep(base_delay * (i + 1))
    raise last  # 理论不可达(循环内已 return/raise)


def _storyboard_frame_worker(task_id, version_id, segment_id, user_id) -> None:
    """后台线程:真正调 ARK 生成一帧故事板图并落库。每次 poll 不阻塞在此——HTTP 永远秒回。"""
    import threading  # noqa: F401 — 仅标注此函数运行在独立线程
    from django.db import connections

    from apps.accounts.models import User

    try:
        task = AITask.objects.select_related("model_config__provider").get(id=task_id)
        version = StoryboardVersion.objects.select_related("project__team").get(id=version_id)
        segment = ScriptSegment.objects.get(id=segment_id)
        user = User.objects.get(id=user_id)
        project = version.project
        model_config = task.model_config
        reservation = task.credit_reservation
        task.status = AITask.Status.SUBMITTED
        task.save(update_fields=["status", "updated_at"])
        try:
            provider = get_image_provider(model_config)
            refs = _storyboard_reference_images(project, segment)
            ref_urls = [r["url"] for r in refs]
            if ref_urls and hasattr(provider, "image_edit"):
                # gpt-image-2 多图参考:必须用 refs 版提示词(点名「参考图N=角色/场景/商品」+锁脸锁商品),
                # 不能复用 request_payload['prompt'](那是建任务时写死的基础提示词,恒为真值会架空一致性约束)。
                frame_prompt = build_storyboard_frame_prompt_refs(project, version, segment, refs)
                # 中转站偶发 Read timeout 会掐掉单帧(4 镜出 3 张的根因)→ 瞬时错误有界重试
                response = _call_image_with_retry(
                    lambda: provider.image_edit(
                        model=model_config.name,
                        prompt=frame_prompt,
                        images=ref_urls,
                        size="1024x1536",
                    )
                )
            else:
                frame_prompt = task.request_payload.get("prompt") or build_storyboard_frame_prompt(project, version, segment)
                response = _call_image_with_retry(
                    lambda: provider.image_generation(
                        model=model_config.name,
                        endpoint=model_config.endpoint,
                        prompt=frame_prompt,
                    )
                )
            media = provider.extract_first_media_url(response)
            # 注意顺序:task 是 poll 端的「占位锁」,必须等帧真正落库后才置 SUCCEEDED。
            # 旧实现先置 SUCCEEDED 再上传 TOS(数秒)最后建帧,中间窗口 poll 会判「无在途且帧缺失」
            # 为同一镜重复起线程 → 重复帧 + 重复扣费(实测 4 帧出 6 帧)。
            asset = _store_generated_media(
                team=project.team,
                user=user,
                project=project,
                task=task,
                media=media,
                name=f"{project.name}-storyboard-{segment.sort_order + 1}",
                category=Asset.Category.STORYBOARD,
                asset_type=Asset.Type.IMAGE,
            )
            with transaction.atomic():
                task.status = AITask.Status.SUCCEEDED
                task.response_payload = response
                task.actual_cost = task.estimated_cost
                task.completed_at = timezone.now()
                task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
                charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
                # 幂等守卫:该镜已有帧(任何残余竞态/双 poll)就不再建,保持一镜一帧
                if not StoryboardFrame.objects.filter(storyboard=version, script_segment=segment).exists():
                    StoryboardFrame.objects.create(
                        storyboard=version,
                        script_segment=segment,
                        asset=asset,
                        sort_order=segment.sort_order,
                        prompt=segment.visual_prompt,
                    )
                # 合规(期3):分镜图含人脸,视频生成前必须过火山审核 → 事务提交后静默送审(best-effort)
                from apps.assets.review import submit_asset_for_review

                transaction.on_commit(lambda a=asset: submit_asset_for_review(a))
        except Exception as exc:  # noqa: BLE001 — 失败回滚额度,标记任务失败供 poll 上报
            task.status = AITask.Status.FAILED
            task.error_message = str(exc)
            task.completed_at = timezone.now()
            task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
            release_credit(reservation=reservation, reason=str(exc))
    finally:
        connections.close_all()  # 释放该线程的 DB 连接


def generate_storyboard_frame(*, project, user) -> dict:
    """异步故事板·轮询(秒回):读取进度;若无帧在生成则后台起线程生成下一帧。永不阻塞在 ARK 调用上。
    返回 {status: generating|succeeded|failed, done, total, version_id}。全部完成→采用版本。"""
    import threading

    version = project.storyboard_versions.filter(is_adopted=False).order_by("-created_at").first()
    adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
    if version is None or adopted_script is None:
        latest = project.storyboard_versions.order_by("-created_at").first()
        n = latest.frames.count() if latest else 0
        return {"status": "succeeded", "done": n, "total": n, "version_id": str(latest.id) if latest else ""}

    segments = list(adopted_script.segments.all().order_by("sort_order"))
    total = len(segments)
    done_segment_ids = set(version.frames.values_list("script_segment_id", flat=True))
    done = len(done_segment_ids)

    if done >= total:
        _finalize_storyboard(project, version)
        return {"status": "succeeded", "done": total, "total": total, "version_id": str(version.id)}

    # 每镜独立「占位锁」(该镜有 CREATED/RESERVED/SUBMITTED 的任务=在生成中)。
    # 旧实现是版本级单锁、一次只生成一帧;改为缺哪几镜就同时起哪几镜的线程。
    # ★ 实测注记(2026-06-10):4 线程并发时当前 Seedream 端点在服务侧排队,单帧 25s→83-105s,
    #   整版总时长 ≈ 串行(117s)。瓶颈是 ARK 端点并发配额而非本机;并行无额外成本,
    #   配额提升后自动受益。可用 settings.STORYBOARD_MAX_PARALLEL 调并发(1=回到串行)。
    # 仅算「近 N 分钟内」的任务:线程意外中断留下的僵尸任务超时后不再占锁,允许重新发起。
    # 窗口须 > worker 最坏运行时长(单帧 HTTP 超时 300s × 重试 2 次 ≈ 10min),否则 worker 正在
    # 重试时其任务被误判为僵尸 → 同一镜重复起线程 + 重复扣费。故默认 12min(覆盖一次瞬时重试)。
    from django.conf import settings as dj_settings
    STORYBOARD_MAX_PARALLEL = int(getattr(dj_settings, "STORYBOARD_MAX_PARALLEL", 4))
    stale_minutes = int(getattr(dj_settings, "STORYBOARD_INFLIGHT_STALE_MINUTES", 12))
    stale_cutoff = timezone.now() - timedelta(minutes=stale_minutes)
    inflight_segment_ids = {
        str(v)
        for v in AITask.objects.filter(
            project=project,
            task_type=AITask.Type.STORYBOARD,
            status__in=[AITask.Status.CREATED, AITask.Status.RESERVED, AITask.Status.SUBMITTED],
            request_payload__storyboard_version=str(version.id),
            created_at__gte=stale_cutoff,
        ).values_list("request_payload__storyboard_segment", flat=True)
        if v
    }

    pending = [s for s in segments if s.id not in done_segment_ids]
    # 单帧失败次数上限,避免持续失败时无限重试;任一镜到上限即整版上报失败
    for segment in pending:
        failed_for_segment = AITask.objects.filter(
            project=project,
            task_type=AITask.Type.STORYBOARD,
            status=AITask.Status.FAILED,
            request_payload__storyboard_segment=str(segment.id),
        ).count()
        if failed_for_segment >= 2:
            last = AITask.objects.filter(project=project, task_type=AITask.Type.STORYBOARD, status=AITask.Status.FAILED,
                                         request_payload__storyboard_segment=str(segment.id)).order_by("-created_at").first()
            return {"status": "failed", "done": done, "total": total, "version_id": str(version.id),
                    "error": last.error_message if last else "storyboard frame failed"}

    spawnable = [s for s in pending if str(s.id) not in inflight_segment_ids]
    slots = max(0, STORYBOARD_MAX_PARALLEL - len(inflight_segment_ids))
    model_config = get_default_model(ModelConfig.Capability.IMAGE)
    for segment in spawnable[:slots]:
        task = create_ai_task(
            project=project,
            user=user,
            task_type=AITask.Type.STORYBOARD,
            model_config=model_config,
            request_payload={
                "model": model_config.name,
                "endpoint": model_config.endpoint,
                "prompt": build_storyboard_frame_prompt(project, version, segment),
                "storyboard_version": str(version.id),
                "storyboard_segment": str(segment.id),
            },
        )
        threading.Thread(
            target=_storyboard_frame_worker,
            args=(str(task.id), str(version.id), str(segment.id), str(user.id)),
            daemon=True,
        ).start()
    return {"status": "generating", "done": done, "total": total, "version_id": str(version.id)}


def _finalize_storyboard(project, version) -> None:
    """全部帧就绪:采用该版本(反采用其余版本)。项目阶段推进由视图负责(与原同步实现一致)。"""
    project.storyboard_versions.exclude(id=version.id).update(is_adopted=False)
    if not version.is_adopted:
        version.is_adopted = True
        version.save(update_fields=["is_adopted", "updated_at"])


def _asset_preview_url(asset) -> str:
    """资产主文件的可公开访问 URL(已写绝对 URL 优先,否则实时签 TOS GET)。"""
    if asset is None:
        return ""
    primary = asset.files.filter(is_primary=True).first() or asset.files.first()
    if primary is None:
        return ""
    if primary.preview_url:
        return primary.preview_url
    try:
        return TosStorage().presigned_get_url(object_key=primary.object_key)
    except Exception:
        return ""


def _video_reference_images(project, video_segment) -> list[dict]:
    """视频参考图(带类型,供 @图N):角色/场景/商品 基础资产 + 本镜故事板帧。
    顺序:角色 → 场景 → 商品 → 分镜图(与用户钦定 @图1角色@图2场景@图3商品@图4分镜图 一致)。
    返回 [{url,label,type}];都取不到时兜底商品图。"""
    out: list[dict] = []
    scene = None
    adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
    if adopted_script is not None:
        scene = adopted_script.segments.filter(sort_order=video_segment.sort_order).first()
    if scene is not None:
        refs = list(_storyboard_reference_images(project, scene))  # 角色/商品/场景 实体图
        _vord = {"character": 0, "scene": 1, "product": 2}
        refs.sort(key=lambda r: _vord.get(r.get("type"), 9))
        out = refs
    # 末位追加本镜故事板帧(@图N 末位 = 分镜图)
    version = (
        project.storyboard_versions.filter(is_adopted=True).order_by("-created_at").first()
        or project.storyboard_versions.order_by("-created_at").first()
    )
    if version is not None:
        frame = (
            version.frames.filter(sort_order=video_segment.sort_order).first()
            or version.frames.order_by("sort_order").first()
        )
        if frame is not None:
            url = _asset_preview_url(frame.asset)
            if url:
                out.append({"url": url, "label": "分镜图", "type": "storyboard"})
    if not out:
        product_group = (
            project.base_asset_groups.filter(kind=BaseAssetGroup.Kind.PRODUCT, adopted_asset__isnull=False)
            .order_by("-created_at").first()
        )
        if product_group is not None:
            url = _asset_preview_url(product_group.adopted_asset)
            if url:
                out.append({"url": url, "label": "商品", "type": "product"})
    return out


def submit_video_segment(*, video_segment: VideoSegment, user, prompt: str) -> VideoSegmentVersion | None:
    model_config = get_default_model(ModelConfig.Capability.VIDEO)
    if model_config is None:
        raise ValueError("no active video model configured")
    project = video_segment.project

    # 衔接:按 sort_order 把视频段绑到对应脚本镜,并织出跟住该镜的提示词。
    scene = None
    adopted_script = project.script_versions.filter(is_adopted=True).prefetch_related("segments").first()
    if adopted_script is not None:
        scene = adopted_script.segments.filter(sort_order=video_segment.sort_order).first()
        if scene is not None and video_segment.script_segment_id != scene.id:
            video_segment.script_segment = scene
            video_segment.save(update_fields=["script_segment", "updated_at"])
    # 参考图(带类型):角色/场景/商品 基础资产 + 本镜故事板帧;@图N 提示词与传图顺序严格对齐。
    refs = _video_reference_images(project, video_segment)
    reference_images = [r["url"] for r in refs]
    final_prompt = build_video_segment_prompt(project, video_segment, scene, refs, prompt)

    task = create_ai_task(
        project=project,
        user=user,
        task_type=AITask.Type.VIDEO_SEGMENT,
        model_config=model_config,
        request_payload={
            "model": model_config.name,
            "endpoint": model_config.endpoint,
            "prompt": final_prompt,
            "duration": video_segment.target_duration_seconds,
            "ratio": "9:16",
            "video_segment_id": str(video_segment.id),
            "reference_images": reference_images,
        },
    )
    try:
        provider = build_provider(model_config)
        try:
            response = provider.create_video_task(
                model=model_config.name,
                endpoint=model_config.endpoint,
                prompt=final_prompt,
                duration=video_segment.target_duration_seconds,
                ratio="9:16",
                resolution="720p",
                reference_images=reference_images or None,
            )
        except Exception:
            # 降级:带参考图被拒时退回纯文生视频(文本里已含本镜旁白/画面,衔接不丢)
            if not reference_images:
                raise
            response = provider.create_video_task(
                model=model_config.name,
                endpoint=model_config.endpoint,
                prompt=final_prompt,
                duration=video_segment.target_duration_seconds,
                ratio="9:16",
                resolution="720p",
                reference_images=None,
            )
        task.provider_task_id = str(response.get("id") or response.get("task_id") or "")
        task.response_payload = response
        task.status = AITask.Status.SUBMITTED
        task.submitted_at = timezone.now()
        task.save(update_fields=["provider_task_id", "response_payload", "status", "submitted_at", "updated_at"])
        video_segment.status = VideoSegment.Status.RUNNING
        video_segment.save(update_fields=["status", "updated_at"])
        return None
    except Exception as exc:
        task.status = AITask.Status.FAILED
        task.error_message = str(exc)
        task.completed_at = timezone.now()
        task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
        release_credit(reservation=task.credit_reservation, reason=str(exc))
        video_segment.status = VideoSegment.Status.FAILED
        video_segment.error_message = str(exc)
        video_segment.save(update_fields=["status", "error_message", "updated_at"])
        raise


def poll_video_segment(*, video_segment: VideoSegment, user) -> VideoSegmentVersion | None:
    # 幂等:已完成的段直接回采用版;已失败的段不再 poll。避免对已成功 task 再 poll → 二次建版 / 二次扣费。
    if video_segment.status == VideoSegment.Status.SUCCEEDED:
        return video_segment.adopted_version or video_segment.versions.order_by("-created_at").first()
    if video_segment.status == VideoSegment.Status.FAILED:
        return None

    # ★ 先找「在途任务」再回退旧版本的任务。旧实现反过来:重跑时段上已有(旧)版本,
    # 取到旧版本挂的已成功任务 → 短路返回旧版,在途的新任务永远没人轮询,
    # 段永远卡「生成中」、新视频取不回来(实测重跑卡 40 分钟,ARK 侧其实早已生成完)。
    ai_task = video_segment.project.ai_tasks.filter(
        task_type=AITask.Type.VIDEO_SEGMENT,
        request_payload__video_segment_id=str(video_segment.id),
        status__in=[AITask.Status.SUBMITTED, AITask.Status.POLLING],
    ).order_by("-created_at").first()
    if ai_task is None:
        latest_version = video_segment.versions.order_by("-created_at").first()
        ai_task = latest_version.task if latest_version else None
    if ai_task is None:
        raise ValueError("no active video generation task")

    # task 已终态(可能被并发的 worker / 另一次 poll 处理过):直接回已有版,不再调 ARK。
    if ai_task.status == AITask.Status.SUCCEEDED:
        return video_segment.versions.filter(task=ai_task).order_by("-created_at").first()
    if ai_task.status in (AITask.Status.FAILED, AITask.Status.CANCELLED):
        return None

    provider = build_provider(ai_task.model_config)
    response = provider.poll_video_task(endpoint=ai_task.model_config.endpoint, provider_task_id=ai_task.provider_task_id)
    remote_status = response.get("status")
    if remote_status in {"queued", "running", "processing"}:
        # 仍在生成:只在状态首次进入 POLLING 时落一次库。旧实现每次 poll(5s 一次)都把完整
        # response JSON 回写远程 MySQL——纯浪费写带宽,终态时反正会存完整 payload。
        if ai_task.status != AITask.Status.POLLING:
            ai_task.status = AITask.Status.POLLING
            ai_task.save(update_fields=["status", "updated_at"])
        return None
    if remote_status in {"failed", "expired", "cancelled"}:
        ai_task.status = AITask.Status.FAILED
        ai_task.response_payload = response
        ai_task.error_message = response.get("error", {}).get("message", "video generation failed")
        ai_task.completed_at = timezone.now()
        ai_task.save(update_fields=["status", "response_payload", "error_message", "completed_at", "updated_at"])
        release_credit(reservation=ai_task.credit_reservation, reason=ai_task.error_message)
        video_segment.status = VideoSegment.Status.FAILED
        video_segment.error_message = ai_task.error_message
        video_segment.save(update_fields=["status", "error_message", "updated_at"])
        return None

    media = provider.extract_first_media_url(response)
    asset = _store_generated_media(
        team=video_segment.project.team,
        user=user,
        project=video_segment.project,
        task=ai_task,
        media=media,
        name=f"{video_segment.project.name}-segment-{video_segment.sort_order + 1}",
        category=Asset.Category.VIDEO_CLIP,
        asset_type=Asset.Type.VIDEO,
    )
    # 终态化必须持锁原子做:两个并发 poll(前端 5s 静默轮询 × 提交后轮询/worker)同时走到这里时,
    # 旧实现会同 task 建两个版本 + charge_reserved_credit 双扣费(实测 03:08:25 同秒双版本)。
    # select_for_update 锁 task 行,后到者看到 SUCCEEDED 直接回已有版,不再建版/扣费。
    with transaction.atomic():
        locked_task = AITask.objects.select_for_update().get(id=ai_task.id)
        if locked_task.status == AITask.Status.SUCCEEDED:
            existing = video_segment.versions.filter(task=locked_task).order_by("-created_at").first()
            if existing is not None:
                return existing
        locked_task.status = AITask.Status.SUCCEEDED
        locked_task.response_payload = response
        locked_task.actual_cost = locked_task.estimated_cost
        locked_task.completed_at = timezone.now()
        locked_task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
        charge_reserved_credit(reservation=locked_task.credit_reservation, actual_amount=locked_task.actual_cost)
        version = VideoSegmentVersion.objects.create(
            video_segment=video_segment,
            task=locked_task,
            asset=asset,
            prompt=locked_task.request_payload.get("prompt", ""),
            is_adopted=True,
        )
        video_segment.versions.exclude(id=version.id).update(is_adopted=False)
        video_segment.adopted_version = version
        video_segment.status = VideoSegment.Status.SUCCEEDED
        video_segment.error_message = ""
        video_segment.save(update_fields=["adopted_version", "status", "error_message", "updated_at"])
    return version


def create_export_job(*, timeline, user) -> ExportJob:
    return ExportJob.objects.create(timeline=timeline, status=ExportJob.Status.QUEUED)


# 图片趴三类(模特库+资产模型重构 期2):
#  · model + product → model_tryon(模特上身图);model 无 product → person(视频角色「生成演员」,留期3)
#  · cover → platform_kit(平台套图);image → free_create(自由创作)
_STANDALONE_CATEGORY = {
    "model": Asset.Category.PERSON,
    "cover": Asset.Category.PLATFORM_KIT,
    "image": Asset.Category.FREE_CREATE,
}
_STANDALONE_TASK_TYPE = {
    "model": AITask.Type.PERSON_IMAGE,
    "cover": AITask.Type.PRODUCT_IMAGE,
    "image": AITask.Type.PRODUCT_IMAGE,
}


def _reap_stale_standalone_image_tasks(*, team) -> None:
    """兜底:worker 崩溃/重启(OOM、部署)可能留下卡在 RESERVED 的出图任务,额度被一直占住、
    前端轮询也永远等不到结果。超过 10 分钟(远大于单张真实出图耗时 ~60s)仍 RESERVED 的判为僵尸:
    标记失败并退还预留额度。趁每次新提交时顺手回收,无需额外的定时任务(与导出僵尸清理同思路)。"""
    cutoff = timezone.now() - timedelta(minutes=10)
    stale = AITask.objects.filter(
        team=team,
        project__isnull=True,
        task_type__in=[AITask.Type.PERSON_IMAGE, AITask.Type.PRODUCT_IMAGE],
        status=AITask.Status.RESERVED,
        updated_at__lt=cutoff,
    )
    for task in stale:
        try:
            with transaction.atomic():
                task.status = AITask.Status.FAILED
                task.error_message = "worker 未在预期时间内完成(僵尸任务自动回收)"
                task.completed_at = timezone.now()
                task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
                try:
                    reservation = task.credit_reservation
                except ObjectDoesNotExist:
                    reservation = None
                if reservation is not None:
                    release_credit(reservation=reservation, reason="僵尸出图任务自动回收")
        except Exception:  # noqa: BLE001 — 单个回收失败不应阻断新任务提交
            continue


def enqueue_standalone_images(*, team, user, prompt: str, mode: str = "image", count: int = 1, product_id: str | None = None, reference_product: bool = False, model_id: str | None = None, model_entity_id: str | None = None, ratio: str | None = None) -> list[AITask]:
    """独立生图(图片创作 / 模特上身图 / 平台套图)改为**异步**:本函数在 Web 请求里只做「建任务 +
    预留额度」这种秒级的活,真正 ~30s 的 ARK 出图交给 Celery worker(generate_standalone_image_task)。

    这样 Web 层(gunicorn)不会被慢出图请求占住 worker → 健康探针不会被饿死 → 根治"几张图就整站 502"。
    且任务一旦提交(额度已预留),浏览器关掉 / 断网都不影响——worker 照样把图生成并落库,扣费/退费在
    worker 内闭环。返回已 RESERVED 的 AITask 列表,前端拿 id 轮询 GET /api/ai/generate-image/?ids=… 取结果。"""
    from apps.ai.tasks import generate_standalone_image_task

    _reap_stale_standalone_image_tasks(team=team)
    model_config = get_default_model(ModelConfig.Capability.IMAGE)
    if model_config is None:
        raise ValueError("no active image model configured")
    task_type = _STANDALONE_TASK_TYPE.get(mode, AITask.Type.PRODUCT_IMAGE)
    count = max(1, min(int(count or 1), 12))
    # 本次提交 = 一组(模特上身图组 / 平台套图组):同一 batch_id 串起这批图,前端可成组展示。
    batch_id = str(uuid.uuid4())
    tasks: list[AITask] = []
    for index in range(count):
        cost = estimate_cost(model_config)
        task = AITask.objects.create(
            team=team,
            created_by=user,
            project=None,
            task_type=task_type,
            status=AITask.Status.CREATED,
            model_config=model_config,
            idempotency_key=f"standalone-image:{team.id}:{uuid.uuid4()}",
            request_payload={"model": model_config.name, "endpoint": model_config.endpoint, "prompt": prompt, "mode": mode, "index": index, "product_id": str(product_id) if product_id else None, "reference_product": bool(reference_product), "model_id": str(model_id) if model_id else None, "model_entity_id": str(model_entity_id) if model_entity_id else None, "batch_id": batch_id, "ratio": str(ratio) if ratio else None},
            estimated_cost=cost,
        )
        # 预留额度若余额不足会抛 ValueError,在同步的 Web 请求里立刻反馈给前端(不会先建半套任务)
        reserve_credit(team=team, user=user, task=task, amount=cost)
        task.status = AITask.Status.RESERVED
        task.save(update_fields=["status", "updated_at"])
        tasks.append(task)
    # 额度都预留成功后再统一派发,避免"派发了任务但后面某张预留失败"的半成品状态
    for task in tasks:
        generate_standalone_image_task.delay(str(task.id))
    return tasks


def run_standalone_image_task(*, task_id: str) -> None:
    """Celery worker 内执行**单张**图的慢活:调 ARK → 成功落库扣费 / 失败退费。
    幂等:只处理 RESERVED 状态的任务,重复投递(celery retry / 重启重放)不会二次出图、二次扣费。"""
    task = AITask.objects.select_related("team", "created_by", "model_config").filter(id=task_id).first()
    if task is None or task.status != AITask.Status.RESERVED:
        return
    team = task.team
    user = task.created_by
    payload = task.request_payload or {}
    prompt = str(payload.get("prompt") or "")
    mode = str(payload.get("mode") or "image")
    index = int(payload.get("index") or 0)
    product_id = payload.get("product_id") or None
    # 模特上身图(mode=model 且绑了商品)= 图片趴「模特上身图」(引用模特库),归 model_tryon、不送审;
    # 「生成演员」同样走 mode=model 但无 product_id = 视频角色,仍归 person(送审,留期3 收编为「角色」)。
    if mode == "model" and product_id:
        category = Asset.Category.MODEL_TRYON
    else:
        category = _STANDALONE_CATEGORY.get(mode, Asset.Category.UNCATEGORIZED)
    model_config = task.model_config
    provider = get_image_provider(model_config)
    reservation = task.credit_reservation
    # 出图策略(都优先 image_edit 锁真实素材,模型不支持/无素材才回落纯文生图):
    #  · 模特上身图(mode=model):参考图1=商品真实主图 + 参考图2=选中模特 → 生成「该模特用该商品」效果图;
    #  · 平台套图(mode=cover):参考图1=商品真实主图(+ 有模特则参考图2=模特)→ 锁包装一致性出套图;
    #  · 商品三视图(reference_product):参考图1=商品真实主图 → 锁包装一致性;
    #  · 其余(图片创作):纯文生图。
    product = None
    if product_id:
        from apps.products.models import Product

        product = Product.objects.filter(id=product_id).first()
    can_edit = hasattr(provider, "image_edit")
    model_url = ""
    if payload.get("model_id"):
        model_asset = Asset.objects.filter(id=payload.get("model_id")).first()
        if model_asset is not None:
            model_url = _asset_preview_url(model_asset)
    product_url = _product_cover_url(product) if product is not None else ""

    edit_images: list[str] = []
    edit_prompt = ""
    if mode == "model" and can_edit and product_url:
        # 模特上身图:商品图必有,模特图可缺(缺则让模型自取真人模特)
        edit_images = [product_url] + ([model_url] if model_url else [])
        edit_prompt = build_model_tryon_prompt_refs(product, has_model=bool(model_url), base_prompt=prompt)
    elif mode == "cover" and can_edit and product_url:
        # 平台套图:参考图1=商品真实主图(锁包装一致性),有模特则参考图2=模特(锁人脸/身形)
        edit_images = [product_url] + ([model_url] if model_url else [])
        edit_prompt = build_platform_cover_prompt_refs(product, has_model=bool(model_url), base_prompt=prompt)
    elif bool(payload.get("reference_product")) and can_edit and product_url:
        edit_images = [product_url]
        edit_prompt = build_product_triview_prompt_refs(product, "")
    use_edit = bool(edit_images)
    try:
        if use_edit:
            if payload.get("reference_product"):
                size = "1536x1024"  # 三视图固定横向
            else:
                size = _ratio_to_image_size(str(payload.get("ratio") or ""))  # 模特图按选中比例
            response = provider.image_edit(model=model_config.name, prompt=edit_prompt, images=edit_images, size=size)
        else:
            response = provider.image_generation(model=model_config.name, endpoint=model_config.endpoint, prompt=prompt)
        media = provider.extract_first_media_url(response)
        with transaction.atomic():
            task.status = AITask.Status.SUCCEEDED
            task.response_payload = response
            task.actual_cost = task.estimated_cost
            task.completed_at = timezone.now()
            task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"])
            charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
            fileobj, content_type = VolcanoArkProvider.media_to_bytes(media)
            suffix = ".jpg" if "jpeg" in content_type else (".webp" if "webp" in content_type else ".png")
            asset_id = uuid.uuid4()
            object_key = f"teams/{team.id}/standalone/{asset_id}{suffix}"
            stored = TosStorage().upload_fileobj(fileobj=fileobj, object_key=object_key, content_type=content_type)
            asset_label = {"model": "模特上身图", "cover": "平台套图", "image": "图片创作"}.get(mode, mode)
            # 资产元数据:product_id(商品详情页据此只展示该商品素材)+ batch_id(成组)+ mode + model_entity_id(上身图溯源模特库)
            asset_meta: dict = {"mode": mode}
            if product_id:
                asset_meta["product_id"] = str(product_id)
            if payload.get("batch_id"):
                asset_meta["batch_id"] = str(payload["batch_id"])
            if payload.get("model_entity_id"):
                asset_meta["model_entity_id"] = str(payload["model_entity_id"])
            asset = Asset.objects.create(
                id=asset_id, team=team, created_by=user, name=f"AI 生成 · {asset_label} · {index + 1}",
                asset_type=Asset.Type.IMAGE, source=Asset.Source.AI_GENERATED, category=category, origin_task=task,
                metadata=asset_meta,
            )
            AssetFile.objects.create(asset=asset, object_key=stored.object_key, bucket=stored.bucket, content_type=stored.content_type, size_bytes=stored.size_bytes, is_primary=True)
    except Exception as exc:  # noqa: BLE001 — 失败要退费并把错误记进 AITask 供前端轮询读取;不向上抛(避免 celery 重试二次扣费)
        task.status = AITask.Status.FAILED
        task.error_message = str(exc)
        task.completed_at = timezone.now()
        task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
        release_credit(reservation=reservation, reason=str(exc))


# ── 旁白配音(TTS):每镜旁白合成一段语音,导出时作为人声轨混在 BGM 之上 ──

# 音色按「语音合成(经典版)」试用包实测可用清单配置;大模型音色(*_bigtts)需另开通「语音合成大模型」服务,当前账号 403
VOICEOVER_VOICES = [
    {"key": "BV700_streaming", "label": "灿灿 · 活力女声"},
    {"key": "BV034_streaming", "label": "知性姐姐 · 沉稳女声"},
    {"key": "BV001_streaming", "label": "通用女声"},
    {"key": "BV056_streaming", "label": "阳光男声"},
    {"key": "BV102_streaming", "label": "儒雅青年 · 解说男声"},
    {"key": "BV002_streaming", "label": "通用男声"},
]
DEFAULT_VOICEOVER_VOICE = VOICEOVER_VOICES[0]["key"]


def synthesize_project_voiceover(*, project, user, items: list[dict], voice_type: str, speed_ratio: float = 1.0) -> dict:
    """每镜旁白 → **逐句** TTS 配音资产(一句一段音频,带句内起点 offset_ms),映射写入
    timeline.metadata["voiceover"]。逐句才能支持「拖动字幕块 = 字幕和它的语音一起移动」;
    一次调用 = 一个 AITask = 计一次费;任何一句失败则整体失败并释放预留(不留半套配音)。"""
    from apps.projects.services.export import _split_subtitle_text

    texts = []  # (片段 index, 句序 cue, 句文本)
    for n, item in enumerate(items or []):
        text = str(item.get("text") or "").strip()
        if not text:
            continue
        idx = int(item.get("index", n))
        pieces = _split_subtitle_text(text) or [text]
        for j, piece in enumerate(pieces):
            texts.append((idx, j, piece))
    if not texts:
        raise ValueError("没有可配音的旁白文本")
    provider = VolcanoTtsProvider()
    if not provider.configured:
        raise TtsNotConfigured(
            "语音合成未配置:请在后端环境变量设置 VOLC_TTS_APPID 和 VOLC_TTS_ACCESS_TOKEN"
            "(火山引擎控制台 → 语音技术 → 语音合成大模型 → 创建应用)"
        )
    voice_type = voice_type or DEFAULT_VOICEOVER_VOICE
    model_config = get_default_model(ModelConfig.Capability.AUDIO)
    if model_config is None:
        raise ValueError("no active audio model configured")
    task = create_ai_task(
        project=project,
        user=user,
        task_type=AITask.Type.VOICEOVER,
        model_config=model_config,
        request_payload={
            "voice_type": voice_type,
            "speed_ratio": float(speed_ratio or 1.0),
            "items": [{"index": idx, "cue": j, "text": text} for idx, j, text in texts],
        },
    )
    reservation = task.credit_reservation
    try:
        synthesized = []
        for idx, j, text in texts:
            audio, duration_ms = provider.synthesize(text=text, voice_type=voice_type, speed_ratio=speed_ratio, uid=str(user.id))
            synthesized.append((idx, j, text, audio, duration_ms))
        with transaction.atomic():
            task.status = AITask.Status.SUCCEEDED
            task.actual_cost = task.estimated_cost
            task.completed_at = timezone.now()
            task.response_payload = {"segments": len(synthesized)}
            task.save(update_fields=["status", "actual_cost", "completed_at", "response_payload", "updated_at"])
            charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost)
            vo_items = []
            offset_acc: dict[int, int] = {}  # 同一片段内逐句顺排:句 j 的默认起点 = 前面句时长之和
            for idx, j, text, audio, duration_ms in synthesized:
                asset_id = uuid.uuid4()
                object_key = f"teams/{project.team_id}/projects/{project.id}/voiceover/{asset_id}.mp3"
                stored = TosStorage().upload_fileobj(fileobj=BytesIO(audio), object_key=object_key, content_type="audio/mpeg")
                asset = Asset.objects.create(
                    id=asset_id, team=project.team, created_by=user,
                    name=f"配音 · 场 {idx + 1} · 句 {j + 1}", asset_type=Asset.Type.AUDIO,
                    source=Asset.Source.AI_GENERATED, category=Asset.Category.UNCATEGORIZED,
                    origin_task=task, description=text,
                )
                AssetFile.objects.create(
                    asset=asset, object_key=stored.object_key, bucket=stored.bucket,
                    content_type=stored.content_type, size_bytes=stored.size_bytes, is_primary=True,
                )
                offset_ms = offset_acc.get(idx, 0)
                offset_acc[idx] = offset_ms + (duration_ms or 0)
                vo_items.append({
                    "index": idx, "cue": j, "text": text, "asset": str(asset.id),
                    "duration_ms": duration_ms, "offset_ms": offset_ms,
                })
            timeline, _ = Timeline.objects.get_or_create(
                project=project, defaults={"name": f"{project.name} Timeline", "duration_seconds": 60}
            )
            metadata = dict(timeline.metadata or {})
            metadata["voiceover"] = {
                "enabled": True,
                "voice_type": voice_type,
                "speed_ratio": float(speed_ratio or 1.0),
                "items": vo_items,
            }
            timeline.metadata = metadata
            timeline.save(update_fields=["metadata", "updated_at"])
        return metadata["voiceover"]
    except Exception as exc:
        task.status = AITask.Status.FAILED
        task.error_message = str(exc)
        task.completed_at = timezone.now()
        task.save(update_fields=["status", "error_message", "completed_at", "updated_at"])
        release_credit(reservation=reservation, reason=str(exc))
        raise