xiaoxia-saas/tests/unit/test_variant_plan_selector_1743.py

"""#1743 批量变体独立选片纯核心测试（packages/domain/variant_plan_selector.py）。

覆盖：
- 固定种子下 N 次独立选片：素材组合/片段顺序/起点显著不同（降重核心）
- 批次内同素材区间重叠 >20% 触发避让重选
- 异常输入：空源片段/空素材池/时长全 0 → ValueError（严禁退回同源）
- 非 main 片段（intro/outro/overlay）保留源骨架素材，仅重算起点
- 跨变体 batch_segments 就地累加（串行选片天然避让）
- 文案/转场/速度等骨架字段透传
"""

from __future__ import annotations

import random
import sys
from pathlib import Path
from unittest.mock import patch

import pytest

REPO_ROOT = Path(__file__).resolve().parents[2]
for sub in ("packages", ""):
    p = str(REPO_ROOT / sub) if sub else str(REPO_ROOT)
    if p not in sys.path:
        sys.path.insert(0, p)

from packages.domain import variant_plan_selector as vps  # noqa: E402
from packages.domain.variant_plan_selector import (  # noqa: E402
    BATCH_CLIP_OVERLAP_LIMIT,
    _clip_overlap_ratio,
    reselect_clips_for_variant,
)


def _source_clips(assets=("a1", "a2", "a3"), dur=5.0, with_non_main=False):
    """构造源片段骨架：main 片段若干，可选 intro/outro 固定角色片段。"""
    clips = []
    order = 0
    if with_non_main:
        clips.append(
            {
                "order": order,
                "asset_id": "intro_asset",
                "start_time": 0.0,
                "duration": 3.0,
                "clip_type": "intro",
                "playback_speed": 1.0,
                "transition_effect": "fade",
                "transition_duration": 0.5,
                "text_content": "片头",
                "config": {"role": "intro"},
            }
        )
        order += 1
    for i, aid in enumerate(assets):
        clips.append(
            {
                "order": order,
                "asset_id": aid,
                "start_time": float(i * 10),
                "duration": dur,
                "clip_type": "main",
                "playback_speed": 1.2,
                "transition_effect": "cut",
                "transition_duration": 0.0,
                "text_content": f"文案{i}",
                "config": {},
            }
        )
        order += 1
    if with_non_main:
        clips.append(
            {
                "order": order,
                "asset_id": "outro_asset",
                "start_time": 0.0,
                "duration": 2.0,
                "clip_type": "outro",
                "playback_speed": 1.0,
                "transition_effect": "fade",
                "transition_duration": 0.5,
                "text_content": "片尾",
                "config": {"role": "outro"},
            }
        )
    return clips


def _durations(asset_ids, total=120.0, extra=None):
    d = {a: total for a in asset_ids}
    if extra:
        d.update(extra)
    return d


class TestReselectClipsValidation:
    def test_empty_source_clips_raises(self):
        with pytest.raises(ValueError, match="源 plan 无片段"):
            reselect_clips_for_variant(
                [],
                ["a1"],
                asset_durations={"a1": 60.0},
                rng=random.Random(1),
            )

    def test_empty_asset_pool_raises(self):
        with pytest.raises(ValueError, match="素材池为空"):
            reselect_clips_for_variant(
                _source_clips(),
                [],
                asset_durations={},
                rng=random.Random(1),
            )

    def test_all_zero_duration_raises(self):
        with pytest.raises(ValueError, match="时长全部未知"):
            reselect_clips_for_variant(
                _source_clips(),
                ["a1", "a2"],
                asset_durations={"a1": 0.0, "a2": 0.0},
                rng=random.Random(1),
            )


class TestReselectClipsDifferentiation:
    def test_three_variants_differ_in_assets_order_and_starts(self):
        """核心验收：固定种子连续 3 次独立选片，素材组合/顺序/起点显著不同。"""
        source = _source_clips(assets=("a1", "a2", "a3", "a4"))
        pool = ["a1", "a2", "a3", "a4", "a5", "a6"]
        durations = _durations(pool, total=300.0)

        batch_segments: dict = {}
        variants = []
        for seed in range(3):
            clips = reselect_clips_for_variant(
                source,
                pool,
                asset_durations=durations,
                batch_segments=batch_segments,  # 串行调用：上一变体区间参与避让
                rng=random.Random(100 + seed),
            )
            variants.append(clips)

        main_sequences = []
        for clips in variants:
            main = [c for c in clips if c["clip_type"] == "main"]
            main_sequences.append([(c["asset_id"], round(c["start_time"], 2)) for c in main])

        # 1) 每个变体片段数与源骨架一致
        for clips in variants:
            main_clips = [c for c in clips if c["clip_type"] == "main"]
            assert len(main_clips) == 4

        # 2) 三个变体的素材序列不全相同（素材洗牌 + main 顺序洗牌生效）
        seq_sets = {tuple(a for a, _ in seq) for seq in main_sequences}
        assert len(seq_sets) >= 2, f"变体素材序列应存在差异，实际全部相同: {seq_sets}"

        # 3) 起点组合不全相同（起点重选生效）
        start_sets = {tuple(s for _, s in seq) for seq in main_sequences}
        assert len(start_sets) >= 2, f"变体起点组合应存在差异，实际全部相同: {start_sets}"

        # 4) batch_segments 跨变体累加（串行避让链路存在）
        total_segments = sum(len(v) for v in batch_segments.values())
        assert total_segments >= 12, f"3 变体 × 4 main 片段应累加 >=12 区间，实际 {total_segments}"

    def test_skeleton_fields_preserved(self):
        """文案/转场/速度等骨架字段随片段透传（只换素材与起点）。"""
        source = _source_clips(assets=("a1", "a2"), with_non_main=True)
        pool = ["a1", "a2", "a3"]
        durations = _durations(pool, total=120.0, extra={"intro_asset": 30.0, "outro_asset": 30.0})

        clips = reselect_clips_for_variant(
            source,
            pool,
            asset_durations=durations,
            rng=random.Random(7),
        )
        by_order = {c["order"]: c for c in clips}

        # intro/outro 骨架字段保留
        intro = next(c for c in clips if c["clip_type"] == "intro")
        outro = next(c for c in clips if c["clip_type"] == "outro")
        assert intro["asset_id"] == "intro_asset"
        assert intro["text_content"] == "片头"
        assert intro["transition_effect"] == "fade"
        assert intro["transition_duration"] == 0.5
        assert outro["asset_id"] == "outro_asset"
        assert outro["text_content"] == "片尾"

        # main 片段文案/速度随骨架 order 保留
        for c in clips:
            if c["clip_type"] == "main":
                assert c["playback_speed"] == 1.2
                assert c["text_content"].startswith("文案")


class TestBatchOverlapAvoidance:
    def test_overlap_ratio_calculation(self):
        segs = {"a1": [(10.0, 20.0)]}  # 已占 10s 区间
        # 新区间 [10,20) 完全重叠 → 1.0
        assert _clip_overlap_ratio("a1", 10.0, 10.0, segs) == pytest.approx(1.0)
        # 新区间 [20,30) 零重叠 → 0.0
        assert _clip_overlap_ratio("a1", 20.0, 10.0, segs) == pytest.approx(0.0)
        # 新区间 [15,25) 重叠 5s / 10s → 0.5
        assert _clip_overlap_ratio("a1", 15.0, 10.0, segs) == pytest.approx(0.5)
        # 空 asset / 零时长 → 0
        assert _clip_overlap_ratio("", 0.0, 10.0, segs) == 0.0
        assert _clip_overlap_ratio("a1", 10.0, 0.0, segs) == 0.0

    def test_over_limit_triggers_reselect_to_non_overlapping(self):
        """批次已占满素材前段时，避让重选应把起点挪到重叠 ≤20% 的位置。"""
        source = _source_clips(assets=("a1",), dur=10.0)
        durations = {"a1": 120.0}
        # 批次已选区间：a1 [0, 100) 几乎占满前段
        batch_segments = {"a1": [(0.0, 100.0)]}

        # _resolve_start_time 第一次返回高重叠起点（2.0），之后 rng 抖动应找到低重叠位置
        call_count = {"n": 0}

        def fake_resolve(asset_id, clip_duration, asset_durations, used_segments, scene_points=None, on_exhausted=None):
            call_count["n"] += 1
            return 2.0 if call_count["n"] == 1 else None  # 后续回退 rng.uniform

        with patch.object(vps, "_resolve_start_time", side_effect=fake_resolve):
            # rng.uniform 返回 105.0（与 [0,100) 零重叠）；rng 是 random.Random 实例，
            # 需 patch 类方法 uniform 才能生效
            with patch.object(random.Random, "uniform", return_value=105.0):
                clips = reselect_clips_for_variant(
                    source,
                    ["a1"],
                    asset_durations=durations,
                    batch_segments=batch_segments,
                    rng=random.Random(3),
                )

        main = [c for c in clips if c["clip_type"] == "main"]
        assert len(main) == 1
        start = main[0]["start_time"]
        ratio = _clip_overlap_ratio("a1", start, 10.0, {"a1": [(0.0, 100.0)]})
        assert ratio <= BATCH_CLIP_OVERLAP_LIMIT, f"避让后重叠应 ≤20%，实际 {ratio:.2%}"
        # #1749：选片改 0.25s 窗口扫描找最优错开起点（不再依赖 rng.uniform 单点），
        # 落点只需满足语义：与已用区间 [0,100) 不完全重叠且重叠 ≤20%
        assert start > 0.0
        # start+10 区间与 [0,100) 的重叠段（若有）≤ 2s
        overlap = max(0.0, min(start + 10.0, 100.0) - max(start, 0.0))
        assert overlap <= 2.0 + 1e-6, f"重叠时长 {overlap}s 应 ≤2s（10s 段的 20%）"

    def test_first_variant_segments_become_avoidance_target(self):
        """变体 0 选定区间后，变体 1 选同素材时批次区间生效（不与源区间完全重合）。"""
        source = _source_clips(assets=("a1", "a2"), dur=8.0)
        pool = ["a1", "a2", "a3"]
        durations = _durations(pool, total=600.0)

        batch_segments: dict = {}
        v1 = reselect_clips_for_variant(
            source, pool, asset_durations=durations, batch_segments=batch_segments, rng=random.Random(11)
        )
        v1_main = [(c["asset_id"], c["start_time"], c["duration"]) for c in v1 if c["clip_type"] == "main"]

        # 快照 v1 之后的批次避让集（v2 调用会就地追加 v2 自身区间，断言必须用调用前快照）
        import copy

        batch_snapshot = copy.deepcopy(batch_segments)

        v2 = reselect_clips_for_variant(
            source, pool, asset_durations=durations, batch_segments=batch_segments, rng=random.Random(12)
        )
        # v2 与 v1（快照）同素材片段的区间重叠占比均 ≤20%
        for c in v2:
            if c["clip_type"] != "main" or not c["asset_id"]:
                continue
            ratio = _clip_overlap_ratio(c["asset_id"], c["start_time"], c["duration"], batch_snapshot)
            assert (
                ratio <= BATCH_CLIP_OVERLAP_LIMIT + 1e-6
            ), f"变体间片段重叠超限: asset={c['asset_id']} start={c['start_time']} ratio={ratio:.2%}"
        # v1 片段确实进入了批次避让集
        assert any(a in batch_segments for a, _, _ in v1_main)