xiaoxia-saas/apps/api/app/services/edit_plan_service.py

"""EditPlanService — 剪辑计划管理业务逻辑.

封装 EditPlan 和 EditPlanClip 的 CRUD 操作、状态机流转、
以及渲染生成流程，提供统一的业务接口供 API 路由层调用。
"""

from __future__ import annotations

import logging
from typing import Any, Dict, List, Optional

from app.services.asset_segment_tracker import (
    REUSE_RATIO_LIMIT,
    get_used_segments,
    make_reuse_callback,
    record_used_segments,
)
from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl import (
    SQLAlchemyEditPlanClipRepository,
    SQLAlchemyEditPlanRepository,
    SQLAlchemyGenerationTaskRepository,
)
from packages.domain.clip_operations import calculate_merge as _calc_merge
from packages.domain.clip_operations import calculate_shift_orders as _calc_shift_orders
from packages.domain.clip_operations import calculate_split as _calc_split
from packages.domain.clip_operations import validate_merge_clips as _validate_merge
from packages.domain.clip_operations import validate_split_time as _validate_split
from packages.domain.edit_plan import EditPlan, EditPlanStatus
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus

logger = logging.getLogger(__name__)


class EditPlanService:
    """剪辑计划管理服务

    职责:
    - 剪辑计划 CRUD（创建、查询、更新、删除）
    - 剪辑片段管理（增删改查、分配素材）
    - 状态机流转（draft → editing → rendering → completed/failed）
    - 渲染生成流程（触发 Celery 任务、查询进度）
    """

    def __init__(self, db: Session) -> None:
        self._plan_repo = SQLAlchemyEditPlanRepository(db)
        self._clip_repo = SQLAlchemyEditPlanClipRepository(db)
        self._generation_task_repo = SQLAlchemyGenerationTaskRepository(db)

    # ── 剪辑计划 CRUD ──────────────────────────────────────────────────────

    def get_plan(self, plan_id: str) -> Optional[EditPlan]:
        """获取计划详情"""
        return self._plan_repo.get(plan_id)

    def get_plan_or_raise(self, plan_id: str) -> EditPlan:
        """获取计划，不存在则抛出 ValueError"""
        plan = self._plan_repo.get(plan_id)
        if plan is None:
            raise ValueError(f"剪辑计划不存在: {plan_id}")
        return plan

    def create_plan(
        self,
        template_id: str,
        name: str,
        *,
        config: Optional[dict[str, Any]] = None,
        total_duration: float = 0.0,
        project_id: str = "",
        created_by_user_id: str = "",
    ) -> EditPlan:
        """创建剪辑计划（基础 CRUD，供内部测试与脚本使用）

        Raises:
            ValueError: 参数校验失败
        """
        plan = EditPlan.create(
            template_id=template_id,
            name=name,
            config=config,
            total_duration=total_duration,
            project_id=project_id,
            created_by_user_id=created_by_user_id,
        )
        created = self._plan_repo.create(plan)
        logger.info("创建剪辑计划: id=%s name=%s", created.id, created.name)
        return created

    def _auto_resume_editing(self, plan_id: str) -> None:
        """如果计划处于 completed/failed 状态，自动切回 editing（编辑操作前置）"""
        plan = self._plan_repo.get(plan_id)
        if plan is None:
            return
        if plan.status in (EditPlanStatus.COMPLETED, EditPlanStatus.FAILED):
            try:
                plan.resume_editing()
                self._plan_repo.update(plan)
                logger.info("自动重新编辑: plan_id=%s", plan_id)
            except ValueError:
                pass

    def update_plan(
        self,
        plan_id: str,
        *,
        name: Optional[str] = None,
        config: Optional[dict[str, Any]] = None,
        total_duration: Optional[float] = None,
    ) -> EditPlan:
        """更新计划基础字段

        Raises:
            ValueError: 计划不存在
        """
        existing = self.get_plan_or_raise(plan_id)

        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(plan_id)
        existing = self.get_plan_or_raise(plan_id)

        updated = EditPlan(
            id=existing.id,
            template_id=existing.template_id,
            name=name.strip() if name is not None else existing.name,
            status=existing.status,
            total_duration=total_duration if total_duration is not None else existing.total_duration,
            source_edit_plan_id=existing.source_edit_plan_id,
            project_id=existing.project_id,
            created_by_user_id=existing.created_by_user_id,
            config=config if config is not None else existing.config,
            created_at=existing.created_at,
            updated_at=existing.updated_at,
        )
        result = self._plan_repo.update(updated)
        logger.info("更新剪辑计划: id=%s", plan_id)
        return result

    # ── 状态机流转 ──────────────────────────────────────────────────────────

    def transition_status(self, plan_id: str, target_status: EditPlanStatus) -> EditPlan:
        """流转计划状态

        状态流转规则:
        - draft → editing (start_editing)
        - editing → rendering (start_rendering)
        - rendering → completed (mark_completed)
        - rendering → failed (mark_failed)
        - failed → draft (reset_to_draft)

        Raises:
            ValueError: 计划不存在或状态流转非法
        """
        plan = self.get_plan_or_raise(plan_id)

        # 如果已是目标状态，直接返回
        if plan.status == target_status:
            return plan

        # 根据目标状态调用对应的状态机方法
        # EDITING 支持从 draft / completed / failed 进入
        if target_status == EditPlanStatus.EDITING:
            if plan.status == EditPlanStatus.DRAFT:
                plan.start_editing()
            elif plan.status in (EditPlanStatus.COMPLETED, EditPlanStatus.FAILED):
                plan.resume_editing()
            else:
                raise ValueError(f"无法从 {plan.status} 切换到 {target_status}")
            result = self._plan_repo.update(plan)
            logger.info(
                "状态流转: plan_id=%s %s → %s",
                plan_id,
                plan.status,
                target_status,
            )
            return result

        transition_map = {
            EditPlanStatus.RENDERING: plan.start_rendering,
            EditPlanStatus.COMPLETED: plan.mark_completed,
            EditPlanStatus.FAILED: plan.mark_failed,
            EditPlanStatus.DRAFT: plan.reset_to_draft,
        }

        transition_fn = transition_map.get(target_status)
        if transition_fn is None:
            raise ValueError(f"无效的目标状态: {target_status}")

        transition_fn()
        result = self._plan_repo.update(plan)
        logger.info(
            "状态流转: plan_id=%s %s → %s",
            plan_id,
            plan.status,
            target_status,
        )
        return result

    # ── 剪辑片段管理 ────────────────────────────────────────────────────────

    def list_clips(
        self,
        plan_id: str,
        *,
        status: Optional[EditPlanClipStatus] = None,
        skip: int = 0,
        limit: int = 100,
    ) -> List[EditPlanClip]:
        """列出计划的片段"""
        # 确保计划存在
        self.get_plan_or_raise(plan_id)
        return self._clip_repo.list_by_plan(plan_id, status=status, skip=skip, limit=limit)

    def count_clips(
        self,
        plan_id: str,
        *,
        status: Optional[EditPlanClipStatus] = None,
    ) -> int:
        """统计片段数量"""
        return self._clip_repo.count(plan_id=plan_id, status=status)

    def get_clip(self, clip_id: str) -> Optional[EditPlanClip]:
        """获取片段详情"""
        return self._clip_repo.get(clip_id)

    def get_clip_or_raise(self, clip_id: str) -> EditPlanClip:
        """获取片段，不存在则抛出 ValueError"""
        clip = self._clip_repo.get(clip_id)
        if clip is None:
            raise ValueError(f"片段不存在: {clip_id}")
        return clip

    def create_clip(
        self,
        plan_id: str,
        clip_type: str,
        order: int,
        *,
        template_clip_config_id: str = "",
        asset_id: str = "",
        text_content: str = "",
        start_time: float = 0.0,
        duration: float = 0.0,
        transition_effect: str = "cut",
        transition_duration: float = 0.0,
        playback_speed: float = 1.0,
        config: Optional[dict[str, Any]] = None,
    ) -> EditPlanClip:
        """创建片段

        Raises:
            ValueError: 计划不存在或参数校验失败
        """
        # 确保计划存在
        self.get_plan_or_raise(plan_id)
        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(plan_id)

        clip = EditPlanClip.create(
            plan_id=plan_id,
            clip_type=clip_type,
            order=order,
            template_clip_config_id=template_clip_config_id,
            asset_id=asset_id,
            text_content=text_content,
            start_time=start_time,
            duration=duration,
            transition_effect=transition_effect,
            transition_duration=transition_duration,
            playback_speed=playback_speed,
            config=config,
        )
        created = self._clip_repo.create(clip)
        logger.info(
            "创建片段: id=%s plan_id=%s clip_type=%s order=%d",
            created.id,
            plan_id,
            created.clip_type,
            created.order,
        )
        return created

    def update_clip(
        self,
        clip_id: str,
        *,
        clip_type: Optional[str] = None,
        order: Optional[int] = None,
        asset_id: Optional[str] = None,
        text_content: Optional[str] = None,
        start_time: Optional[float] = None,
        duration: Optional[float] = None,
        transition_effect: Optional[str] = None,
        transition_duration: Optional[float] = None,
        playback_speed: Optional[float] = None,
        config: Optional[dict[str, Any]] = None,
    ) -> EditPlanClip:
        """更新片段

        Raises:
            ValueError: 片段不存在
        """
        existing = self.get_clip_or_raise(clip_id)

        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(existing.plan_id)

        # 速度边界钳制
        if playback_speed is not None:
            if playback_speed <= 0:
                playback_speed = 1.0
            elif playback_speed < 0.25:
                playback_speed = 0.25
            elif playback_speed > 4.0:
                playback_speed = 4.0

        updated = EditPlanClip(
            id=existing.id,
            plan_id=existing.plan_id,
            clip_type=clip_type.strip() if clip_type is not None else existing.clip_type,
            order=order if order is not None else existing.order,
            template_clip_config_id=existing.template_clip_config_id,
            asset_id=asset_id.strip() if asset_id is not None else existing.asset_id,
            text_content=text_content.strip() if text_content is not None else existing.text_content,
            start_time=start_time if start_time is not None else existing.start_time,
            duration=duration if duration is not None else existing.duration,
            transition_effect=(
                transition_effect.strip() if transition_effect is not None else existing.transition_effect
            ),
            transition_duration=(
                transition_duration if transition_duration is not None else existing.transition_duration
            ),
            playback_speed=playback_speed if playback_speed is not None else existing.playback_speed,
            status=existing.status,
            config=config if config is not None else existing.config,
            created_at=existing.created_at,
            updated_at=existing.updated_at,
        )
        result = self._clip_repo.update(updated)
        logger.info("更新片段: id=%s", clip_id)
        return result

    def assign_asset(self, clip_id: str, asset_id: str) -> EditPlanClip:
        """为片段分配素材

        Raises:
            ValueError: 片段不存在或 asset_id 为空
        """
        clip = self.get_clip_or_raise(clip_id)
        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(clip.plan_id)
        clip.assign_asset(asset_id)
        result = self._clip_repo.update(clip)
        logger.info("分配素材: clip_id=%s asset_id=%s", clip_id, asset_id)
        return result

    def delete_clip(self, clip_id: str) -> bool:
        """删除片段

        Returns:
            bool: 是否删除成功
        """
        deleted = self._clip_repo.delete(clip_id)
        if deleted:
            logger.info("删除片段: id=%s", clip_id)
        return deleted

    def delete_all_clips(self, plan_id: str) -> int:
        """删除计划下所有片段

        Returns:
            int: 删除的片段数量
        """
        count = self._clip_repo.delete_by_plan(plan_id)
        logger.info("删除所有片段: plan_id=%s count=%d", plan_id, count)
        return count

    def replace_all_clips_transactional(
        self,
        plan_id: str,
        clips_data: list[dict],
    ) -> int:
        """事务性地替换所有片段：清空→创建→标记ready，单事务保证原子性。

        Args:
            plan_id: 计划 ID
            clips_data: 片段数据列表，每项包含 asset_id/start_time/duration/order

        Returns:
            int: 创建的片段数量

        Raises:
            Exception: 任何步骤失败时自动回滚
        """
        from packages.adapters.sqlalchemy_impl.models import EditPlanClipModel

        db = self._clip_repo.session
        try:
            # 1. 清空现有 clips（不 commit）
            deleted_count = db.query(EditPlanClipModel).filter(EditPlanClipModel.plan_id == plan_id).delete()

            # 2. 批量创建新 clips（不 commit）
            for i, clip_item in enumerate(clips_data):
                order = clip_item.get("order") or i
                clip = EditPlanClip.create(
                    plan_id=plan_id,
                    clip_type=clip_item.get("clip_type", "main"),
                    order=order,
                    asset_id=clip_item.get("asset_id", ""),
                    text_content=clip_item.get("text_content", ""),
                    start_time=clip_item.get("start_time", 0.0),
                    duration=clip_item.get("duration", 0.0),
                    transition_effect=clip_item.get("transition_effect", "cut"),
                    transition_duration=clip_item.get("transition_duration", 0.0),
                    playback_speed=clip_item.get("playback_speed", 1.0),
                    config=clip_item.get("config") or None,
                )
                model = EditPlanClipModel(
                    id=clip.id,
                    plan_id=clip.plan_id,
                    clip_type=clip.clip_type,
                    order=clip.order,
                    asset_id=clip.asset_id,
                    text_content=clip.text_content,
                    start_time=clip.start_time,
                    duration=clip.duration,
                    transition_effect=clip.transition_effect,
                    transition_duration=clip.transition_duration,
                    playback_speed=clip.playback_speed,
                    status=clip.status.value,
                    config=clip.config,
                )
                db.add(model)

            # flush 让新建 clip 写入当前事务（未 commit），后续查询才能找到它们
            db.flush()

            # 3. 标记有 asset_id 的 clips 为 ready（不 commit）
            pending_with_asset = (
                db.query(EditPlanClipModel)
                .filter(
                    EditPlanClipModel.plan_id == plan_id,
                    EditPlanClipModel.status == "pending",
                    EditPlanClipModel.asset_id != "",
                )
                .all()
            )
            for m in pending_with_asset:
                m.status = "ready"

            # 4. 一次性提交
            db.commit()
            logger.info(
                "事务性替换片段: plan_id=%s deleted=%d created=%d",
                plan_id,
                deleted_count,
                len(clips_data),
            )
            return len(clips_data)
        except Exception:
            db.rollback()
            logger.exception("事务性替换片段失败: plan_id=%s", plan_id)
            raise

    def reselect_plan_for_variant(
        self,
        source_plan_id: str,
        candidate_asset_ids: list[str],
        *,
        created_by_user_id: str = "",
        name_suffix: str = "变体",
        voice_duration: float = 0.0,
        rng=None,
        batch_segments: dict[str, list[tuple[float, float]]] | None = None,
    ) -> EditPlan:
        """为批量变体生成独立 plan：完整重跑单视频选片流程（#1743）。

        与 clone_plan_for_variant（只重算起点、素材/顺序不变）不同，本方法：
        - 源 plan 片段骨架（clip_type/order/duration/文案/转场）保留；
        - 素材池 shuffle 随机分配 + main 片段顺序洗牌；
        - 起点走场景镜头洗牌/随机起点/历史区间避让（与单视频同一入口）；
        - 批次内同素材区间重叠 >20% 自动重选起点；
        - 新片段区间 record_used_segments 写回素材 metadata（跨变体/跨任务避让）。

        Args:
            source_plan_id: 源 plan（任务 0 / 预览源）。
            candidate_asset_ids: 素材池（源 plan 素材 ∪ 批次素材）。
            created_by_user_id: 新 plan 归属用户。
            name_suffix: plan 名后缀。
            rng: 可选随机数（测试注入种子）。
            batch_segments: 可选，外部传入的批次内已使用素材区间（前序变体避让用）。
                传入时作为初始避让对象；未传则保持原逻辑从源 plan clips 自建（向后兼容）。

        Raises:
            ValueError: 源 plan 不存在/无片段、素材池为空或时长全未知。
        """
        from packages.adapters.sqlalchemy_impl.models import AssetModel
        from packages.domain.plan_generator_utils import extract_scene_points_from_metadata
        from packages.domain.variant_plan_selector import reselect_clips_for_variant

        source = self.get_plan_or_raise(source_plan_id)

        # 分页读取源 plan 全部片段
        clips: List[EditPlanClip] = []
        skip, page = 0, 500
        while True:
            batch = self._clip_repo.list_by_plan(source_plan_id, skip=skip, limit=page)
            if not batch:
                break
            clips.extend(batch)
            if len(batch) < page:
                break
            skip += page
        if not clips:
            raise ValueError(f"源 plan 无片段，无法生成变体: {source_plan_id}")

        source_clips_data: list[dict[str, Any]] = [
            {
                "order": c.order if c.order is not None else i,
                "asset_id": c.asset_id,
                "start_time": float(c.start_time or 0.0),
                "duration": float(c.duration or 0.0),
                "clip_type": c.clip_type,
                "playback_speed": float(c.playback_speed or 1.0),
                "transition_effect": c.transition_effect,
                "transition_duration": float(c.transition_duration or 0.0),
                "text_content": c.text_content or "",
                "config": c.config or {},
            }
            for i, c in enumerate(clips)
        ]

        db = self._clip_repo.session

        # #1749：配音时长 → 每段目标段长（片段数=模板片段数定死；素材不足由渲染末帧冻结铺满）
        target_durations: list[float] | None = None
        try:
            voice = float(voice_duration or 0.0)
        except (TypeError, ValueError):
            voice = 0.0
        rhythm_template_for_reselect = None
        if source.config:
            rhythm_template_for_reselect = source.config.get("rhythm_template")

        if voice > 0 and source_clips_data:
            from packages.domain.voice_duration_planner import plan_clip_durations

            _effects: list[str | None] = [c.get("transition_effect") for c in source_clips_data]
            _tdurs: list[float] = [float(c.get("transition_duration") or 0.0) for c in source_clips_data]
            # #1855 P0：先占位durations为空dict，真正查durations在后面pool_ids确定后执行；
            # plan_clip_durations 的 asset_durations 参数在该函数中仅作最大段长钳制，
            # 这里先不依赖它（durations 还没查），传 None 让planner用默认策略；
            # 真正的asset_durations会在后面 clips_data 生成时传入 reselect_clips_for_variant
            target_durations = plan_clip_durations(
                len(source_clips_data),
                voice,
                transition_effects=_effects,
                transition_durations=_tdurs,
                rhythm_template=rhythm_template_for_reselect,
                asset_durations=None,
            )
            if target_durations:
                for _c, _d in zip(source_clips_data, target_durations, strict=False):
                    _c["duration"] = _d

        # 素材池 = 源 plan 素材 ∪ 调用方传入素材（去重保序）
        pool_ids: list[str] = []
        seen = set()
        for aid in [c.asset_id for c in clips if c.asset_id] + list(candidate_asset_ids or []):
            if aid and aid not in seen:
                seen.add(aid)
                pool_ids.append(aid)

        # 时长 + 场景点
        durations: dict[str, float] = {}
        scene_points: dict[str, list[float]] = {}
        if pool_ids:
            for m in db.query(AssetModel).filter(AssetModel.id.in_(pool_ids)).all():
                durations[m.id] = float(getattr(m, "duration", 0.0) or 0.0)
                pts = extract_scene_points_from_metadata(getattr(m, "metadata", None))
                if pts:
                    scene_points[m.id] = pts

        historical = get_used_segments(db, pool_ids)

        # 创建新 plan（复制模板归属与 config）
        new_plan = self.create_plan(
            template_id=source.template_id,
            name=f"{source.name or '剪辑计划'} · {name_suffix}",
            config=dict(source.config or {}),
            total_duration=source.total_duration,
            project_id=source.project_id or "",
            created_by_user_id=created_by_user_id or (source.created_by_user_id or ""),
        )

        # 批次内区间：外部传入时使用外部传入（含前序变体已用区间）；
        # 否则保持原逻辑从源 plan clips 自建（向后兼容）
        if batch_segments is not None:
            batch_segments_resolved: dict[str, list[tuple[float, float]]] = {
                k: list(v) for k, v in batch_segments.items()
            }
        else:
            batch_segments_resolved = {}
            for c in clips:
                if c.asset_id and float(c.duration or 0) > 0:
                    st = float(c.start_time or 0.0)
                    batch_segments_resolved.setdefault(c.asset_id, []).append((st, st + float(c.duration)))

        clips_data = reselect_clips_for_variant(
            source_clips_data,
            pool_ids,
            asset_durations=durations,
            asset_scene_points=scene_points,
            historical_used_segments=historical,
            batch_segments=batch_segments_resolved,
            target_durations=target_durations,
            rng=rng,
        )

        # 片段区间写回素材 metadata（与落库同事务；replace_all_clips_transactional 内 commit）
        for item in clips_data:
            aid = item.get("asset_id", "")
            if aid:
                st = float(item.get("start_time", 0.0))
                record_used_segments(db, aid, st, st + float(item.get("duration", 0.0)), new_plan.id)

        self.replace_all_clips_transactional(new_plan.id, clips_data)
        logger.info(
            "变体独立选片完成: source=%s new=%s clips=%d assets=%d",
            source_plan_id,
            new_plan.id,
            len(clips_data),
            len(pool_ids),
        )
        return new_plan

    def clone_plan_for_variant(
        self,
        source_plan_id: str,
        *,
        created_by_user_id: str = "",
        name_suffix: str = "变体",
        reuse_tracker: Optional[dict] = None,
    ) -> EditPlan:
        """为批量/多预览场景克隆一份独立 plan，片段起点全部重算（受控随机/复用）。

        复制源 plan 的模板归属、config 与片段结构（asset_id / duration / clip_type /
        order 不变），每个片段重新调用 ``_calc_random_start_time``：读取素材持久化的
        历史已用区间避让，耗尽时受控复用（use_count<3、最久未用），从而保证 N 条
        成片片段区间互不相同，且复用占比受控。

        - 不替换/不修改源 plan，源 plan 保留用户手动编辑结果。
        - 片段区间记录（record_used_segments）随新片段写入素材 metadata，与新 plan
          同事务；复用历史区间时由 record 自动累加 use_count。
        - 克隆的 clips 复用区间累计时长写入 reuse_tracker（可选），供调用方统计占比。

        Raises:
            ValueError: 源 plan 不存在或无可用片段。
        """
        from packages.adapters.sqlalchemy_impl.models import AssetModel
        from packages.domain.plan_generator_utils import _calc_random_start_time

        source = self.get_plan_or_raise(source_plan_id)

        # 分页读取源 plan 全部片段
        clips: List[EditPlanClip] = []
        skip, page = 0, 500
        while True:
            batch = self._clip_repo.list_by_plan(source_plan_id, skip=skip, limit=page)
            if not batch:
                break
            clips.extend(batch)
            if len(batch) < page:
                break
            skip += page
        if not clips:
            raise ValueError(f"源 plan 无片段，无法克隆变体: {source_plan_id}")

        # 创建新 plan（复制模板归属与 config）
        new_plan = self.create_plan(
            template_id=source.template_id,
            name=f"{source.name or '剪辑计划'} · {name_suffix}",
            config=dict(source.config or {}),
            total_duration=source.total_duration,
            project_id=source.project_id or "",
            created_by_user_id=created_by_user_id or (source.created_by_user_id or ""),
        )

        # 素材时长映射（O(N) 单查）
        asset_ids = list({c.asset_id for c in clips if c.asset_id})
        db = self._clip_repo.session
        durations: dict[str, float] = {}
        if asset_ids:
            for m in db.query(AssetModel).filter(AssetModel.id.in_(asset_ids)).all():
                durations[m.id] = float(getattr(m, "duration", 0.0) or 0.0)

        used_segments = get_used_segments(db, asset_ids)
        reused: dict[str, float] = reuse_tracker if reuse_tracker is not None else {}
        asset_assigned: dict[str, float] = {}
        # 回调内部预判复用后占比超 15% 则拒绝复用（calc 返回 None → 保留原起点）
        reuse_cb = make_reuse_callback(db, durations, reused, assigned_tracker=asset_assigned)

        clips_data: list[dict] = []
        for i, c in enumerate(clips):
            aid = c.asset_id
            dur = float(c.duration or 0.0)
            total = durations.get(aid, 0.0)
            if aid and total > 0 and dur > 0:
                # 复用占比闸门：本片段尚未分配，检查当前已复用占比
                # reused / assigned 是否超 15%，超则该素材不再分配（保留原起点）；
                # assigned=0（首个片段）放行，reused=0 时不误拦正常分配
                assigned = asset_assigned.get(aid, 0.0)
                eff_dur = min(dur, total)
                reused_amt = reused.get(aid, 0.0)
                ratio_blocked = assigned > 0 and reused_amt / assigned > REUSE_RATIO_LIMIT
                start = None
                if not ratio_blocked:
                    start = _calc_random_start_time(aid, eff_dur, durations, used_segments, on_exhausted=reuse_cb)
                if start is None:
                    start = float(c.start_time or 0.0)
                asset_assigned[aid] = assigned + eff_dur
                used_segments.setdefault(aid, []).append((start, start + eff_dur))
                record_used_segments(db, aid, start, start + eff_dur, new_plan.id)
            else:
                start = float(c.start_time or 0.0)

            clips_data.append(
                {
                    "order": c.order if c.order is not None else i,
                    "asset_id": aid,
                    "start_time": start,
                    "duration": dur,
                    "clip_type": c.clip_type,
                }
            )

        # 事务性写入新 plan 的片段（内部统一 commit/rollback）
        self.replace_all_clips_transactional(new_plan.id, clips_data)
        logger.info(
            "克隆变体 plan: source=%s new=%s clips=%d",
            source_plan_id,
            new_plan.id,
            len(clips_data),
        )
        return new_plan

    # ── #1749 配音时长分配 / 素材时长查询 / 批量变体 plan 确保 ──────────────

    def get_asset_durations(self, asset_ids: list[str]) -> dict[str, float]:
        """批量查询素材时长（秒），O(N) 单查；缺失/异常返回 0.0。"""
        from packages.adapters.sqlalchemy_impl.models import AssetModel

        ids = [a for a in dict.fromkeys(asset_ids or []) if a]
        if not ids:
            return {}
        db = self._clip_repo.session
        out: dict[str, float] = {}
        for m in db.query(AssetModel).filter(AssetModel.id.in_(ids)).all():
            try:
                out[m.id] = float(getattr(m, "duration", 0.0) or 0.0)
            except (TypeError, ValueError):
                out[m.id] = 0.0
        return out

    def apply_voice_duration_to_plan(self, plan_id: str, voice_duration: float) -> Optional[EditPlan]:
        """把配音时长分配到 plan 的每段（#1749）。

        - 片段数保持不变（= 模板片段数，定死）；
        - 每段 duration 按 voice_duration_planner 分配（含转场重叠扣减）；
        - 素材短于段长 → start_time 钳制为 0（末帧冻结由渲染侧 tpad/apad 铺满）；
        - plan.total_duration 回写为成片净时长（≈ 配音时长）；
        - 幂等：配音时长相同则分配结果不变，可重复调用。

        无配音（<=0）或无片段时直接返回 None，不报错。
        """
        try:
            voice = float(voice_duration or 0.0)
        except (TypeError, ValueError):
            return None
        if voice <= 0:
            return None

        plan = self.get_plan(plan_id)
        if plan is None:
            return None

        # #1855 P0：幂等判断——如果已成功分配过且当前 total_duration 已接近 voice_duration，直接返回
        try:
            existing_mark = None
            if plan.config:
                existing_mark = plan.config.get("voice_duration_applied")
            cur_total = float(plan.total_duration or 0.0)
            if existing_mark is not None and abs(existing_mark - voice) < 1e-6 and abs(cur_total - voice) < 0.5:
                return plan
        except Exception:
            pass

        clips: List[EditPlanClip] = []
        skip, page = 0, 500
        while True:
            batch = self._clip_repo.list_by_plan(plan_id, skip=skip, limit=page)
            if not batch:
                break
            clips.extend(batch)
            if len(batch) < page:
                break
            skip += page
        if not clips:
            return None

        clips.sort(key=lambda c: (c.order if c.order is not None else 0))

        from packages.domain.voice_duration_planner import plan_clip_durations, total_output_duration

        # #1764：从 plan config 读取节奏模板
        rhythm_template = None
        if plan and hasattr(plan, "config") and plan.config:
            rhythm_template = plan.config.get("rhythm_template")

        # #1768：先获取素材时长，传入 plan_clip_durations 用于最大片段钳制
        asset_ids = [c.asset_id for c in clips if c.asset_id]
        durations = self.get_asset_durations(asset_ids)
        asset_durations_for_plan = [durations.get(c.asset_id, 0.0) for c in clips]

        target = plan_clip_durations(
            len(clips),
            voice,
            transition_effects=[c.transition_effect for c in clips],
            transition_durations=[float(c.transition_duration or 0.0) for c in clips],
            rhythm_template=rhythm_template,
            asset_durations=asset_durations_for_plan,
        )
        if not target:
            return None

        clips_data: list[dict] = []
        for i, c in enumerate(clips):
            dur = float(target[i])
            total = durations.get(c.asset_id, 0.0)
            start = float(c.start_time or 0.0)
            if c.asset_id and total > 0:
                # 素材短于段长：起点钳 0，段长超出部分渲染侧末帧冻结
                max_start = max(0.0, total - min(dur, total))
                start = min(start, max_start)
            clips_data.append(
                {
                    "order": c.order if c.order is not None else i,
                    "asset_id": c.asset_id or "",
                    "start_time": round(start, 3),
                    "duration": dur,
                    "clip_type": c.clip_type,
                    "playback_speed": float(c.playback_speed or 1.0),
                    "transition_effect": c.transition_effect,
                    "transition_duration": float(c.transition_duration or 0.0),
                    "text_content": c.text_content or "",
                    "config": c.config or {},
                }
            )

        self.replace_all_clips_transactional(plan_id, clips_data)

        net = total_output_duration(
            target,
            transition_effects=[c.transition_effect for c in clips],
            transition_durations=[float(c.transition_duration or 0.0) for c in clips],
        )
        try:
            plan.total_duration = net
            # #1855 P0：写入幂等标记，避免二次调用时只重分配 duration 不重算 start_time
            new_cfg = dict(plan.config or {})
            new_cfg["voice_duration_applied"] = voice
            plan.config = new_cfg
            db = self._clip_repo.session
            db.commit()
        except Exception:
            db.rollback()
            logger.exception("回写 plan.total_duration 失败（不阻断）: plan_id=%s", plan_id)

        logger.info(
            "配音时长分配完成: plan=%s clips=%d voice=%.2fs 成片净时长=%.2fs",
            plan_id,
            len(clips),
            voice,
            net,
        )
        return plan

    def ensure_variant_plans(
        self,
        source_plan_id: str,
        count: int,
        candidate_asset_ids: list[str],
        *,
        created_by_user_id: str = "",
        voice_durations: Optional[list[float]] = None,
        rng=None,
    ) -> list[str]:
        """确保批量 N 个变体各自拥有独立 plan（#1749 批量正式生成/预览共用）。

        - 变体 0：clone 源 plan（不污染源 plan，片段独立可改），并按配音分配段长；
        - 变体 1..N-1：reselect_plan_for_variant 完整重跑选片（素材级去重）；
        - voice_durations：每个变体的配音时长（独立配音各自时长；统一配音同值）；
          缺省/为 0 时不分配（段长保持骨架/模板值）。

        Returns:
            plan_id 列表，长度 == count，index 即 variant_index。
        """
        import random as _random

        rng = rng or _random.Random()
        plan_ids: list[str] = []

        # #1855 P0：先确定片段数 clip_count（用于节奏模板生成长度匹配）
        from packages.domain.bgm_pool import allocate_bgm_pool_for_variants
        from packages.domain.variant_plan_selector import (
            generate_pixel_perturbation,
            generate_visual_perturbation,
        )
        from packages.domain.voice_duration_planner import RHYTHM_TEMPLATES, adapt_template_length

        clip_count = 0
        # 从源 plan 获取片段数（分页读，避免关系加载问题）
        _sclips: list = []
        _sk, _pg = 0, 500
        while True:
            _b = self._clip_repo.list_by_plan(source_plan_id, skip=_sk, limit=_pg)
            if not _b:
                break
            _sclips.extend(_b)
            if len(_b) < _pg:
                break
            _sk += _pg
        clip_count = len(_sclips)

        # 预先生成所有 N 个变体的节奏模板/BGM/扰动参数（时机提前到选片前写入config）
        rhythm_templates_for_variants: list = []
        for _idx in range(count):
            if clip_count > 0:
                variant_seed = rng.randint(0, 999999)
                _tpl = adapt_template_length(RHYTHM_TEMPLATES[variant_seed % len(RHYTHM_TEMPLATES)], clip_count)
                rhythm_templates_for_variants.append(_tpl)
            else:
                rhythm_templates_for_variants.append(None)

        source_bgm_config: dict = {}
        source_plan = self.get_plan(source_plan_id)
        if source_plan and source_plan.config:
            source_bgm_config = source_plan.config.get("bgm", {}) or {}
        variant_seeds_for_bgm = [rng.randint(0, 999999) for _ in range(count)]
        bgm_pool_assignments = allocate_bgm_pool_for_variants(source_bgm_config, variant_seeds_for_bgm)

        def _build_variant_config_update(idx: int) -> dict:
            """构建单个变体的 config 更新（节奏模板/BGM/视觉/像素扰动）。"""
            upd: dict = {}
            try:
                perturbation = generate_visual_perturbation(rng)
                if idx == 0:
                    perturbation["hflip"] = False
                upd["visual_perturbation"] = perturbation
            except Exception:
                logger.exception("变体 %d 视觉扰动生成失败（不阻断）", idx)
            try:
                pixel_pert = generate_pixel_perturbation(rng)
                upd["pixel_perturbation"] = pixel_pert
            except Exception:
                logger.exception("变体 %d 像素扰动生成失败（不阻断）", idx)
            rt = rhythm_templates_for_variants[idx] if idx < len(rhythm_templates_for_variants) else None
            if rt is not None:
                upd["rhythm_template"] = rt
            if idx < len(bgm_pool_assignments):
                existing_bgm = dict((source_plan.config or {}).get("bgm", {}) or {})
                existing_bgm.update(bgm_pool_assignments[idx])
                upd["bgm"] = existing_bgm
            return upd

        # 变体 0：clone（片段结构同源 plan，起点重算），不污染源 plan
        plan0 = self.clone_plan_for_variant(
            source_plan_id,
            created_by_user_id=created_by_user_id,
            name_suffix="变体1",
        )
        v0_voice = 0.0
        if voice_durations and len(voice_durations) > 0:
            try:
                v0_voice = float(voice_durations[0] or 0.0)
            except (TypeError, ValueError):
                v0_voice = 0.0

        # #1855 P0：在配音分配前先写入变体0的节奏模板/扰动/BGM，确保 apply_voice_duration_to_plan 能读到 rhythm_template
        try:
            _cfg0 = _build_variant_config_update(0)
            if _cfg0:
                self.update_plan_config(plan0.id, _cfg0)
        except Exception:
            logger.exception("变体0 配置写入失败（不阻断）: plan=%s", plan0.id)

        if v0_voice > 0:
            try:
                self.apply_voice_duration_to_plan(plan0.id, v0_voice)
            except Exception:
                logger.exception("变体0 配音分配失败（不阻断）: plan=%s", plan0.id)
        plan_ids.append(plan0.id)

        # #1855 P0：批次内素材区间避让表——从变体0实际落库的clips构建初始值（公共函数）
        from app.services.generation_common import collect_plan_segments as _collect_plan_segments

        batch_segments_acc: dict[str, list[tuple[float, float]]] = _collect_plan_segments(plan0.id, self._clip_repo)

        # 变体 1..N-1：独立选片（传入累积的 batch_segments 做区间避让）
        for i in range(1, count):
            voice = 0.0
            if voice_durations and i < len(voice_durations):
                try:
                    voice = float(voice_durations[i] or 0.0)
                except (TypeError, ValueError):
                    voice = 0.0

            # #1855 P0：在reselect前先为"变体i"准备配置更新——但reselect内部复制的是source.config，
            # 所以每个变体独立的节奏模板需要在reselect后单独写入config；
            # 但 plan_clip_durations 用的是 source.config.rhythm_template（即源plan的节奏模板），
            # 为了让每个变体在选片阶段就使用自己的节奏模板分配段长，这里采用：
            #   - reselect 仍使用源 plan 的 rhythm_template（保持片段骨架一致）
            #   - 选片完成后立即写入该变体自己的 rhythm_template/扰动/BGM 到config，
            #     后续不再二次 apply_voice_duration_to_plan（由幂等标记跳过）
            variant = self.reselect_plan_for_variant(
                source_plan_id,
                candidate_asset_ids,
                created_by_user_id=created_by_user_id,
                name_suffix=f"变体{i + 1}",
                voice_duration=voice,
                rng=rng,
                batch_segments=batch_segments_acc,
            )

            # 选片完成后写入该变体的独立配置（节奏模板/扰动/BGM）
            try:
                _cfgi = _build_variant_config_update(i)
                if _cfgi:
                    self.update_plan_config(variant.id, _cfgi)
            except Exception:
                logger.exception("变体 %d 配置写入失败（不阻断）: plan=%s", i, variant.id)

            plan_ids.append(variant.id)

            # #1855 P0：把当前新变体的 clips 区间追加到 batch_segments，供下一变体避让
            try:
                _new_segs = _collect_plan_segments(variant.id, self._clip_repo)
                for _aid, _ivs in _new_segs.items():
                    batch_segments_acc.setdefault(_aid, []).extend(_ivs)
            except Exception:
                logger.exception("变体 %d 区间收集失败（不阻断）: plan=%s", i, variant.id)

        # 标记所有变体 plan 的 clips 为 ready（已分配素材+起点，语义上就是 ready）
        for pid in plan_ids:
            try:
                self.mark_clips_ready(pid)
            except Exception:
                logger.exception("标记 clips ready 失败（不阻断）: plan=%s", pid)

        return plan_ids

    # ── 片段分割与合并 ──────────────────────────────────────────────────────

    def split_clip(self, clip_id: str, split_time: float) -> Dict[str, Any]:
        """将一个片段从指定位置分割为两个片段

        Args:
            clip_id: 要分割的片段 ID
            split_time: 分割点（相对于片段起始的秒数），必须在 (0, duration) 范围内

        Returns:
            dict: {"left_clip": EditPlanClip, "right_clip": EditPlanClip}

        Raises:
            ValueError: 片段不存在、分割时间越界
        """
        clip = self.get_clip_or_raise(clip_id)
        plan_id = clip.plan_id

        # 纯逻辑：校验 + 计算
        _validate_split(split_time, clip.duration)
        split = _calc_split(
            duration=clip.duration,
            split_time=split_time,
            start_time=clip.start_time,
        )

        self._auto_resume_editing(plan_id)

        original_order = clip.order

        # 更新左半部分（原片段）
        clip.duration = split.left_duration
        left_clip = self._clip_repo.update(clip)

        # 后面片段的 order 全部 +1（给右半部分腾位置）
        all_clips = self._clip_repo.list_by_plan(plan_id)
        shifts = _calc_shift_orders(
            all_clips,
            threshold_order=original_order,
            shift=1,
            excluded_ids={clip_id},
            id_attr="id",
            order_attr="order",
        )
        for c, new_order in shifts:
            c.order = new_order
            self._clip_repo.update(c)

        # 创建右半部分新片段（继承原片段的大部分属性）
        right_config = dict(clip.config) if clip.config else {}
        # 素材裁剪信息
        if clip.asset_id:
            # 右半部分从 split_time 开始播放
            right_config["trim_start"] = split.right_trim_start
            # 左半部分在 split_time 处结束
            left_config = dict(left_clip.config) if left_clip.config else {}
            left_config["trim_end"] = split.left_trim_end
            left_clip.config = left_config
            left_clip = self._clip_repo.update(left_clip)

        right_clip = EditPlanClip.create(
            plan_id=plan_id,
            clip_type=clip.clip_type,
            order=original_order + 1,
            template_clip_config_id=clip.template_clip_config_id,
            asset_id=clip.asset_id,
            text_content=clip.text_content,
            start_time=split.right_start_time,
            duration=split.right_duration,
            transition_effect=clip.transition_effect,
            transition_duration=clip.transition_duration,
            playback_speed=clip.playback_speed,
            config=right_config,
        )
        created_right = self._clip_repo.create(right_clip)

        logger.info(
            "分割片段: clip_id=%s plan_id=%s split_time=%.3fs left_dur=%.3fs right_dur=%.3fs",
            clip_id,
            plan_id,
            split_time,
            split.left_duration,
            split.right_duration,
        )

        return {
            "left_clip": left_clip,
            "right_clip": created_right,
        }

    def merge_clips(self, clip_ids: List[str]) -> EditPlanClip:
        """合并多个连续片段为一个片段

        Args:
            clip_ids: 要合并的片段 ID 列表（至少2个），必须属于同一个计划且 order 连续

        Returns:
            EditPlanClip: 合并后的新片段

        Raises:
            ValueError: 数量不足、不属于同一计划、不连续、类型不一致
        """
        if len(clip_ids) < 2:
            raise ValueError("至少需要 2 个片段才能合并")

        # 读取所有片段
        clips = []
        for cid in clip_ids:
            clip = self.get_clip_or_raise(cid)
            clips.append(clip)

        # 纯逻辑：校验 + 计算
        plan_id, first_order = _validate_merge(clips)
        merge = _calc_merge(clips)

        self._auto_resume_editing(plan_id)

        # 更新第一个片段（保留它作为合并结果）
        first_clip = sorted(clips, key=lambda c: c.order)[0]
        first_clip.duration = merge.total_duration
        first_clip.text_content = merge.merged_text
        first_clip.config = merge.merged_config
        # 转场保留第一个的（合并后的入点转场）
        # playback_speed 取第一个的
        merged_clip = self._clip_repo.update(first_clip)

        # 删除其余片段
        rest_ids = [c.id for c in clips if c.id != merged_clip.id]
        for cid in rest_ids:
            self._clip_repo.delete(cid)

        # 后面的片段 order 前移 (len - 1) 位
        all_clips = self._clip_repo.list_by_plan(plan_id)
        shifts = _calc_shift_orders(
            all_clips,
            threshold_order=first_order,
            shift=-merge.shift_amount,
            excluded_ids={merged_clip.id},
            id_attr="id",
            order_attr="order",
        )
        for c, new_order in shifts:
            c.order = new_order
            self._clip_repo.update(c)

        logger.info(
            "合并片段: plan_id=%s count=%d total_duration=%.3fs",
            plan_id,
            len(clips),
            merge.total_duration,
        )

        return merged_clip

    # ── 渲染生成流程 ────────────────────────────────────────────────────────

    def get_generation_status(self, plan_id: str) -> Dict[str, Any]:
        """获取渲染进度状态

        Returns:
            dict: {
                "plan": EditPlan,
                "clips": List[EditPlanClip],
                "generation_task_id": Optional[str],
                "generation_task_status": Optional[str],
                "progress": float,
                "error_message": str,
            }

        Raises:
            ValueError: 计划不存在
        """
        plan = self.get_plan_or_raise(plan_id)
        clips = self._clip_repo.list_by_plan(plan_id)

        # 从 plan.config 中获取 generation_task_id
        generation_task_id = plan.config.get("generation_task_id")
        generation_task_status = None
        progress = 0.0
        error_message = ""

        if generation_task_id:
            task = self._generation_task_repo.get(generation_task_id)
            if task:
                generation_task_status = task.status.value if hasattr(task.status, "value") else task.status
                progress = getattr(task, "progress", 0.0) or 0.0
                error_message = getattr(task, "error_message", "") or ""

        return {
            "plan": plan,
            "clips": clips,
            "generation_task_id": generation_task_id,
            "generation_task_status": generation_task_status,
            "progress": progress,
            "error_message": error_message,
        }

    def can_generate(self, plan_id: str) -> tuple[bool, str]:
        """检查是否可以触发渲染

        包含最后一道防线的自动修复：
        - 如果 clips 存在但都没有 asset_id，且 config.asset_ids 非空，
          直接在内部执行素材分配，不再依赖前置 fallback 链路。

        Returns:
            tuple: (can_generate, reason)
        """
        plan = self.get_plan_or_raise(plan_id)

        # 检查状态
        if plan.status != EditPlanStatus.EDITING:
            return False, "请先编辑并保存模板后再生成视频"

        # 检查是否有片段
        clips = self._clip_repo.list_by_plan(plan_id)
        if not clips:
            return False, "请先添加片段后再生成视频"

        # 检查是否至少有一个片段分配了素材
        has_asset = any(c.asset_id for c in clips)
        config_asset_ids_count = len((plan.config or {}).get("asset_ids", []))
        clips_with_asset_count = sum(1 for c in clips if c.asset_id)
        logger.info(
            "can_generate 诊断: plan=%s status=%s total_clips=%d clips_with_asset=%d config_asset_ids_count=%d",
            plan_id,
            plan.status,
            len(clips),
            clips_with_asset_count,
            config_asset_ids_count,
        )
        if not has_asset:
            # ── 最后防线：自动从 config.asset_ids 分配素材 ──
            config_asset_ids = (plan.config or {}).get("asset_ids", [])
            if config_asset_ids:
                logger.warning(
                    "can_generate 最后防线触发: plan=%s clips=%d 均无素材，从 config.asset_ids(%d个) 自动分配",
                    plan_id,
                    len(clips),
                    len(config_asset_ids),
                )
                clips_without_asset = [c for c in clips if not c.asset_id]
                assigned_count = 0
                for i, clip in enumerate(clips_without_asset):
                    asset_idx = i % len(config_asset_ids)
                    try:
                        self.assign_asset(clip.id, config_asset_ids[asset_idx])
                        assigned_count += 1
                    except Exception as exc:
                        logger.warning(
                            "can_generate 最后防线: plan=%s clip=%s 分配素材 %s 失败: %s",
                            plan_id,
                            clip.id,
                            config_asset_ids[asset_idx],
                            exc,
                        )
                logger.info(
                    "can_generate 最后防线: plan=%s 已为 %d/%d 个片段分配素材",
                    plan_id,
                    assigned_count,
                    len(clips_without_asset),
                )
                # 重新加载 clips 验证分配结果
                clips = self._clip_repo.list_by_plan(plan_id)
                if not any(c.asset_id for c in clips):
                    return False, "没有可渲染的就绪片段，自动修复后仍未分配素材"
            else:
                logger.warning(
                    "can_generate 失败: plan=%s clips=%d 均无素材，且 config.asset_ids 为空，无法自动修复",
                    plan_id,
                    len(clips),
                )
                return False, "没有可渲染的就绪片段，请确保已选择素材"

        return True, ""

    def mark_clips_ready(self, plan_id: str) -> int:
        """将已分配素材的 pending 片段标记为 ready

        只标记同时满足以下条件的片段：
        - status == PENDING
        - asset_id 非空（已分配素材）

        Returns:
            int: 标记的片段数量
        """
        clips = self._clip_repo.list_by_plan(
            plan_id,
            status=EditPlanClipStatus.PENDING,
        )
        count = 0
        for clip in clips:
            if clip.asset_id:
                clip.mark_ready()
                self._clip_repo.update(clip)
                count += 1
        logger.info(
            "标记片段就绪: plan_id=%s marked=%d total_pending=%d",
            plan_id,
            count,
            len(clips),
        )
        return count

    def update_plan_config(self, plan_id: str, config_updates: Dict[str, Any]) -> EditPlan:
        """更新计划配置（合并更新）

        Args:
            plan_id: 计划 ID
            config_updates: 要合并的配置

        Returns:
            更新后的计划
        """
        plan = self.get_plan_or_raise(plan_id)
        # 自动从 completed/failed 切回 editing
        self._auto_resume_editing(plan_id)
        plan = self.get_plan_or_raise(plan_id)
        new_config = {**plan.config, **config_updates}

        updated = EditPlan(
            id=plan.id,
            template_id=plan.template_id,
            name=plan.name,
            status=plan.status,
            total_duration=plan.total_duration,
            source_edit_plan_id=plan.source_edit_plan_id,
            project_id=plan.project_id,
            created_by_user_id=plan.created_by_user_id,
            config=new_config,
            created_at=plan.created_at,
            updated_at=plan.updated_at,
        )
        return self._plan_repo.update(updated)