131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
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]
|
||
|