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

"""M9E:抖音 50+ 进评分——35/35/30、not_attempted 走 35/35、画像缺失→TECHNICAL_RETRY+停游走、非抖音仍 50/50。"""

from copy import deepcopy
from pathlib import Path

from content_agent.business_modules import walk_strategy
from content_agent.business_modules.rule_judgment import evaluator
from content_agent.integrations.policy_json import JsonPolicyBundleStore

POLICY = JsonPolicyBundleStore(Path(".")).load_policy_bundle("V4")


def _bundle(query, platform_score, fifty=None):
    bundle = {
        "schema_version": "evidence_bundle.v1",
        "source_evidence": {"source_kind": "pg_pattern_v2"},
        "content": {
            "decision_target_type": "content",
            "decision_target_id": "c1",
            "content_discovery_id": "cd1",
            "search_query_id": "q_001",
            "platform": "douyin",
            "platform_content_id": "c1",
            "platform_content_format": "video",
            "description": "x",
            "author": {"platform_author_id": "a1"},
            "platform_author_id": "a1",
        },
        "pattern_match_result": {
            "query_relevance_score": query,
            "judge_status": "ok",
            "final_status": "ok",
        },
        "content_engagement_metrics": {
            "platform_performance": {
                "platform_performance_score": platform_score,
                "platform_performance_components": [],
                "missing_observable_fields": [],
            }
        },
        "content_risk_check": {"risk_level": "low", "age_50_plus_safety": "safe"},
        "run_context": {
            "run_id": "run",
            "policy_run_id": "policy_run",
            "runtime_record_schema_version": "runtime_record.v1",
            "decision_input_snapshot_id": "evidence:c1",
        },
    }
    if fifty is not None:
        bundle["content_audience_50plus"] = fifty
    return bundle


def _decide(bundle):
    return evaluator.decide("run", "policy_run", 1, bundle, POLICY)


def test_douyin_ok_uses_35_35_30():
    fifty = {"status": "ok", "score": 76, "components": {"band": "50+", "band_tgi": 171.5}}
    decision = _decide(_bundle(80, 70, fifty))
    scorecard = decision["scorecard"]
    assert scorecard["fifty_plus_status"] == "ok"
    assert scorecard["fifty_plus_score"] == 76
    assert scorecard["score_weights"] == {
        "query_relevance": 0.35,
        "platform_performance": 0.35,
        "fifty_plus": 0.3,
    }
    assert abs(decision["score"] - (0.35 * 80 + 0.35 * 70 + 0.30 * 76)) < 0.01  # 75.3
    assert decision["decision_action"] == "ADD_TO_CONTENT_POOL"


def test_douyin_not_attempted_uses_35_35_two_terms():
    decision = _decide(_bundle(80, 70, {"status": "not_attempted"}))
    assert decision["scorecard"]["fifty_plus_status"] == "not_attempted"
    assert "fifty_plus_score" not in decision["scorecard"]
    assert decision["scorecard"]["score_weights"] == {
        "query_relevance": 0.35,
        "platform_performance": 0.35,
    }
    assert abs(decision["score"] - (0.35 * 80 + 0.35 * 70)) < 0.01  # 52.5


def test_douyin_portrait_unavailable_is_technical_retry_and_stops_walk():
    decision = _decide(_bundle(80, 70, {"status": "unavailable"}))
    assert decision["score"] is None
    assert decision["decision_action"] == "TECHNICAL_RETRY_REQUIRED"
    assert decision["decision_reason_code"] == "portrait_unavailable"
    # 停游走:TECHNICAL_RETRY_REQUIRED 映射 path_stop。
    assert walk_strategy._action_for_decision("TECHNICAL_RETRY_REQUIRED")["edge_id"] == "path_stop"


def test_douyin_portrait_incomplete_reason():
    decision = _decide(_bundle(80, 70, {"status": "incomplete"}))
    assert decision["decision_action"] == "TECHNICAL_RETRY_REQUIRED"
    assert decision["decision_reason_code"] == "portrait_incomplete"


def test_non_douyin_keeps_50_50_when_no_block():
    decision = _decide(_bundle(80, 70))  # 无 content_audience_50plus 块
    scorecard = decision["scorecard"]
    assert "fifty_plus_status" not in scorecard
    assert "fifty_plus_score" not in scorecard
    assert scorecard["score_weights"] == {
        "query_relevance": 0.5,
        "platform_performance": 0.5,
    }
    assert abs(decision["score"] - (0.5 * 80 + 0.5 * 70)) < 0.01  # 75.0


def test_douyin_ok_uses_policy_score_weight_profile():
    policy = deepcopy(POLICY)
    profiles = policy["rule_pack"]["scorecard"]["score_weight_profiles"]["profiles"]
    ok_profile = next(profile for profile in profiles if profile["profile_id"] == "audience_50plus_available_v1")
    ok_profile["weights"] = {"query_relevance": 40, "platform_performance": 30, "fifty_plus": 30}

    decision = evaluator.decide(
        "run",
        "policy_run",
        1,
        _bundle(80, 70, {"status": "ok", "score": 76}),
        policy,
    )

    assert abs(decision["score"] - (0.40 * 80 + 0.30 * 70 + 0.30 * 76)) < 0.01
    assert decision["scorecard"]["score_weights"] == {
        "query_relevance": 0.4,
        "platform_performance": 0.3,
        "fifty_plus": 0.3,
    }