content-find-agent-new/content_agent/business_modules/content_discovery/fifty_plus.py

"""M9A:抖音 50+ 受众子分(作者粉丝画像 → 0-100)。

口径(已拍板):band=50+(仅 50-=50 岁及以上,不含 41-50 中年);
TGI_score=clamp(bandTGI/2,0,100)(TGI 200 满分);percent_score=clamp(band占比,0,100);
50+子分=0.6·TGI_score+0.4·percent_score。TGI=画像 preference,占比=画像 percentage,同一接口。
"""

from __future__ import annotations

from typing import Any

SCHEMA_VERSION = "v4_fifty_plus.v1"
BAND_KEYS = ("50-",)  # 50- = 50 岁及以上(纯 50+,已弃用 41-50)


def fifty_plus_score(fans_age_data: dict[str, Any] | None) -> dict[str, Any]:
    """输入 = 画像 fans.age.data(逐桶 {percentage, preference});输出 50+ 子分 dict。

    取不到任一有效 band 桶(缺桶 / 缺 percentage / 缺 preference)→ status="incomplete"。
    """
    buckets = fans_age_data if isinstance(fans_age_data, dict) else {}
    pairs: list[tuple[float, float]] = []
    for key in BAND_KEYS:
        bucket = buckets.get(key)
        if not isinstance(bucket, dict):
            continue
        pct = _to_float(bucket.get("percentage"))
        tgi = _to_float(bucket.get("preference"))
        if pct is None or tgi is None:
            continue
        pairs.append((pct, tgi))

    if not pairs:
        return {"schema_version": SCHEMA_VERSION, "status": "incomplete", "band": "50+"}

    band_percent = sum(pct for pct, _ in pairs)
    weight = band_percent if band_percent > 0 else float(len(pairs))
    band_tgi = sum(pct * tgi for pct, tgi in pairs) / weight
    tgi_score = _clamp(band_tgi / 2)
    percent_score = _clamp(band_percent)
    score = round(0.6 * tgi_score + 0.4 * percent_score, 2)
    return {
        "schema_version": SCHEMA_VERSION,
        "status": "ok",
        "score": score,
        "band": "50+",
        "band_tgi": round(band_tgi, 2),
        "band_percent": round(band_percent, 2),
        "tgi_score": round(tgi_score, 2),
        "percent_score": round(percent_score, 2),
    }


def _to_float(value: Any) -> float | None:
    if isinstance(value, (int, float)):
        return float(value)
    if isinstance(value, str):
        text = value.strip().rstrip("%").strip()
        try:
            return float(text)
        except ValueError:
            return None
    return None


def _clamp(value: float, low: float = 0.0, high: float = 100.0) -> float:
    return max(low, min(high, value))