xiaoxia-saas/tests/unit/test_smart_match_integration.py

"""测试 smart_match 评分集成到素材选取路径。

验证：
- 使用次数多的素材评分低于使用次数少的（unused 维度降权生效）
- from-assets 路径中 sorted_candidates 按 smart_match 评分排序
- 一键生成路径中 _sort_assets_by_smart_score 按评分降序
"""

from __future__ import annotations

import os
import random
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
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"))

import pytest

from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX, score_asset, smart_select_assets

# ── 辅助工厂 ──────────────────────────────────────────────────────────────────


@dataclass
class FakeAsset:
    """Minimal Asset-like object."""

    id: str
    quality_score: float | None = 70.0
    duration: float = 15.0
    created_at: datetime | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    status: str = "ready"
    file_type: str = "video"


class _ZeroNoiseRandom(random.Random):
    """零噪声随机源：uniform(0, NOISE_MAX) 恒返回 0，消除排序随机性；
    其他随机调用（非噪声区间）保持正常随机行为。"""

    def uniform(self, a, b):
        from packages.domain.smart_match import SCORE_RANDOM_NOISE_MAX

        if a == 0.0 and b == SCORE_RANDOM_NOISE_MAX:
            return 0.0
        return super().uniform(a, b)


def _asset_with_use_count(asset_id: str, use_count: int) -> FakeAsset:
    """创建指定使用次数的素材，其他维度保持一致。"""
    return FakeAsset(
        id=asset_id,
        quality_score=70.0,
        duration=15.0,  # 最优区间 5-30s
        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
        metadata={"generation_use_count": use_count},
    )


# ── score_asset 单元测试：unused 维度降权 ─────────────────────────────────────


class TestScoreAssetUnusedDiminsh:
    """验证 unused 维度：使用次数越多，评分越低。"""

    def test_unused_scores_higher_than_used(self):
        """use_count=0 的素材评分高于 use_count>0 的。"""
        fresh = _asset_with_use_count("fresh", 0)
        used = _asset_with_use_count("used", 1)
        fresh_score, _ = score_asset(fresh)
        used_score, _ = score_asset(used)
        assert fresh_score > used_score

    def test_high_use_count_scores_lower_than_low(self):
        """use_count=5 的素材评分低于 use_count=1 的。"""
        low_use = _asset_with_use_count("low", 1)
        high_use = _asset_with_use_count("high", 5)
        low_score, _ = score_asset(low_use)
        high_score, _ = score_asset(high_use)
        assert low_score > high_score

    def test_unused_breakdown_values(self):
        """验证 unused 维度的具体分值。"""
        fresh = _asset_with_use_count("fresh", 0)
        low = _asset_with_use_count("low", 2)
        high = _asset_with_use_count("high", 10)

        _, fresh_bd = score_asset(fresh)
        _, low_bd = score_asset(low)
        _, high_bd = score_asset(high)

        # use_count=0 → unused_score=100 → component=10.0
        assert fresh_bd["unused"] == 10.0
        # use_count=2 → unused_score=70 → component=7.0
        assert low_bd["unused"] == 7.0
        # use_count=10 → unused_score=30 → component=3.0
        assert high_bd["unused"] == 3.0

    def test_monotonically_decreasing_scores(self):
        """使用次数递增时，总评分单调不增。"""
        scores = []
        for count in [0, 1, 2, 3, 5, 10, 50]:
            a = _asset_with_use_count(f"a{count}", count)
            s, _ = score_asset(a)
            scores.append(s)
        # 验证非递增
        for i in range(len(scores) - 1):
            assert (
                scores[i] >= scores[i + 1]
            ), f"use_count 递增时评分应不增: scores[{i}]={scores[i]} < scores[{i+1}]={scores[i+1]}"


# ── smart_select_assets 排序测试 ─────────────────────────────────────────────


class TestSmartSelectAssetsOrdering:
    """验证 smart_select_assets 返回结果按评分降序。"""

    def test_less_used_assets_ranked_higher(self):
        """使用次数少的素材在结果中排名更高（注入零噪声 rng 验证纯评分排序）。"""
        assets = [
            _asset_with_use_count("heavily_used", 10),
            _asset_with_use_count("never_used", 0),
            _asset_with_use_count("lightly_used", 2),
        ]
        results = smart_select_assets(assets, rng=_ZeroNoiseRandom())
        ids = [r.asset.id for r in results]
        # never_used 排第一，heavily_used 排最后
        assert ids[0] == "never_used"
        assert ids[-1] == "heavily_used"

    def test_same_quality_different_use_count(self):
        """质量相同时，使用次数少的排名更高（注入零噪声 rng）。"""
        assets = [
            _asset_with_use_count("used_5", 5),
            _asset_with_use_count("used_0", 0),
        ]
        results = smart_select_assets(assets, rng=_ZeroNoiseRandom())
        assert results[0].asset.id == "used_0"
        assert results[1].asset.id == "used_5"


# ── from-assets 路径集成测试 ─────────────────────────────────────────────────


def _make_mock_asset_for_clips(aid, duration, use_count=0):
    """创建带 score_asset 所需属性的 mock 素材。"""
    asset = MagicMock()
    asset.id = aid
    asset.duration = duration
    asset.quality_score = None
    asset.created_at = None
    asset.metadata = {"generation_use_count": use_count}
    return asset


def _make_auth_user():
    auth = MagicMock()
    auth.user.id = "user-001"
    auth.user.email = "test@example.com"
    auth.user.display_name = "测试用户"
    auth.user_id = "user-001"
    return auth


class _ZeroNoiseCtx:
    """同时 patch module.random.uniform（排序噪声归零）与 module.random.shuffle
    （排序后的素材洗牌保持原序），使「按评分排序」类断言确定可复现。

    生产环境排序噪声与 shuffle 都是降重随机的一部分；测试需要验证纯评分
    排序时用本上下文消除随机性。其他 uniform 调用（片段时长随机等）委托给
    独立 Random 实例，避免递归回已 patch 的全局函数。
    """

    def __init__(self, module):
        import random as _stdlib_random

        self._module = module
        self._fallback = _stdlib_random.Random()
        self._patches = []

    def __enter__(self):
        import random as _stdlib_random

        def _fake_uniform(a, b):
            if b == SCORE_RANDOM_NOISE_MAX:
                return 0.0
            return self._fallback.uniform(a, b)

        def _no_shuffle(seq):
            return None  # 保持原序，不洗牌

        self._patches = [
            patch.object(self._module.random, "uniform", _fake_uniform).start(),
            patch.object(self._module.random, "shuffle", _no_shuffle).start(),
        ]
        return self

    def __exit__(self, *exc):
        patch.stopall()
        return False


def _make_zero_noise_patcher(module):
    """构造零噪声上下文（uniform 噪声归零 + shuffle 保持原序）。"""
    return _ZeroNoiseCtx(module)


def _patch_zero_noise_clips():
    """消除 clips.py 排序噪声，其他 uniform 调用不受影响。"""
    from app.api.routes.templates_editor import clips as clips_module

    return _make_zero_noise_patcher(clips_module)


def _patch_zero_noise_plan_service():
    """消除 plan_generator_service.py 排序噪声，其他 uniform 调用不受影响。"""
    from app.services import plan_generator_service as svc_module

    return _make_zero_noise_patcher(svc_module)


class TestFromAssetsSmartMatchIntegration:
    """验证 clips.py 中 sorted_candidates 使用 smart_match 评分。"""

    def test_sorted_candidates_prefers_high_score_low_use_count(self):
        """在 from-assets 路径中，smart_match 分高且使用次数少的素材排在前面。"""
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        mock_asset_repo = MagicMock()

        def _get_asset(aid):
            use_count = {"a_heavy": 10, "a_fresh": 0}[aid]
            return _make_mock_asset_for_clips(aid, 30.0, use_count)

        mock_asset_repo.get = MagicMock(side_effect=_get_asset)

        mock_plan_svc = MagicMock()
        mock_plan_svc.replace_all_clips_transactional = MagicMock(return_value=2)

        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0)]

        with (
            _patch_zero_noise_clips(),
            patch(
                "app.api.routes.templates_editor.clips._get_template_segments",
                return_value=segments,
            ),
            patch(
                "app.api.routes.templates_editor.clips.get_used_segments",
                return_value={},
            ),
            patch(
                "app.api.routes.templates_editor.clips.record_used_segments",
                return_value=None,
            ),
        ):
            body = ClipsFromAssetsRequest(
                asset_ids=["a_heavy", "a_fresh"],
                required_clips_count=2,
            )
            create_clips_from_assets_editor(
                template_id="tmpl-1",
                body=body,
                background_tasks=MagicMock(),
                plan_id="test-plan-001",
                services=(MagicMock(), mock_plan_svc),
                asset_repo=mock_asset_repo,
                db=MagicMock(),
                current_user=_make_auth_user(),
            )

        # 验证 replace_all_clips_transactional 被调用
        assert mock_plan_svc.replace_all_clips_transactional.called
        call_args = mock_plan_svc.replace_all_clips_transactional.call_args
        clips_data = call_args.args[1]

        # 第一个片段应该分配给 a_fresh（smart_match 分更高）
        first_clip_asset = clips_data[0]["asset_id"]
        assert (
            first_clip_asset == "a_fresh"
        ), f"第一个片段应分配给 smart_match 分更高的 a_fresh，实际是 {first_clip_asset}"

    def test_score_noise_causes_varied_selection(self):
        """得分接近（差距 < SCORE_RANDOM_NOISE_MAX）的素材，多次生成的素材组合应有变化。

        两条同等质量/时长/新鲜度的素材（use_count 相同），smart_match 得分一致，
        噪声让两者的相对排名随机浮动，多次调用首个片段的素材分布应两者都出现。
        """
        from app.api.routes.templates_editor.clips import create_clips_from_assets_editor
        from app.api.routes.templates_editor.schemas import ClipsFromAssetsRequest

        def _get_asset(aid):
            return _make_mock_asset_for_clips(aid, 30.0, 0)

        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(side_effect=_get_asset)

        segments = [(0, 3.0, 5.0), (1, 3.0, 5.0)]

        first_assets: set[str] = set()
        for _ in range(30):
            mock_plan_svc = MagicMock()
            mock_plan_svc.replace_all_clips_transactional = MagicMock(return_value=2)
            with (
                patch(
                    "app.api.routes.templates_editor.clips._get_template_segments",
                    return_value=segments,
                ),
                patch(
                    "app.api.routes.templates_editor.clips.get_used_segments",
                    return_value={},
                ),
                patch(
                    "app.api.routes.templates_editor.clips.record_used_segments",
                    return_value=None,
                ),
            ):
                body = ClipsFromAssetsRequest(
                    asset_ids=["a_x", "a_y"],
                    required_clips_count=2,
                )
                create_clips_from_assets_editor(
                    template_id="tmpl-1",
                    body=body,
                    background_tasks=MagicMock(),
                    plan_id=f"plan-noise-{len(first_assets)}-{_}",
                    services=(MagicMock(), mock_plan_svc),
                    asset_repo=mock_asset_repo,
                    db=MagicMock(),
                    current_user=_make_auth_user(),
                )
            clips_data = mock_plan_svc.replace_all_clips_transactional.call_args.args[1]
            first_assets.add(clips_data[0]["asset_id"])

        assert first_assets == {
            "a_x",
            "a_y",
        }, f"噪声应使两条等分素材的排名浮动，30 次调用首个片段应覆盖两者，实际 {first_assets}"


# ── 一键生成路径集成测试 ─────────────────────────────────────────────────────


class TestPlanGeneratorSmartMatchIntegration:
    """验证 PlanGeneratorService._sort_assets_by_smart_score 排序正确。"""

    def test_sort_assets_by_smart_score_descending(self):
        """_sort_assets_by_smart_score 返回按评分降序排列的素材 ID。"""
        from app.services.plan_generator_service import PlanGeneratorService

        mock_asset_repo = MagicMock()

        def _get_asset(aid):
            use_count = {"high_use": 10, "low_use": 0, "mid_use": 3}[aid]
            asset = MagicMock()
            asset.id = aid
            asset.duration = 15.0
            asset.quality_score = None
            asset.created_at = None
            asset.metadata = {"generation_use_count": use_count}
            return asset

        mock_asset_repo.get = MagicMock(side_effect=_get_asset)

        db = MagicMock()
        svc = PlanGeneratorService(db, asset_repo=mock_asset_repo)

        with _patch_zero_noise_plan_service():
            sorted_ids = svc._sort_assets_by_smart_score(["high_use", "low_use", "mid_use"])

        # low_use (0次) 应排第一，high_use (10次) 应排最后
        assert sorted_ids[0] == "low_use"
        assert sorted_ids[-1] == "high_use"
        assert sorted_ids[1] == "mid_use"

    def test_distribute_assets_uses_smart_score_ordering(self):
        """_distribute_assets 在非随机模式下按 smart_match 评分排序素材。"""
        from app.services.plan_generator_service import PlanGeneratorService

        from packages.domain.edit_plan_clip import EditPlanClip
        from packages.domain.editing_mode import EditingMode

        mock_asset_repo = MagicMock()

        def _get_asset(aid):
            use_count = {"old_asset": 10, "new_asset": 0}[aid]
            asset = MagicMock()
            asset.id = aid
            asset.duration = 30.0
            asset.quality_score = None
            asset.created_at = None
            asset.metadata = {"generation_use_count": use_count}
            return asset

        mock_asset_repo.get = MagicMock(side_effect=_get_asset)

        db = MagicMock()
        svc = PlanGeneratorService(db, asset_repo=mock_asset_repo)

        # 创建 2 个 main clips（需要提供 id 参数）
        clips = [
            EditPlanClip(id="c1", plan_id="p1", clip_type="main", duration=5.0, order=0),
            EditPlanClip(id="c2", plan_id="p1", clip_type="main", duration=5.0, order=1),
        ]

        with (
            _patch_zero_noise_plan_service(),
            patch("app.services.plan_generator_service.distribute_assets") as mock_dist,
        ):
            svc._distribute_assets(
                clips,
                ["old_asset", "new_asset"],
                EditingMode.ONE_TAKE.value,
                random_selection=False,
            )
            # 验证传给 distribute_assets 的 asset_ids 按 smart_match 排序
            call_args = mock_dist.call_args
            passed_ids = call_args.args[1]
            # new_asset (0次使用) 应排在 old_asset (10次使用) 前面
            assert passed_ids[0] == "new_asset"
            assert passed_ids[1] == "old_asset"

    def test_random_selection_skips_smart_score_sort(self):
        """random_selection=True 时不执行 smart_match 排序。"""
        from app.services.plan_generator_service import PlanGeneratorService

        from packages.domain.edit_plan_clip import EditPlanClip
        from packages.domain.editing_mode import EditingMode

        mock_asset_repo = MagicMock()
        db = MagicMock()
        svc = PlanGeneratorService(db, asset_repo=mock_asset_repo)

        clips = [
            EditPlanClip(id="c1", plan_id="p1", clip_type="main", duration=5.0, order=0),
        ]

        with patch("app.services.plan_generator_service.distribute_assets") as mock_dist:
            svc._distribute_assets(
                clips,
                ["a1", "a2"],
                EditingMode.ONE_TAKE.value,
                random_selection=True,
            )
            # random_selection=True 时不应调用 asset_repo.get（不执行排序）
            mock_asset_repo.get.assert_not_called()


class TestPlanGeneratorScoreNoise:
    """验证一键生成路径的评分排序注入了随机噪声。"""

    def test_equal_scores_produce_varied_order(self):
        """两条 smart_match 得分相同的素材，多次排序的首位应覆盖两者。"""
        from app.services.plan_generator_service import PlanGeneratorService

        def _get_asset(aid):
            asset = MagicMock()
            asset.id = aid
            asset.duration = 15.0
            asset.quality_score = None
            asset.created_at = None
            asset.metadata = {"generation_use_count": 0}
            return asset

        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(side_effect=_get_asset)
        svc = PlanGeneratorService(MagicMock(), asset_repo=mock_asset_repo)

        first_ids: set[str] = set()
        for _ in range(30):
            order = svc._sort_assets_by_smart_score(["equal_a", "equal_b"])
            first_ids.add(order[0])

        assert first_ids == {
            "equal_a",
            "equal_b",
        }, f"噪声应使等分素材排名浮动，30 次排序首位应覆盖两者，实际 {first_ids}"

    def test_large_score_gap_not_flipped(self):
        """得分差距远大于噪声上限时，低分素材不会因噪声超过高分素材。

        quality 100 vs 0 → quality 维度差距 40 分 > 噪声上限 20，
        其余维度完全一致，50 次排序高质量素材必须始终排第一。
        """
        from app.services.plan_generator_service import PlanGeneratorService

        def _get_asset(aid):
            quality = {"top": 100.0, "bad": 0.0}[aid]
            asset = MagicMock()
            asset.id = aid
            asset.duration = 15.0
            asset.quality_score = quality
            asset.created_at = None
            asset.metadata = {"generation_use_count": 0}
            return asset

        mock_asset_repo = MagicMock()
        mock_asset_repo.get = MagicMock(side_effect=_get_asset)
        svc = PlanGeneratorService(MagicMock(), asset_repo=mock_asset_repo)

        for _ in range(50):
            order = svc._sort_assets_by_smart_score(["top", "bad"])
            assert order[0] == "top", f"质量差距 40 分 > 噪声上限，top 应始终排第一，实际 {order}"