re-xxm/tests/test_analytics.py

# -*- coding: utf-8 -*-
"""看板指标纯函数测试（不联网）: uv run python tests/test_analytics.py"""

import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.stdout.reconfigure(encoding="utf-8")

from reaction_board.analytics import (  # noqa: E402
    DEFAULT_HEAT_WEIGHTS,
    METRICS,
    compute_growth,
    heat_scores,
    interaction_rate,
    is_loopback,
    normalize_cover,
    normalized_weights,
    percentile,
    robust_center_scale,
    score_percentiles,
)


def row(aid, **metrics):
    base = {metric: 0 for metric in METRICS}
    base.update(metrics)
    base["aid"] = aid
    return base


def test_percentile_and_robust_scale():
    assert percentile([], 0.5) == 0.0
    assert percentile([7], 0.25) == 7.0
    assert percentile([1, 2, 3, 4], 0.5) == 2.5
    assert percentile([1, 2, 3, 4], 0) == 1.0
    assert percentile([1, 2, 3, 4], 1) == 4.0

    center, scale = robust_center_scale([1, 2, 3, 4, 5])
    assert center == 3.0
    assert scale > 0

    # IQR 为 0 时退化为总体标准差
    center, scale = robust_center_scale([10, 10, 10, 20])
    assert center == 10.0
    assert scale > 0

    # 全相同则 scale=0，调用方把 z 视为 0
    assert robust_center_scale([5, 5, 5]) == (5.0, 0.0)
    assert robust_center_scale([]) == (0.0, 0.0)


def test_normalized_weights():
    weights = normalized_weights()
    assert abs(sum(weights.values()) - 1.0) < 1e-12
    assert abs(weights["view"] - DEFAULT_HEAT_WEIGHTS["view"]) < 1e-12

    doubled = normalized_weights({metric: value * 2 for metric, value in DEFAULT_HEAT_WEIGHTS.items()})
    assert doubled == weights

    zero = normalized_weights({metric: 0 for metric in METRICS})
    assert all(value == 0 for value in zero.values())


def test_heat_scores_order_and_empty():
    rows = [
        row(1, view=100, like=10, coin=5, favorite=4, share=1, reply=1, danmaku=2),
        row(2, view=1000, like=100, coin=50, favorite=40, share=10, reply=10, danmaku=20),
        row(3, view=10_000, like=1000, coin=500, favorite=400, share=100, reply=100, danmaku=200),
    ]
    scores = heat_scores(rows)
    assert set(scores) == {1, 2, 3}
    assert scores[1] < scores[2] < scores[3]

    assert heat_scores([]) == {}

    # 所有指标完全相同：所有热度为 0，而不是除零
    same = heat_scores([row(1, view=5), row(2, view=5)])
    assert same == {1: 0.0, 2: 0.0}


def test_heat_scores_weights_and_missing_values():
    rows = [
        row(1, view=100),
        row(2, view=200),
    ]
    view_only = heat_scores(rows, {"view": 1.0})
    assert view_only[2] > view_only[1]
    assert all(isinstance(value, float) for value in view_only.values())

    # 缺失/非法指标按 0 处理，不抛异常
    rows = [
        {"aid": 1, "view": None, "like": "bad"},
        {"aid": 2, "view": 10},
    ]
    scores = heat_scores(rows)
    assert scores[2] > scores[1]


def test_score_percentiles():
    assert score_percentiles({}) == {}
    assert score_percentiles({5: 1.0}) == {5: 100.0}
    result = score_percentiles({1: 0.0, 2: 1.0, 3: 2.0})
    assert result[1] == 0.0
    assert result[2] == 50.0
    assert result[3] == 100.0


def test_interaction_rate_thresholds():
    assert interaction_rate({"view": 99, "like": 1, "coin": 1, "favorite": 1}) is None
    assert interaction_rate({"view": 100, "like": 1, "coin": 1, "favorite": 2}) == 0.04
    assert interaction_rate({"view": 1000, "like": 10}) == 0.01
    assert interaction_rate({"view": 0, "like": 10}) is None
    assert interaction_rate({"view": None}) is None
    assert interaction_rate({"view": 100, "like": None, "coin": "x"}) == 0.0


def test_compute_growth():
    unavailable = compute_growth(100, None, 7)
    assert unavailable.available is False
    assert unavailable.delta is None

    no_gap = compute_growth(100, 90, 0)
    assert no_gap.available is False

    growth = compute_growth(1000, 800, 7)
    assert growth.available is True
    assert growth.delta == 200
    assert growth.per_day == 200 / 7
    assert abs(growth.rate - 0.25) < 1e-12
    assert growth.gap_days == 7

    # 基线为 0 时用 1 做分母，避免除零
    from_zero = compute_growth(50, 0, 3)
    assert from_zero.delta == 50
    assert from_zero.rate == 50.0

    # 累计指标倒退（接口修正）时保留负值，由调用方决定是否展示
    negative = compute_growth(90, 100, 2)
    assert negative.delta == -10
    assert negative.per_day == -5.0


def test_normalize_cover_and_hosts():
    assert normalize_cover(None) is None
    assert normalize_cover("") is None
    assert normalize_cover("  ") is None
    assert normalize_cover(123) is None
    assert normalize_cover("//i0.hdslb.com/a.jpg") == "https://i0.hdslb.com/a.jpg"
    assert normalize_cover("http://i0.hdslb.com/a.jpg") == "https://i0.hdslb.com/a.jpg"
    assert normalize_cover("https://i0.hdslb.com/a.jpg") == "https://i0.hdslb.com/a.jpg"

    assert is_loopback("127.0.0.1")
    assert is_loopback("localhost")
    assert is_loopback("::1")
    assert not is_loopback("0.0.0.0")
    assert not is_loopback("192.168.1.2")


def main_test():
    test_percentile_and_robust_scale()
    test_normalized_weights()
    test_heat_scores_order_and_empty()
    test_heat_scores_weights_and_missing_values()
    test_score_percentiles()
    test_interaction_rate_thresholds()
    test_compute_growth()
    test_normalize_cover_and_hosts()
    print("OK: analytics 热度/增长/互动率/封面测试全部通过")


if __name__ == "__main__":
    main_test()