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

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.db import transaction
from django.utils import timezone

from apps.ai.models import AITask, ModelConfig
from apps.ai.providers import TtsNotConfigured, VolcanoArkProvider, VolcanoTtsProvider, YunqiProvider
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()
    )


def get_image_provider(model_config: ModelConfig):
    """生图按 provider 分流:yunqi 走 OpenAI 兼容网关(gpt-image-2),其余沿用火山 ARK。"""
    if model_config.provider.name == "yunqi":
        return YunqiProvider(base_url=model_config.provider.base_url or None)
    return VolcanoArkProvider(base_url=model_config.provider.base_url or None)


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 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 = VolcanoArkProvider(base_url=model_config.provider.base_url or None)
        response = provider.chat_completion(model=model_config.name, endpoint=model_config.endpoint, messages=messages)
        content = provider.extract_text(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)

            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,
                )

            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 _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) -> 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,
            )
            group = BaseAssetGroup.objects.create(project=project, kind=kind, task=task, prompt=prompt)
            group.candidate_assets.add(asset)
            group.adopted_asset = asset
            group.save(update_fields=["adopted_asset", "updated_at"])
            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


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)
            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 = VolcanoArkProvider(base_url=model_config.provider.base_url or None)
        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 = VolcanoArkProvider(base_url=ai_task.model_config.provider.base_url or None)
    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 generate_standalone_image(*, team, user, prompt: str, mode: str = "image", count: int = 1) -> list[Asset]:
    """不绑定项目的独立生图(图片创作 / 模特上身图 / 平台套图)。复用项目内生图链路,AITask.project=None。"""
    model_config = get_default_model(ModelConfig.Capability.IMAGE)
    if model_config is None:
        raise ValueError("no active image model configured")
    category = _STANDALONE_CATEGORY.get(mode, Asset.Category.UNCATEGORIZED)
    task_type = _STANDALONE_TASK_TYPE.get(mode, AITask.Type.PRODUCT_IMAGE)
    count = max(1, min(int(count or 1), 4))
    provider = get_image_provider(model_config)
    assets: list[Asset] = []
    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},
            estimated_cost=cost,
        )
        reserve_credit(team=team, user=user, task=task, amount=cost)
        task.status = AITask.Status.RESERVED
        task.save(update_fields=["status", "updated_at"])
        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)
                assets.append(asset)
        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
    return assets


# ── 旁白配音(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