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:
    return (
        ModelConfig.objects.select_related("provider")
        .filter(capability=capability, status=ModelConfig.Status.ACTIVE, provider__status="active")
        .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)
    return OpenAICompatibleProvider(base_url=base_url, api_key=api_key)


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 build_script_prompt(*, project, user_prompt: str, selling_point_ids: list[str] | None = None) -> list[dict[str, str]]:
    product = project.product
    selling_points = product.selling_points.all()
    if selling_point_ids:
        selling_points = selling_points.filter(id__in=selling_point_ids)
    selling_text = "\n".join(f"- {item.title}: {item.detail}" for item in selling_points)
    system = (
        "你是电商短视频脚本导演。请为 9:16 竖屏带货短视频生成 60 秒脚本，"
        "拆成 4 个 15 秒段落。严格按以下格式输出，段落之间空一行，不要输出其他内容：\n"
        "镜头1\n旁白：这一镜要念出来的口播文案（一两句话）\n画面：这一镜的画面描述、商品露出方式和转场建议\n\n"
        "镜头2\n旁白：…\n画面：…（依此类推到镜头4）"
    )
    user = f"""
商品标题：{product.title}
品牌：{product.brand or "未填写"}
类目：{product.category or "未填写"}
目标人群：{product.target_audience or "未填写"}
商品描述：{product.description or "未填写"}
卖点：
{selling_text or "未选择卖点，请根据商品信息自行提炼。"}

用户补充需求：
{user_prompt or "生成一条结构完整、节奏清晰、适合投放的带货短视频脚本。"}
""".strip()
    return [{"role": "system", "content": system}, {"role": "user", "content": user}]


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 竖屏)。"
        "人物指出镜的角色(例:女主、同事、闺蜜);场景指画面发生的地点或环境(例:卫生间、地铁、办公室)。"
        "去重,人物与场景各最多 6 个。只输出一个 JSON 对象,不要 markdown 代码块,不要任何额外文字,格式如下：\n"
        '{"cast":[{"name":"女主","prompt":"26岁都市女性,自然妆容,米色针织衫,柔和室内光,9:16竖屏"}],'
        '"scenes":[{"name":"卫生间","prompt":"现代简约浴室,暖色灯光,干净台面,9:16竖屏"}]}'
    )
    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 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 generate_project_script(*, project, user, user_prompt: str, selling_point_ids: list[str] | None = None, source: str = "ai") -> ScriptVersion:
    model_config = get_default_model(ModelConfig.Capability.TEXT)
    if model_config is None:
        raise ValueError("no active text model configured")

    messages = build_script_prompt(project=project, user_prompt=user_prompt, selling_point_ids=selling_point_ids)
    payload = {"model": model_config.name, "endpoint": model_config.endpoint, "messages": messages}
    task = create_ai_task(
        project=project,
        user=user,
        task_type=AITask.Type.SCRIPT_GENERATION,
        model_config=model_config,
        request_payload=payload,
    )
    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)
        content = provider.extract_text(response)

        # 出稿后自动提取人物 / 场景(best-effort,失败返回空,不挡主流程),供脚本页标签与基础资产 seed
        extracted = extract_cast_and_scenes(project=project, user=user, content=content)

        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)

            script = ScriptVersion.objects.create(
                project=project,
                task=task,
                title="AI 脚本",
                content=content,
                source=source if source in ("ai", "theme", "manual") else "ai",
                is_adopted=False,
            )
            for index, segment_text in enumerate(split_script_into_segments(content)):
                narration, visual = parse_segment_fields(segment_text)
                ScriptSegment.objects.create(
                    script_version=script,
                    sort_order=index,
                    duration_seconds=15,
                    narration=narration,
                    visual_prompt=visual,
                )

            # 把提取到的人物 / 场景(含每个标签的建议生图提示词)回填进 project.metadata:
            # 脚本页标签自动读 cast/scenes,基础资产据 cast_prompts/scene_prompts seed 每张卡。
            # 仅在提取到内容时覆盖,空结果不清掉用户已有标签。
            if extracted["cast"] or extracted["scenes"]:
                metadata = dict(project.metadata or {})
                if extracted["cast"]:
                    metadata["cast"] = extracted["cast"]
                    metadata["cast_prompts"] = extracted["cast_prompts"]
                if extracted["scenes"]:
                    metadata["scenes"] = extracted["scenes"]
                    metadata["scene_prompts"] = extracted["scene_prompts"]
                project.metadata = metadata
                project.save(update_fields=["metadata", "updated_at"])

            stage, _ = ProjectStage.objects.get_or_create(project=project, stage=ProjectStage.Stage.SCRIPT)
            stage.status = ProjectStage.Status.NEEDS_REVIEW
            stage.save(update_fields=["status", "updated_at"])
            return script
    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))
        raise


def build_segment_rerun_prompt(*, project, segment, instruction: str = "") -> list[dict[str, str]]:
    """单镜重跑提示词:带商品/卖点上下文 + 该镜当前内容 + 前后镜上下文(保连贯)+ 用户修改意见。
    只重写这一镜,严格输出「旁白：…/画面：…」两行,供 parse_segment_fields 精确解析。"""
    product = project.product
    selling_points = product.selling_points.all()
    selling_text = "\n".join(f"- {item.title}: {item.detail}" for item in selling_points)

    script = segment.script_version
    siblings = list(script.segments.order_by("sort_order"))
    prev_seg = next((s for s in reversed(siblings) if s.sort_order < segment.sort_order), None)
    next_seg = next((s for s in siblings if s.sort_order > segment.sort_order), None)

    def _brief(seg) -> str:
        narration = (seg.narration or "").strip()
        visual = (seg.visual_prompt or "").strip()
        return f"旁白：{narration or '无'}；画面：{visual or '无'}"

    system = (
        "你是电商短视频脚本导演。现在只需要**重写一条分镜**(其它分镜保持不变)。"
        "结合商品卖点、该镜当前内容、前后镜上下文和用户的修改意见,重新生成这一镜的旁白口播与画面描述,"
        "并保证与前后镜衔接连贯。严格按以下格式输出两行,不要输出镜头编号或任何其它内容：\n"
        "旁白：这一镜要念出来的口播文案（一两句话）\n画面：这一镜的画面描述、商品露出方式和转场建议"
    )
    context_lines = [
        f"商品标题：{product.title}",
        f"品牌：{product.brand or '未填写'}",
        f"类目：{product.category or '未填写'}",
        f"目标人群：{product.target_audience or '未填写'}",
        f"卖点：\n{selling_text or '未选择卖点，请根据商品信息自行提炼。'}",
        "",
        f"这是第 {segment.sort_order + 1} 镜(共 {len(siblings)} 镜),时长约 {segment.duration_seconds} 秒。",
        f"该镜当前内容：{_brief(segment)}",
    ]
    if prev_seg is not None:
        context_lines.append(f"上一镜(保持不变,用于衔接)：{_brief(prev_seg)}")
    if next_seg is not None:
        context_lines.append(f"下一镜(保持不变,用于衔接)：{_brief(next_seg)}")
    context_lines.append("")
    context_lines.append(f"用户的修改意见：{instruction.strip() or '让这一镜更有吸引力、表达更清晰，并与前后镜自然衔接。'}")
    user = "\n".join(context_lines).strip()
    return [{"role": "system", "content": system}, {"role": "user", "content": user}]


def regenerate_script_segment(*, project, user, segment, instruction: str = "") -> ScriptVersion:
    """单镜 AI 重跑:只重新生成该 ScriptSegment 的 narration + visual_prompt(其它镜不动),
    沿用 generate_project_script 的 AITask + 计费(reserve/charge/release)闭环,同步调 LLM。
    返回该 segment 所属的 ScriptVersion。"""
    model_config = get_default_model(ModelConfig.Capability.TEXT)
    if model_config is None:
        raise ValueError("no active text model configured")

    messages = build_segment_rerun_prompt(project=project, segment=segment, instruction=instruction)
    payload = {
        "model": model_config.name,
        "endpoint": model_config.endpoint,
        "messages": messages,
        "script_segment": str(segment.id),
    }
    task = create_ai_task(
        project=project,
        user=user,
        task_type=AITask.Type.SCRIPT_OPTIMIZATION,
        model_config=model_config,
        request_payload=payload,
    )
    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)
        content = provider.extract_text(response)
        narration, visual = parse_segment_fields(content)

        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)

            segment.narration = narration
            segment.visual_prompt = visual
            segment.save(update_fields=["narration", "visual_prompt", "updated_at"])
            return segment.script_version
    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))
        raise


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 generate_base_asset(*, project, user, kind: str, prompt: str, label: str = "") -> BaseAssetGroup:
    model_config = get_default_model(ModelConfig.Capability.IMAGE)
    if model_config is None:
        raise ValueError("no active image model configured")
    payload = {"model": model_config.name, "endpoint": model_config.endpoint, "prompt": prompt, "kind": kind}
    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,
    )
    reservation = task.credit_reservation
    try:
        provider = get_image_provider(model_config)
        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)
            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,
            )
            # label = 该资产对应的脚本提取标签(人物/场景名),用于把生成结果归回对应的标签卡
            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)
            group.candidate_assets.add(asset)
            group.adopted_asset = asset
            group.save(update_fields=["adopted_asset", "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))
            return group
    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


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 build_storyboard_frame_prompt(project, version, segment) -> str:
    """单帧故事板提示词:风格锚点 + 本镜画面(回退旁白)+ 版本统一指令。"""
    visual = (segment.visual_prompt or segment.narration or "").strip()
    lines = [
        _scene_context(project),
        f"第 {segment.sort_order + 1} 镜画面:{visual}" if visual else f"第 {segment.sort_order + 1} 镜",
    ]
    if version.prompt:
        lines.append(version.prompt.strip())
    lines.append("电商竖屏分镜图,构图清晰,可直接指导视频生成")
    return "\n".join(line for line in lines if line)


def build_video_segment_prompt(project, video_segment, scene, user_prompt: str) -> str:
    """单段视频提示词:把本镜旁白 + 画面 + 风格锚点织进去,让每个视频片段跟住对应脚本/故事板。"""
    lines = [_scene_context(project)]
    if scene is not None:
        if scene.narration:
            lines.append(f"旁白:{scene.narration.strip()}")
        visual = (scene.visual_prompt or scene.narration or "").strip()
        if visual:
            lines.append(f"画面:{visual}")
    if user_prompt:
        lines.append(user_prompt.strip())
    lines.append(
        f"第 {video_segment.sort_order + 1} 段 · {video_segment.target_duration_seconds}s · "
        "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 _storyboard_reference_images(project, segment) -> list[dict]:
    """按本镜 entity_refs 取参考图(角色/场景/商品的已采用基础资产),供 gpt-image-2 多图合成 @图N。
    返回 [{url,label,type}],最多 4 张;无匹配时兜底商品组。依赖脚本 agent 落进 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,
        "product": BaseAssetGroup.Kind.PRODUCT,
    }
    groups = list(project.base_asset_groups.filter(adopted_asset__isnull=False).select_related("adopted_asset"))
    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"))
        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) >= 4:
            break
    if not out:
        pg = next((g for g in groups if g.kind == BaseAssetGroup.Kind.PRODUCT), None)
        if pg:
            url = _asset_preview_url(pg.adopted_asset)
            if url:
                out.append({"url": url, "label": "商品", "type": "product"})
    return out


def build_storyboard_frame_prompt_refs(project, version, segment, refs: list[dict]) -> str:
    """参考图合成版故事板提示词:在基础提示词上点名每张参考图,要求锁脸/锁商品外观。"""
    base = build_storyboard_frame_prompt(project, version, segment)
    if not refs:
        return base
    ref_lines = "；".join(
        f"参考图{i + 1}={r['label']}({_ENTITY_TYPE_CN.get(r.get('type'), '参考')})" for i, r in enumerate(refs)
    )
    return (
        f"{base}\n参考图对应:{ref_lines}。"
        "请严格保持各参考图中角色的同一张脸、同一商品的外观与配色,按本镜画面重新构图合成为一张电商竖屏分镜图。"
    )


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)
                response = 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 = 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.SCENE,
                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,
                    )
        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=回到串行)。
    # 仅算「近 3 分钟内」的任务:线程意外中断留下的僵尸任务超时后不再占锁,允许重新发起。
    from django.conf import settings as dj_settings
    STORYBOARD_MAX_PARALLEL = int(getattr(dj_settings, "STORYBOARD_MAX_PARALLEL", 4))
    stale_cutoff = timezone.now() - timedelta(minutes=3)
    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[str]:
    """为本视频段挑一张视觉参考图:优先本镜故事板帧,兜底已采用商品基础资产。"""
    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:
                return [url]
    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:
            return [url]
    return []


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"])
    final_prompt = build_video_segment_prompt(project, video_segment, scene, prompt)

    # 参考图:优先用本镜故事板帧,其次商品/人物基础资产,给视频做视觉锚点(衔接故事板→视频)。
    reference_images = _video_reference_images(project, video_segment)

    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)


_STANDALONE_CATEGORY = {
    "model": Asset.Category.PERSON,
    "cover": Asset.Category.PRODUCT_IMAGE,
    "image": Asset.Category.PRODUCT_IMAGE,
}
_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) -> 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))
    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},
            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)
    category = _STANDALONE_CATEGORY.get(mode, Asset.Category.UNCATEGORIZED)
    model_config = task.model_config
    provider = get_image_provider(model_config)
    reservation = task.credit_reservation
    try:
        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 = Asset.objects.create(
                id=asset_id, team=team, created_by=user, name=f"AI 生成 · {mode} · {index + 1}",
                asset_type=Asset.Type.IMAGE, 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)
    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