content-find-agent-new/tests/test_portrait_cache.py

"""M9 修复:作者画像按作者缓存(画像是作者级,同 run 内同作者只拉一次)。"""

from content_agent.business_modules.progressive_screening import _ProgressiveContext


class CountingPortraitClient:
    def __init__(self):
        self.calls = []

    def fetch_account_fans_portrait(self, account_id):
        self.calls.append(account_id)
        return {
            "fans": {
                "age": {
                    "data": {
                        "41-50": {"percentage": "31.73%", "preference": "210"},
                        "50-": {"percentage": "29.76%", "preference": "130"},
                    }
                }
            }
        }


def _ctx(client):
    return _ProgressiveContext(
        run_id="r", policy_run_id="p", source_context={}, policy_bundle={},
        platform_client=client, runtime=None, gemini_video_client=None,
        limiter=None, archive_dispatcher=None, platform="douyin", portrait_client=client,
    )


def _bundle(author, q=80, p=70):
    return {
        "pattern_match_result": {"query_relevance_score": q},
        "content_engagement_metrics": {"platform_performance": {"platform_performance_score": p}},
        "content": {"author": {"platform_author_id": author}},
    }


def test_portrait_cached_by_author():
    client = CountingPortraitClient()
    ctx = _ctx(client)
    bundles = [_bundle("a1"), _bundle("a1"), _bundle("a2")]
    ctx._inject_fifty_plus(bundles)
    assert client.calls == ["a1", "a2"]  # 同作者只拉一次
    assert all(b["content_audience_50plus"]["status"] == "ok" for b in bundles)


def test_portrait_fetched_when_query_passes_regardless_of_platform():
    # 修复:门只看 query 相关性,平台分再低也拉画像、算适老性(原来 platform<65 会跳过)。
    client = CountingPortraitClient()
    ctx = _ctx(client)
    bundles = [_bundle("a1", q=70, p=20)]  # 平台仅 20
    ctx._inject_fifty_plus(bundles)
    assert client.calls == ["a1"]
    assert bundles[0]["content_audience_50plus"]["status"] == "ok"


def test_portrait_skipped_when_query_below_gate():
    client = CountingPortraitClient()
    ctx = _ctx(client)
    bundles = [_bundle("a1", q=60, p=90)]  # 相关性 60 < 65 → 不拉
    ctx._inject_fifty_plus(bundles)
    assert client.calls == []
    assert bundles[0]["content_audience_50plus"]["status"] == "not_attempted"


class FlakyPortraitClient:
    def __init__(self, fail_times):
        self.fail_times = fail_times
        self.calls = 0

    def fetch_account_fans_portrait(self, account_id):
        self.calls += 1
        if self.calls <= self.fail_times:
            raise RuntimeError("热点宝 429")
        return {"fans": {"age": {"data": {"50-": {"percentage": "40%", "preference": "150"}}}}}


def test_portrait_retry_recovers_transient_failure():
    client = FlakyPortraitClient(fail_times=1)  # 第一次失败、第二次成功
    ctx = _ctx(client)
    bundles = [_bundle("a1")]
    ctx._inject_fifty_plus(bundles)
    assert client.calls == 2  # 重试了一次
    assert bundles[0]["content_audience_50plus"]["status"] == "ok"


def test_portrait_unavailable_records_failure_reason():
    client = FlakyPortraitClient(fail_times=9)  # 一直失败
    ctx = _ctx(client)
    bundles = [_bundle("a1")]
    ctx._inject_fifty_plus(bundles)
    block = bundles[0]["content_audience_50plus"]
    assert block["status"] == "unavailable"
    assert "热点宝 429" in block["failure_reason"]  # 原因被记下,不再吞掉