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

from __future__ import annotations

from copy import deepcopy

from content_agent.business_modules.rule_judgment.evaluator import decide


def test_v4_scorecard_uses_query_and_platform_50_50():
    decision = decide("run_1", "policy_1", 1, _bundle(query=80, platform=70), _policy())

    assert decision["score"] == 75
    assert decision["decision_action"] == "ADD_TO_CONTENT_POOL"
    assert decision["decision_reason_code"] == "v4_query_and_platform_pass"
    assert decision["search_query_effect_status"] == "success"
    assert decision["scorecard"]["schema_version"] == "v4_scorecard.v1"
    assert decision["scorecard"]["query_relevance_score"] == 80
    assert decision["scorecard"]["platform_performance_score"] == 70
    assert decision["decision_replay_data"]["allow_walk"] is True
    assert decision["decision_replay_data"]["walk_gate_snapshot"] == {
        "query_relevance_score": 80,
        "platform_performance_score": 70,
        "score": 75,
    }
    assert decision["scorecard"]["score_weights"] == {
        "query_relevance": 0.5,
        "platform_performance": 0.5,
    }


def test_v4_scorecard_uses_policy_score_weight_profile():
    policy = _policy()
    policy["rule_pack"]["scorecard"]["score_weight_profiles"] = {
        "default_two_dimension": {
            "query_relevance": 60,
            "platform_performance": 40,
        },
        "douyin_fifty_plus_ok": {
            "query_relevance": 35,
            "platform_performance": 35,
            "fifty_plus": 30,
        },
        "douyin_fifty_plus_not_attempted": {
            "query_relevance": 35,
            "platform_performance": 35,
        },
    }

    decision = decide("run_1", "policy_1", 1, _bundle(query=80, platform=70), policy)

    assert decision["score"] == 76
    assert decision["scorecard"]["score_weights"] == {
        "query_relevance": 0.6,
        "platform_performance": 0.4,
    }


def test_v4_threshold_boundaries():
    cases = [
        (54, 100, "REJECT_CONTENT", "v4_query_or_score_below_threshold"),
        (55, 55, "KEEP_CONTENT_FOR_REVIEW", "v4_score_review_needed"),
        (69, 69, "KEEP_CONTENT_FOR_REVIEW", "v4_score_review_needed"),
        (70, 70, "ADD_TO_CONTENT_POOL", "v4_query_and_platform_pass"),
    ]
    for query, platform, action, reason in cases:
        decision = decide("run_1", "policy_1", 1, _bundle(query=query, platform=platform), _policy())
        assert decision["decision_action"] == action
        assert decision["decision_reason_code"] == reason


def test_v4_allow_walk_requires_platform_65():
    decision = decide("run_1", "policy_1", 1, _bundle(query=80, platform=64), _policy())

    assert decision["decision_action"] == "ADD_TO_CONTENT_POOL"
    assert decision["score"] == 72
    assert decision["decision_replay_data"]["allow_walk"] is False


def test_v4_missing_score_routes_to_technical_retry():
    decision = decide("run_1", "policy_1", 1, _bundle(query=None, platform=70), _policy())

    assert decision["score"] is None
    assert decision["scorecard"]["score_missing"] is True
    assert decision["decision_action"] == "TECHNICAL_RETRY_REQUIRED"
    assert decision["decision_reason_code"] == "v4_technical_retry_needed"
    assert decision["search_query_effect_status"] == "failed"
    assert decision["decision_replay_data"]["allow_walk"] is False


def test_v4_scorecard_and_replay_data_do_not_contain_legacy_fields():
    decision = decide("run_1", "policy_1", 1, _bundle(query=80, platform=70), _policy())
    keys = _keys({"scorecard": decision["scorecard"], "replay": decision["decision_replay_data"]})

    assert not (keys & {"fit_senior_50plus", "fit_confidence", "platform_heat", "relevance_score"})


def _bundle(query, platform):
    return {
        "source_evidence": {"source_kind": "pattern_itemset"},
        "content": {
            "decision_target_type": "content",
            "decision_target_id": "content_1",
            "platform_content_id": "content_1",
        },
        "pattern_match_result": {
            "query_relevance_score": query,
            "judge_status": "ok",
        },
        "content_engagement_metrics": {
            "platform_performance": {
                "schema_version": "v4_platform_performance.v1",
                "platform": "douyin",
                "platform_performance_score": platform,
                "platform_performance_components": [
                    {
                        "field": "statistics.digg_count",
                        "value": 100,
                        "weight": 1,
                        "normalized_score": platform or 0,
                    }
                ],
                "missing_observable_fields": [],
            }
        },
        "run_context": {"decision_input_snapshot_id": "evidence_bundle:run_1:content_1"},
    }


def _policy():
    return deepcopy(
        {
            "policy_bundle_id": "policy_bundle_v4",
            "rule_pack_id": "douyin_content_discovery_rule_pack_v1",
            "rule_pack_version": "4.0.0",
            "strategy_id": "douyin_content_find_v4",
            "strategy_version": "V4",
            "policy_bundle_hash": "hash",
            "dispatch_id": "dispatch_content",
            "runtime_stage": "V1.0",
            "rule_package_id": "douyin_rule_packs_v1",
            "rule_pack": {
                "input_contract": {
                    "required_fields": [
                        "source_evidence",
                        "pattern_match_result.query_relevance_score",
                        "content_engagement_metrics.platform_performance.platform_performance_score",
                    ]
                },
                "hard_gates": [],
                "scorecard": {"schema_version": "v4_scorecard.v1"},
            },
            "effect_status_mapping": [
                {
                    "mapping_id": "map_add_to_pool_success",
                    "target_level": "content",
                    "decision_action": "ADD_TO_CONTENT_POOL",
                    "reason_category": "score_pass",
                    "is_hard_gate": False,
                    "content_effect_status": "success",
                    "priority": 10,
                    "enabled": True,
                },
                {
                    "mapping_id": "map_keep_for_review_pending",
                    "target_level": "content",
                    "decision_action": "KEEP_CONTENT_FOR_REVIEW",
                    "reason_category": "review_needed",
                    "is_hard_gate": False,
                    "content_effect_status": "pending",
                    "priority": 20,
                    "enabled": True,
                },
                {
                    "mapping_id": "map_reject_failed",
                    "target_level": "content",
                    "decision_action": "REJECT_CONTENT",
                    "reason_category": "score_or_data_failed",
                    "is_hard_gate": False,
                    "content_effect_status": "failed",
                    "priority": 40,
                    "enabled": True,
                },
                {
                    "mapping_id": "map_technical_retry_failed",
                    "target_level": "content",
                    "decision_action": "TECHNICAL_RETRY_REQUIRED",
                    "reason_category": "technical_error",
                    "is_hard_gate": False,
                    "content_effect_status": "failed",
                    "priority": 30,
                    "enabled": True,
                },
            ],
        }
    )


def _keys(value):
    if isinstance(value, dict):
        result = set(value)
        for child in value.values():
            result |= _keys(child)
        return result
    if isinstance(value, list):
        result = set()
        for child in value:
            result |= _keys(child)
        return result
    return set()