新功能：个性化推荐算法

server/tests/test_rerank_freqcap.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+import pytest
+
+from app.features.personalized_reco.content_repository.types import ContentProfileDTO
+from app.features.personalized_reco.rerank_freqcap.rerank import rerank_and_freqcap
+from app.features.personalized_reco.rerank_freqcap.types import ScoredCandidate
+
+
+def _c(
+    *,
+    content_id: int,
+    score: float,
+    author_id: str | None = None,
+    template_id: str | None = None,
+    stage: str = "general",
+    need_key: str = "emotional_support",
+    context_key: str = "family",
+) -> ScoredCandidate:
+    cp = ContentProfileDTO(
+        content_id=content_id,
+        text="t",
+        stage=stage,  # type: ignore[arg-type]
+        emotion_score=None,
+        context_suitability={context_key: 1.0},
+        need_suitability={need_key: 1.0},
+        personalization_power=0.0,
+        risk_flags=[],
+        author_id=author_id,
+        template_id=template_id,
+    )
+    return ScoredCandidate(
+        content_id=content_id,
+        final_score=score,
+        author_id=author_id,
+        template_id=template_id,
+        content_profile=cp,
+    )
+
+
+def test_dedup_normalizes_str_int_ids() -> None:
+    cands = [_c(content_id=1, score=0.9), _c(content_id=2, score=0.8), _c(content_id=3, score=0.7)]
+    r = rerank_and_freqcap(
+        scene="push",
+        scored_candidates=cands,
+        already_recommended_ids=["1", "bad"],
+        touched_or_viewed_ids=[2],
+        k=10,
+    )
+    got_ids = [x.content_id for x in r.ranked_items]
+    assert got_ids == [3]
+    assert r.meta.candidate_pool_size_after_dedup == 1
+    assert r.meta.freqcap_filtered_counts["sentence"] == 2
+
+
+def test_freqcap_missing_recent_author_template_is_recorded() -> None:
+    cands = [
+        _c(content_id=1, score=0.9, author_id="a1", template_id="t1"),
+        _c(content_id=2, score=0.8, author_id="a2", template_id="t2"),
+    ]
+    r = rerank_and_freqcap(
+        scene="push",
+        scored_candidates=cands,
+        already_recommended_ids=[],
+        touched_or_viewed_ids=[],
+        k=10,
+        recent_author_ids=None,
+        recent_template_ids=None,
+    )
+    assert r.meta.missing_history_fields == ["author", "template"]
+    assert "author" not in r.meta.freqcap_filtered_counts  # 未提供则不执行该维度
+    assert "template" not in r.meta.freqcap_filtered_counts
+
+
+def test_freqcap_filters_by_recent_author_when_provided() -> None:
+    cands = [
+        _c(content_id=1, score=0.9, author_id="a1", template_id="t1"),
+        _c(content_id=2, score=0.8, author_id="a2", template_id="t2"),
+    ]
+    r = rerank_and_freqcap(
+        scene="widget",
+        scored_candidates=cands,
+        already_recommended_ids=[],
+        touched_or_viewed_ids=[],
+        k=10,
+        recent_author_ids=["a1"],
+        recent_template_ids=None,
+    )
+    got_ids = [x.content_id for x in r.ranked_items]
+    assert got_ids == [2]
+    assert r.meta.missing_history_fields == ["template"]
+    assert r.meta.freqcap_filtered_counts["author"] == 1
+
+
+def test_feed_mmr_picks_diverse_second_item() -> None:
+    # 构造：Top1 是 a1；c2 分数略高但同作者；c3 分数略低但不同作者/阶段
+    c1 = _c(content_id=1, score=1.0, author_id="a1", template_id="t1", stage="general")
+    c2 = _c(content_id=2, score=0.99, author_id="a1", template_id="t2", stage="general")
+    c3 = _c(content_id=3, score=0.95, author_id="a2", template_id="t3", stage="expecting")
+
+    r = rerank_and_freqcap(
+        scene="feed",
+        scored_candidates=[c1, c2, c3],
+        already_recommended_ids=[],
+        touched_or_viewed_ids=[],
+        k=2,
+    )
+    got_ids = [x.content_id for x in r.ranked_items]
+    assert got_ids[0] == 1
+    assert got_ids[1] == 3
+
+
+def test_push_topk_sorted_by_score_after_filters() -> None:
+    cands = [
+        _c(content_id=1, score=0.1),
+        _c(content_id=2, score=0.9),
+        _c(content_id=3, score=0.8),
+    ]
+    r = rerank_and_freqcap(
+        scene="push",
+        scored_candidates=cands,
+        already_recommended_ids=[],
+        touched_or_viewed_ids=[],
+        k=2,
+        recent_author_ids=[],
+        recent_template_ids=[],
+    )
+    got_ids = [x.content_id for x in r.ranked_items]
+    assert got_ids == [2, 3]
+