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

from __future__ import annotations

from content_agent.business_modules.content_discovery.platform_observable_performance import (
    performance_score,
)


# M11:平台表现改"量级分 + 收缩后比例分"(配置驱动 observable_performance)。
# 三平台按真实字段可得性走不同 component:抖音互动间比例 / 快手 per 播放真互动率 / 视频号仅点赞量级。


def test_douyin_v2_volume_plus_inter_metric_ratios():
    result = performance_score(
        {"digg_count": 100000, "comment_count": 1000, "share_count": 500, "collect_count": 1000},
        "douyin",
    )
    assert 0 <= result["platform_performance_score"] <= 100
    assert [row["field"] for row in result["platform_performance_components"]] == [
        "total_interaction",
        "share_ratio",
        "collect_ratio",
        "comment_ratio",
    ]
    assert result["platform_performance_components"][0]["type"] == "absolute"
    assert result["platform_performance_components"][1]["type"] == "ratio"
    # play_count 抖音天然缺失仍如实上报
    assert {
        "field": "statistics.play_count",
        "missing_type": "natural_platform_missing",
        "platform": "douyin",
        "evidence": "跨平台字段映射.json",
    } in result["missing_observable_fields"]
    assert "platform_heat" not in result


def test_kuaishou_v2_per_view_rates_no_share():
    result = performance_score(
        {"play_count": 10000, "digg_count": 2000, "comment_count": 200, "share_count": 100, "collect_count": 100},
        "kuaishou",
    )
    # 快手:播放量级 + 三个 per 播放真互动率;无转发项(share 恒 0)
    assert [row["field"] for row in result["platform_performance_components"]] == [
        "play_count",
        "like_rate",
        "collect_rate",
        "comment_rate",
    ]
    assert result["platform_performance_score"] is not None
    assert result["missing_observable_fields"] == []  # play 在 → 比例分母不缺


def test_kuaishou_missing_play_count_marks_ratio_denominator_missing():
    # 快手缺 play_count → per 播放比例分母为 0 → 记 runtime_missing、只剩播放量级一项
    result = performance_score({"digg_count": 2000, "comment_count": 200, "collect_count": 100}, "kuaishou")
    runtime_missing = [
        row for row in result["missing_observable_fields"] if row.get("missing_type") == "runtime_missing"
    ]
    assert {row["field"] for row in runtime_missing} == {"like_rate", "collect_rate", "comment_rate"}
    assert [row["field"] for row in result["platform_performance_components"]] == ["play_count"]


def test_shipinhao_only_digg_volume():
    result = performance_score({"digg_count": 500}, "shipinhao")
    assert [row["field"] for row in result["platform_performance_components"]] == ["digg_count"]
    assert result["platform_performance_components"][0]["type"] == "absolute"
    assert {row["field"] for row in result["missing_observable_fields"]} == {
        "statistics.comment_count",
        "statistics.share_count",
        "statistics.collect_count",
        "statistics.play_count",
    }


def test_ratio_shrinkage_pulls_low_sample_toward_prior():
    # 抖音转发率:小样本(3 赞 1 转=33%)被收缩拉回正常,不给虚高分;大样本真高比例保留。
    low = performance_score({"digg_count": 3, "share_count": 1, "comment_count": 0, "collect_count": 0}, "douyin")
    high = performance_score({"digg_count": 100000, "share_count": 35000, "comment_count": 5000, "collect_count": 50000}, "douyin")
    low_share = next(c for c in low["platform_performance_components"] if c["field"] == "share_ratio")
    high_share = next(c for c in high["platform_performance_components"] if c["field"] == "share_ratio")
    assert low_share["raw_ratio"] > 0.3  # 原始 33%
    assert low_share["shrunk_ratio"] < 0.15  # 收缩后被拉回(prior 0.12 附近)
    assert low_share["normalized_score"] < 60
    assert high_share["normalized_score"] == 100  # 真·高转发占满