xiaoxia-saas/tests/unit/test_plan_generator.py

"""
PlanGeneratorService 单元测试

覆盖（6 组测试）：
- ONE_TAKE 模式：素材顺序分配给 main clips
- PIP 模式：第1个素材→main，其余→overlay
- VOICE_OVER 模式：素材→main clips (B-roll)
- VOICE_PIP 模式：第1个→background, 第2个→corner_voice, 其余→b_roll
- 无 clip_configs 时自动生成默认结构
- 空素材列表时 clips 创建但无素材分配
"""

from __future__ import annotations

import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Optional
from unittest.mock import MagicMock, patch

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")


sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

from packages.domain.edit_plan import EditPlan, EditPlanStatus
from packages.domain.edit_plan_clip import EditPlanClip, EditPlanClipStatus
from packages.domain.edit_template import EditTemplate, EditTemplateStatus
from packages.domain.editing_mode import EditingMode
from packages.domain.template_clip_config import ClipType, TemplateClipConfig, TransitionEffect

# ---------------------------------------------------------------------------
# Stub Repositories
# ---------------------------------------------------------------------------


class StubEditPlanRepository:
    """内存中的 EditPlan 仓储 stub"""

    def __init__(self) -> None:
        self._plans: dict[str, EditPlan] = {}
        self._counter = 0

    def _next_id(self) -> str:
        self._counter += 1
        return f"plan-{self._counter:03d}"

    def get(self, plan_id: str) -> Optional[EditPlan]:
        return self._plans.get(plan_id)

    def create(self, plan: EditPlan) -> EditPlan:
        if not plan.id:
            plan = EditPlan(
                id=self._next_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=plan.config,
                created_at=plan.created_at,
                updated_at=plan.updated_at,
            )
        self._plans[plan.id] = plan
        return plan

    def update(self, plan: EditPlan) -> EditPlan:
        self._plans[plan.id] = plan
        return plan

    def list_all(self, **kwargs) -> List[EditPlan]:
        return list(self._plans.values())

    def count(self, **kwargs) -> int:
        return len(self._plans)

    def delete(self, plan_id: str) -> bool:
        return self._plans.pop(plan_id, None) is not None


class StubEditPlanClipRepository:
    """内存中的 EditPlanClip 仓储 stub"""

    def __init__(self) -> None:
        self._clips: dict[str, EditPlanClip] = {}
        self._counter = 0

    def _next_id(self) -> str:
        self._counter += 1
        return f"clip-{self._counter:03d}"

    def get(self, clip_id: str) -> Optional[EditPlanClip]:
        return self._clips.get(clip_id)

    def create(self, clip: EditPlanClip) -> EditPlanClip:
        if not clip.id:
            clip = EditPlanClip(
                id=self._next_id(),
                plan_id=clip.plan_id,
                template_clip_config_id=clip.template_clip_config_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,
                status=clip.status,
                config=clip.config,
                created_at=clip.created_at,
                updated_at=clip.updated_at,
            )
        self._clips[clip.id] = clip
        return clip

    def update(self, clip: EditPlanClip) -> EditPlanClip:
        self._clips[clip.id] = clip
        return clip

    def list_by_plan(
        self,
        plan_id: str,
        *,
        status: Optional[EditPlanClipStatus] = None,
        skip: int = 0,
        limit: int = 100,
    ) -> List[EditPlanClip]:
        items = [c for c in self._clips.values() if c.plan_id == plan_id]
        items.sort(key=lambda c: c.order)
        if status:
            items = [c for c in items if c.status == status]
        return items[skip : skip + limit]

    def delete(self, clip_id: str) -> bool:
        return self._clips.pop(clip_id, None) is not None


# ---------------------------------------------------------------------------
# Helper: 构建 PlanGeneratorService（patch 仓储）
# ---------------------------------------------------------------------------


def _make_generator():
    """创建使用 stub 仓储的 PlanGeneratorService"""
    from apps.api.app.services.plan_generator_service import PlanGeneratorService

    plan_repo = StubEditPlanRepository()
    clip_repo = StubEditPlanClipRepository()

    with (
        patch(
            "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanRepository",
            return_value=plan_repo,
        ),
        patch(
            "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanClipRepository",
            return_value=clip_repo,
        ),
    ):
        db = MagicMock()
        svc = PlanGeneratorService(db)
        # 替换为 stub
        svc._plan_repo = plan_repo
        svc._clip_repo = clip_repo

    return svc, plan_repo, clip_repo


def _make_template(
    editing_mode: str = "one_take",
    config: Optional[dict] = None,
) -> EditTemplate:
    """创建测试用 EditTemplate"""
    return EditTemplate(
        id="tpl-001",
        name="测试模板",
        description="",
        template_type="default",
        editing_mode=editing_mode,
        config=config or {},
        preview_url="",
        sort_weight=0,
        status=EditTemplateStatus.ACTIVE,
        created_at=datetime.now(timezone.utc),
        updated_at=datetime.now(timezone.utc),
    )


def _make_clip_configs(
    template_id: str = "tpl-001",
    specs: Optional[List[dict]] = None,
) -> List[TemplateClipConfig]:
    """创建测试用 TemplateClipConfig 列表

    specs 示例: [{"clip_type": ClipType.INTRO, "order": 0}, ...]
    """
    if specs is None:
        specs = [
            {"clip_type": ClipType.INTRO, "order": 0, "min_duration": 2.0, "max_duration": 4.0},
            {"clip_type": ClipType.MAIN, "order": 1, "min_duration": 3.0, "max_duration": 7.0},
            {"clip_type": ClipType.OUTRO, "order": 2, "min_duration": 2.0, "max_duration": 4.0},
        ]
    configs = []
    for i, spec in enumerate(specs):
        cfg = TemplateClipConfig(
            id=f"cfg-{i:03d}",
            template_id=template_id,
            clip_type=spec.get("clip_type", ClipType.MAIN),
            order=spec.get("order", i),
            min_duration=spec.get("min_duration", 0.0),
            max_duration=spec.get("max_duration", 0.0),
            text_template=spec.get("text_template", ""),
            material_requirements=spec.get("material_requirements"),
            transition_effect=spec.get("transition_effect", TransitionEffect.CUT),
            config=spec.get("config"),
            created_at=datetime.now(timezone.utc),
            updated_at=datetime.now(timezone.utc),
        )
        configs.append(cfg)
    return configs


# ---------------------------------------------------------------------------
# 测试：ONE_TAKE 模式
# ---------------------------------------------------------------------------


class TestGenerateOneTakePlan:
    """ONE_TAKE 模式：素材顺序分配给 main clips"""

    def test_generate_one_take_plan(self):
        """3个clip_configs + 3个asset_ids → 按顺序分配"""
        svc, plan_repo, clip_repo = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
        clip_configs = _make_clip_configs()
        asset_ids = ["asset-1", "asset-2", "asset-3"]

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=asset_ids,
            project_id="proj-001",
            created_by_user_id="user-001",
        )

        plan = result["plan"]
        clips = result["clips"]

        assert plan.template_id == "tpl-001"
        assert plan.config["editing_mode"] == "one_take"
        assert plan.status == EditPlanStatus.EDITING
        assert len(clips) == 3

        # 按 order 排序后检查素材分配
        sorted_clips = sorted(clips, key=lambda c: c.order)
        # intro clip (order=0) 不是 main 类型，不分配素材
        # main clip (order=1) → asset-2（ONE_TAKE 只分配给 main clips）
        # outro clip (order=2) 不是 main 类型
        main_clips = [c for c in sorted_clips if c.clip_type == ClipType.MAIN.value]
        assert len(main_clips) == 1
        assert main_clips[0].asset_id == "asset-1"

    def test_one_take_plan_name_from_template(self):
        """name 为空时自动取模板名"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=["asset-1"],
        )

        assert "测试模板" in result["plan"].name

    def test_one_take_custom_name(self):
        """指定 name 时使用自定义名称"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=["asset-1"],
            name="我的剪辑",
        )

        assert result["plan"].name == "我的剪辑"


# ---------------------------------------------------------------------------
# 测试：PIP 模式
# ---------------------------------------------------------------------------


class TestGeneratePipPlan:
    """PIP 模式：第1个素材→main，其余→overlay"""

    def test_generate_pip_plan(self):
        """4个素材 → 第1个→main，其余→overlay"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.PIP.value)

        # 无 clip_configs，自动生成默认结构
        asset_ids = ["bg-asset", "overlay-1", "overlay-2", "overlay-3"]

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=asset_ids,
        )

        plan = result["plan"]
        clips = result["clips"]

        assert plan.config["editing_mode"] == "pip"
        # 自动生成: 1 main + 3 overlay
        assert len(clips) == 4

        main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
        overlay_clips = [c for c in clips if c.clip_type == "overlay"]

        assert len(main_clips) == 1
        assert len(overlay_clips) == 3

        # 第1个素材 → main
        assert main_clips[0].asset_id == "bg-asset"
        # 其余 → overlay
        assert overlay_clips[0].asset_id == "overlay-1"
        assert overlay_clips[1].asset_id == "overlay-2"
        assert overlay_clips[2].asset_id == "overlay-3"

    def test_pip_with_clip_configs(self):
        """PIP + 有 clip_configs（都是 MAIN 类型）→ 自动映射为 main + overlay，素材正确分配"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.PIP.value)

        # 4 个 MAIN 类型的 clip_config（模拟 PIP 模板配置）
        clip_configs = [
            TemplateClipConfig.create(
                template_id=template.id,
                clip_type=ClipType.MAIN,
                order=i,
                min_duration=3.0,
                max_duration=5.0,
            )
            for i in range(4)
        ]
        asset_ids = ["bg-asset", "overlay-1", "overlay-2", "overlay-3"]

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=asset_ids,
        )

        plan = result["plan"]
        clips = result["clips"]

        assert plan.config["editing_mode"] == "pip"
        assert len(clips) == 4

        # 第1个 MAIN → main（背景），其余 MAIN → overlay（画中画）
        main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
        overlay_clips = [c for c in clips if c.clip_type == "overlay"]

        assert len(main_clips) == 1
        assert len(overlay_clips) == 3

        # 素材分配正确
        assert main_clips[0].asset_id == "bg-asset"
        assert overlay_clips[0].asset_id == "overlay-1"
        assert overlay_clips[1].asset_id == "overlay-2"
        assert overlay_clips[2].asset_id == "overlay-3"

    def test_voice_pip_with_clip_configs(self):
        """VOICE_PIP + 有 clip_configs（都是 MAIN 类型）→ 自动映射为 background + corner_voice + b_roll"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.VOICE_PIP.value)

        # 4 个 MAIN 类型的 clip_config
        clip_configs = [
            TemplateClipConfig.create(
                template_id=template.id,
                clip_type=ClipType.MAIN,
                order=i,
                min_duration=3.0,
                max_duration=5.0,
            )
            for i in range(4)
        ]
        asset_ids = ["bg-asset", "voice-asset", "broll-1", "broll-2"]

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=asset_ids,
        )

        plan = result["plan"]
        clips = result["clips"]

        assert plan.config["editing_mode"] == "voice_pip"
        assert len(clips) == 4

        bg_clips = [c for c in clips if c.clip_type == "background"]
        corner_clips = [c for c in clips if c.clip_type == "corner_voice"]
        broll_clips = [c for c in clips if c.clip_type == "b_roll"]

        assert len(bg_clips) == 1
        assert len(corner_clips) == 1
        assert len(broll_clips) == 2

        # 素材分配正确
        assert bg_clips[0].asset_id == "bg-asset"
        assert corner_clips[0].asset_id == "voice-asset"
        assert broll_clips[0].asset_id == "broll-1"
        assert broll_clips[1].asset_id == "broll-2"


# ---------------------------------------------------------------------------
# 测试：VOICE_OVER 模式
# ---------------------------------------------------------------------------


class TestGenerateVoiceOverPlan:
    """VOICE_OVER 模式：素材→main clips (B-roll)"""

    def test_generate_voice_over_plan(self):
        """3个素材 → 3个 main clips，每个标记为 b_roll"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.VOICE_OVER.value)
        asset_ids = ["video-1", "video-2", "video-3"]

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=asset_ids,
        )

        plan = result["plan"]
        clips = result["clips"]

        assert plan.config["editing_mode"] == "voice_over"
        assert len(clips) == 3

        # 所有 clips 都是 main 类型
        for clip in clips:
            assert clip.clip_type == ClipType.MAIN.value

        # 素材按顺序分配
        assert clips[0].asset_id == "video-1"
        assert clips[1].asset_id == "video-2"
        assert clips[2].asset_id == "video-3"

        # 每个 clip 的 config 标记为 b_roll
        for clip in clips:
            assert clip.config.get("role") == "b_roll"


# ---------------------------------------------------------------------------
# 测试：VOICE_PIP 模式
# ---------------------------------------------------------------------------


class TestGenerateVoicePipPlan:
    """VOICE_PIP 模式：第1个→background, 第2个→corner_voice, 其余→b_roll"""

    def test_generate_voice_pip_plan(self):
        """4个素材 → background + corner_voice + 2 b_roll"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.VOICE_PIP.value)
        asset_ids = ["bg-video", "corner-video", "broll-1", "broll-2"]

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=asset_ids,
        )

        plan = result["plan"]
        clips = result["clips"]

        assert plan.config["editing_mode"] == "voice_pip"
        assert len(clips) == 4

        bg_clips = [c for c in clips if c.clip_type == "background"]
        corner_clips = [c for c in clips if c.clip_type == "corner_voice"]
        broll_clips = [c for c in clips if c.clip_type == "b_roll"]

        assert len(bg_clips) == 1
        assert len(corner_clips) == 1
        assert len(broll_clips) == 2

        # 素材分配
        assert bg_clips[0].asset_id == "bg-video"
        assert corner_clips[0].asset_id == "corner-video"
        assert broll_clips[0].asset_id == "broll-1"
        assert broll_clips[1].asset_id == "broll-2"


# ---------------------------------------------------------------------------
# 测试：无 clip_configs 时自动生成默认结构
# ---------------------------------------------------------------------------


class TestGenerateWithoutClipConfigs:
    """无 clip_configs 时根据 editing_mode 生成默认 clip 结构"""

    def test_one_take_default_clips(self):
        """ONE_TAKE + 3个素材 → 3个 main clips"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=["a1", "a2", "a3"],
        )

        clips = result["clips"]
        assert len(clips) == 3
        for clip in clips:
            assert clip.clip_type == ClipType.MAIN.value

    def test_pip_default_clips(self):
        """PIP + 3个素材 → 1 main + 2 overlay"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.PIP.value)

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=["a1", "a2", "a3"],
        )

        clips = result["clips"]
        assert len(clips) == 3
        main_clips = [c for c in clips if c.clip_type == ClipType.MAIN.value]
        overlay_clips = [c for c in clips if c.clip_type == "overlay"]
        assert len(main_clips) == 1
        assert len(overlay_clips) == 2

    def test_voice_pip_default_clips(self):
        """VOICE_PIP + 4个素材 → 1 background + 1 corner_voice + 2 b_roll"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.VOICE_PIP.value)

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=["a1", "a2", "a3", "a4"],
        )

        clips = result["clips"]
        assert len(clips) == 4


# ---------------------------------------------------------------------------
# 测试：空素材列表
# ---------------------------------------------------------------------------


class TestGenerateEmptyAssets:
    """空素材列表时 clips 创建但无素材分配"""

    def test_empty_assets(self):
        """空 asset_ids → clips 创建但 asset_id 为空"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
        clip_configs = _make_clip_configs()

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=[],
        )

        plan = result["plan"]
        clips = result["clips"]

        assert plan.total_duration > 0  # clips 有默认时长
        assert len(clips) == 3
        for clip in clips:
            assert clip.asset_id == ""

    def test_empty_assets_pip(self):
        """PIP 模式空素材 → 1个 main clip（至少1个）"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.PIP.value)

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=[],
        )

        clips = result["clips"]
        # 至少1个 main clip（n = max(asset_count, 1) = 1）
        assert len(clips) == 1
        assert clips[0].clip_type == ClipType.MAIN.value


# ---------------------------------------------------------------------------
# 测试：plan config 继承模板配置
# ---------------------------------------------------------------------------


class TestPlanConfigInheritance:
    """plan config 继承模板的 cover/title/subtitle/bgm"""

    def test_inherit_template_config(self):
        """模板有 cover/title/bgm 配置 → plan 继承"""
        svc, _, _ = _make_generator()
        template_config = {
            "editing_mode": "one_take",
            "cover": {"type": "ai_frame"},
            "title": {"text": "测试标题", "font": "思源黑体"},
            "bgm": {"url": "https://example.com/bgm.mp3"},
        }
        template = _make_template(
            editing_mode=EditingMode.ONE_TAKE.value,
            config=template_config,
        )

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=["a1"],
        )

        plan_config = result["plan"].config
        assert plan_config["editing_mode"] == "one_take"
        assert plan_config["cover"]["type"] == "ai_frame"
        assert plan_config["title"]["text"] == "测试标题"
        assert plan_config["bgm"]["url"] == "https://example.com/bgm.mp3"


# ---------------------------------------------------------------------------
# 测试：total_duration 计算
# ---------------------------------------------------------------------------


class TestTotalDuration:
    """total_duration 正确计算"""

    def test_duration_from_clip_configs(self):
        """有 clip_configs 时，duration 取 min/max 中间值"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
        clip_configs = _make_clip_configs(
            specs=[
                {"clip_type": ClipType.INTRO, "order": 0, "min_duration": 2.0, "max_duration": 4.0},
                {"clip_type": ClipType.MAIN, "order": 1, "min_duration": 4.0, "max_duration": 6.0},
                {"clip_type": ClipType.OUTRO, "order": 2, "min_duration": 2.0, "max_duration": 4.0},
            ]
        )

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=["a1"],
        )

        # intro: (2+4)/2=3, main: (4+6)/2=5, outro: (2+4)/2=3 → total=11
        assert result["plan"].total_duration == 11.0

    def test_duration_from_default_clips(self):
        """无 clip_configs 时，每个 clip 默认 5 秒"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)

        result = svc.generate_from_template(
            template=template,
            clip_configs=[],
            asset_ids=["a1", "a2", "a3"],
        )

        # 3 个 clips × 5 秒 = 15 秒
        assert result["plan"].total_duration == 15.0


# ---------------------------------------------------------------------------
# 测试：模板配置传递（export / filter / 片段级 speed_ratio）
# ---------------------------------------------------------------------------


class TestTemplateConfigPropagation:
    """模板配置到剪辑计划的传递验证"""

    def test_export_config_propagated(self):
        """模板 export 配置（watermark/resolution）应传递到 plan.config.export"""
        svc, _, _ = _make_generator()
        template = _make_template(
            editing_mode=EditingMode.ONE_TAKE.value,
            config={
                "export": {
                    "resolution": "1080x1920",
                    "fps": 30,
                    "watermark_enabled": True,
                    "watermark_text": "小虾剪辑",
                },
            },
        )
        clip_configs = _make_clip_configs(
            specs=[
                {"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0},
            ]
        )

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=["a1"],
        )

        export_cfg = result["plan"].config.get("export", {})
        assert export_cfg.get("watermark_enabled") is True
        assert export_cfg.get("watermark_text") == "小虾剪辑"
        assert export_cfg.get("resolution") == "1080x1920"
        assert export_cfg.get("fps") == 30

    def test_filter_config_propagated(self):
        """模板 filter 配置应传递到 plan.config.filter"""
        svc, _, _ = _make_generator()
        template = _make_template(
            editing_mode=EditingMode.ONE_TAKE.value,
            config={
                "filter": {
                    "enabled": True,
                    "preset_id": "filter_vivid",
                    "intensity": 80,
                },
            },
        )
        clip_configs = _make_clip_configs(
            specs=[
                {"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0},
            ]
        )

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=["a1"],
        )

        filter_cfg = result["plan"].config.get("filter", {})
        assert filter_cfg.get("enabled") is True
        assert filter_cfg.get("preset_id") == "filter_vivid"
        assert filter_cfg.get("intensity") == 80

    def test_subtitle_asr_config_propagated(self):
        """模板 subtitle ASR 配置应传递到 plan.config.subtitle"""
        svc, _, _ = _make_generator()
        template = _make_template(
            editing_mode=EditingMode.ONE_TAKE.value,
            config={
                "subtitle": {
                    "enabled": True,
                    "auto_generated": True,
                    "position": "bottom",
                    "source": "asr",
                },
            },
        )
        clip_configs = _make_clip_configs(
            specs=[
                {"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0},
            ]
        )

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=["a1"],
        )

        subtitle_cfg = result["plan"].config.get("subtitle", {})
        assert subtitle_cfg.get("enabled") is True
        assert subtitle_cfg.get("auto_generated") is True
        assert subtitle_cfg.get("position") == "bottom"

    def test_speed_ratio_mapped_to_playback_speed(self):
        """clip_config.config.speed_ratio 应映射到 clip.playback_speed"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
        clip_configs = _make_clip_configs(
            specs=[
                {
                    "clip_type": ClipType.MAIN,
                    "order": 0,
                    "min_duration": 3.0,
                    "max_duration": 5.0,
                    "config": {"speed_ratio": 1.2, "name": "开场"},
                },
                {
                    "clip_type": ClipType.MAIN,
                    "order": 1,
                    "min_duration": 3.0,
                    "max_duration": 5.0,
                    "config": {"speed_ratio": 0.8, "name": "结尾"},
                },
            ]
        )

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=["a1", "a2"],
        )

        clips = result["clips"]
        assert clips[0].playback_speed == 1.2
        assert clips[1].playback_speed == 0.8

    def test_playback_speed_takes_priority_over_speed_ratio(self):
        """clip_config.config.playback_speed 优先于 speed_ratio"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
        clip_configs = _make_clip_configs(
            specs=[
                {
                    "clip_type": ClipType.MAIN,
                    "order": 0,
                    "min_duration": 3.0,
                    "max_duration": 5.0,
                    "config": {"speed_ratio": 1.2, "playback_speed": 1.5},
                },
            ]
        )

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=["a1"],
        )

        assert result["clips"][0].playback_speed == 1.5

    def test_default_playback_speed_is_1_0(self):
        """无 speed_ratio / playback_speed 时，默认 1.0"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
        clip_configs = _make_clip_configs(
            specs=[
                {"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0},
            ]
        )

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=["a1"],
        )

        assert result["clips"][0].playback_speed == 1.0

    def test_clip_config_dict_propagated(self):
        """clip_config.config 整体应传递到 clip.config"""
        svc, _, _ = _make_generator()
        template = _make_template(editing_mode=EditingMode.ONE_TAKE.value)
        clip_configs = _make_clip_configs(
            specs=[
                {
                    "clip_type": ClipType.MAIN,
                    "order": 0,
                    "min_duration": 3.0,
                    "max_duration": 5.0,
                    "config": {"speed_ratio": 1.2, "name": "开场", "custom_field": "value"},
                },
            ]
        )

        result = svc.generate_from_template(
            template=template,
            clip_configs=clip_configs,
            asset_ids=["a1"],
        )

        clip_cfg = result["clips"][0].config
        assert clip_cfg.get("speed_ratio") == 1.2
        assert clip_cfg.get("name") == "开场"
        assert clip_cfg.get("custom_field") == "value"


class TestAssetDurationsAlwaysFetched:
    """验证 asset_durations 不再受 random_preview 条件限制。

    修复前：asset_durations 仅在 random_preview=True 时传入 distribute_assets
    修复后：只要 _asset_repo 存在，就始终获取 asset_durations
    """

    def _make_service_with_asset_repo(self):
        """创建带 mock asset_repo 的 PlanGeneratorService"""
        from apps.api.app.services.plan_generator_service import PlanGeneratorService

        plan_repo = StubEditPlanRepository()
        clip_repo = StubEditPlanClipRepository()

        # mock asset_repo: 返回带 duration 的素材
        asset_repo = MagicMock()

        def fake_get(asset_id):
            mock_asset = MagicMock()
            mock_asset.duration = 30.0  # 每个素材 30 秒
            # score_asset 所需的属性
            mock_asset.quality_score = None
            mock_asset.created_at = None
            mock_asset.metadata = {}
            return mock_asset

        asset_repo.get = MagicMock(side_effect=fake_get)

        with (
            patch(
                "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanRepository",
                return_value=plan_repo,
            ),
            patch(
                "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanClipRepository",
                return_value=clip_repo,
            ),
        ):
            db = MagicMock()
            svc = PlanGeneratorService(db, asset_repo=asset_repo)
            svc._plan_repo = plan_repo
            svc._clip_repo = clip_repo

        return svc, asset_repo

    def test_asset_durations_fetched_without_random_preview(self):
        """random_preview=False 时也应获取 asset_durations"""
        svc, asset_repo = self._make_service_with_asset_repo()

        template = _make_template("one_take")
        clip_configs = _make_clip_configs(
            template_id=template.id,
            specs=[{"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}],
        )

        # patch distribute_assets 以捕获传入的参数
        with patch("apps.api.app.services.plan_generator_service.distribute_assets") as mock_distribute:
            svc.generate_from_template(
                template=template,
                clip_configs=clip_configs,
                asset_ids=["a1", "a2"],
                random_preview=False,  # 关键：非随机预览模式
            )

            # 验证 asset_durations 被传入（不是 None）
            mock_distribute.assert_called_once()
            call_kwargs = mock_distribute.call_args
            asset_durations = call_kwargs.kwargs.get("asset_durations", call_kwargs[1].get("asset_durations"))
            assert asset_durations is not None, "asset_durations should be fetched even when random_preview=False"
            assert "a1" in asset_durations
            assert "a2" in asset_durations
            assert asset_durations["a1"] == 30.0

        # 验证 asset_repo.get 被调用（说明 _fetch_asset_durations 执行了）
        assert asset_repo.get.call_count >= 2

    def test_asset_durations_fetched_with_random_preview(self):
        """random_preview=True 时仍正常获取 asset_durations（行为不变）"""
        svc, asset_repo = self._make_service_with_asset_repo()

        template = _make_template("one_take")
        clip_configs = _make_clip_configs(
            template_id=template.id,
            specs=[{"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}],
        )

        with patch("apps.api.app.services.plan_generator_service.distribute_assets") as mock_distribute:
            svc.generate_from_template(
                template=template,
                clip_configs=clip_configs,
                asset_ids=["a1"],
                random_preview=True,
            )

            call_kwargs = mock_distribute.call_args
            asset_durations = call_kwargs.kwargs.get("asset_durations", call_kwargs[1].get("asset_durations"))
            assert asset_durations is not None
            assert "a1" in asset_durations

    def test_no_asset_repo_means_no_durations(self):
        """_asset_repo 为 None 时 asset_durations 应为 None"""
        svc, _, _ = _make_generator()  # 默认不带 asset_repo

        template = _make_template("one_take")
        clip_configs = _make_clip_configs(
            template_id=template.id,
            specs=[{"clip_type": ClipType.MAIN, "order": 0, "min_duration": 3.0, "max_duration": 5.0}],
        )

        with patch("apps.api.app.services.plan_generator_service.distribute_assets") as mock_distribute:
            svc.generate_from_template(
                template=template,
                clip_configs=clip_configs,
                asset_ids=["a1"],
            )

            call_kwargs = mock_distribute.call_args
            asset_durations = call_kwargs.kwargs.get("asset_durations", call_kwargs[1].get("asset_durations"))
            assert asset_durations is None


# ---------------------------------------------------------------------------
# 测试：正式生成片段随机重排（Issue #1663）
# ---------------------------------------------------------------------------


class TestFormalGenerationShuffle:
    """验证正式生成时片段顺序随机化。

    Issue #1663: 正式生成时 smart_match 排序后对 asset_ids 做 random.shuffle，
    使得同一批素材每次生成的视频片段顺序不同，有利于查重降重。
    """

    def _make_service_with_asset_repo(self):
        """创建带 mock asset_repo 的 PlanGeneratorService（复用 TestAssetDurationsAlwaysFetched 模式）"""
        from apps.api.app.services.plan_generator_service import PlanGeneratorService

        plan_repo = StubEditPlanRepository()
        clip_repo = StubEditPlanClipRepository()

        asset_repo = MagicMock()

        def fake_get(asset_id):
            mock_asset = MagicMock()
            mock_asset.duration = 30.0
            mock_asset.quality_score = None
            mock_asset.created_at = None
            mock_asset.metadata = {}
            return mock_asset

        asset_repo.get = MagicMock(side_effect=fake_get)

        with (
            patch(
                "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanRepository",
                return_value=plan_repo,
            ),
            patch(
                "apps.api.app.services.plan_generator_service.SQLAlchemyEditPlanClipRepository",
                return_value=clip_repo,
            ),
        ):
            db = MagicMock()
            svc = PlanGeneratorService(db, asset_repo=asset_repo)
            svc._plan_repo = plan_repo
            svc._clip_repo = clip_repo

        return svc, asset_repo

    def test_formal_generation_shuffles_asset_ids(self):
        """正式生成路径下 asset_ids 应被打乱，多次调用顺序应不同"""
        svc, _ = self._make_service_with_asset_repo()

        template = _make_template("one_take")
        # 6 个 clip 容纳 6 个素材
        clip_configs = _make_clip_configs(
            template_id=template.id,
            specs=[
                {"clip_type": ClipType.MAIN, "order": i, "min_duration": 3.0, "max_duration": 5.0} for i in range(6)
            ],
        )

        asset_ids = ["a1", "a2", "a3", "a4", "a5", "a6"]

        # 收集多次调用中 distribute_assets 收到的 asset_ids 顺序
        captured_orders = []
        with patch(
            "apps.api.app.services.plan_generator_service.distribute_assets",
            side_effect=lambda clips, asset_ids, *a, **kw: captured_orders.append(list(asset_ids)),
        ):
            # mock _sort_assets_by_smart_score 返回固定顺序，验证 shuffle 会打乱
            with patch.object(
                svc,
                "_sort_assets_by_smart_score",
                side_effect=lambda ids: list(ids),  # 原样返回
            ):
                with patch.object(
                    svc,
                    "_fetch_asset_scene_points",
                    return_value={},
                ):
                    for _ in range(10):
                        svc.generate_from_template(
                            template=template,
                            clip_configs=clip_configs,
                            asset_ids=list(asset_ids),  # 每次传新列表
                            random_preview=False,  # 正式生成
                        )

        assert len(captured_orders) == 10
        # 每次 order 应该是 asset_ids 的一个排列
        expected_set = set(asset_ids)
        for order in captured_orders:
            assert set(order) == expected_set

        # 10 次调用中应至少出现 2 种不同顺序（概率 > 99.9%）
        unique_orders = set(tuple(o) for o in captured_orders)
        assert (
            len(unique_orders) >= 2
        ), f"Expected shuffled orders to vary, but got only {len(unique_orders)} unique order(s): {unique_orders}"

    def test_formal_generation_does_not_mutate_original_list(self):
        """shuffle 不应修改调用方的原始 asset_ids 列表"""
        svc, _ = self._make_service_with_asset_repo()

        template = _make_template("one_take")
        clip_configs = _make_clip_configs(
            template_id=template.id,
            specs=[
                {"clip_type": ClipType.MAIN, "order": i, "min_duration": 3.0, "max_duration": 5.0} for i in range(4)
            ],
        )

        original = ["a1", "a2", "a3", "a4"]
        original_copy = list(original)

        with patch("apps.api.app.services.plan_generator_service.distribute_assets"):
            with patch.object(svc, "_sort_assets_by_smart_score", side_effect=lambda ids: list(ids)):
                with patch.object(svc, "_fetch_asset_scene_points", return_value={}):
                    svc.generate_from_template(
                        template=template,
                        clip_configs=clip_configs,
                        asset_ids=original,
                        random_preview=False,
                    )

        assert original == original_copy, "Original asset_ids list should not be mutated"

    def test_preview_random_mode_unaffected_by_shuffle(self):
        """预览随机模式不走 shuffle 路径，行为不变"""
        svc, _ = self._make_service_with_asset_repo()

        template = _make_template("one_take")
        clip_configs = _make_clip_configs(
            template_id=template.id,
            specs=[
                {"clip_type": ClipType.MAIN, "order": i, "min_duration": 3.0, "max_duration": 5.0} for i in range(4)
            ],
        )

        asset_ids = ["a1", "a2", "a3", "a4"]

        captured_orders = []
        with patch(
            "apps.api.app.services.plan_generator_service.distribute_assets",
            side_effect=lambda clips, asset_ids, *a, **kw: captured_orders.append(list(asset_ids)),
        ):
            for _ in range(5):
                svc.generate_from_template(
                    template=template,
                    clip_configs=clip_configs,
                    asset_ids=list(asset_ids),
                    random_preview=True,  # 预览随机模式
                )

        assert len(captured_orders) == 5
        # 预览模式下 random.shuffle 不应被调用（在 _distribute_assets 的 if not random_selection 块内）
        # 所以 asset_ids 应该保持调用方传入的顺序（可能已由上层 shuffle 过）