SupplyAgent/agents/find_agent/support/age_portrait.py

"""find_agent 的确定性年龄画像标准化实现。"""
from __future__ import annotations

import json
import re
from typing import Any


def _number(value: Any) -> float | None:
    if value is None or isinstance(value, bool):
        return None
    text = str(value).strip().replace("%", "")
    if not text:
        return None
    try:
        return float(text)
    except ValueError:
        return None


def _ratio(value: Any) -> float | None:
    number = _number(value)
    if number is None:
        return None
    if number > 1:
        number /= 100
    return min(max(number, 0.0), 1.0)


def _age_dimension(portrait: dict[str, Any] | None) -> dict[str, Any]:
    if not isinstance(portrait, dict):
        return {}
    for wrapper_key in ("portrait_data", "content", "account"):
        wrapped = portrait.get(wrapper_key)
        if isinstance(wrapped, dict):
            wrapped_dimension = _age_dimension(wrapped)
            if wrapped_dimension:
                return wrapped_dimension
    for key in ("年龄", "age", "Age", "年龄分布"):
        value = portrait.get(key)
        if isinstance(value, dict):
            return value
    if portrait and all(isinstance(value, dict) for value in portrait.values()):
        return portrait
    return {}


def _bucket_kind(label: str) -> str:
    compact = (
        label.strip()
        .lower()
        .replace("岁", "")
        .replace("以上", "+")
        .replace("及", "")
        .replace(" ", "")
    )
    if compact in {"50-", "50+", ">=50", "≥50", "51+", "60+", ">=60", "≥60"}:
        return "older"
    if re.search(r"(?:^|[^0-9])(50|51|60)\+$", compact):
        return "older"

    numbers = [int(value) for value in re.findall(r"\d+", compact)]
    if not numbers:
        return "unknown"
    if len(numbers) == 1:
        if numbers[0] >= 50 and any(
            marker in compact for marker in ("+", ">=", "≥", "以上")
        ):
            return "older"
        return "unknown"

    lower, upper = min(numbers), max(numbers)
    if lower >= 50:
        return "older"
    if lower >= 40 or upper >= 50:
        return "mature"
    return "younger"


def _normalize_portrait(portrait: dict[str, Any] | None) -> dict[str, Any]:
    dimension = _age_dimension(portrait)
    buckets: list[dict[str, Any]] = []
    older_ratio = 0.0
    mature_ratio = 0.0
    older_tgi_values: list[tuple[float, float]] = []

    for label, raw_metrics in dimension.items():
        metrics = raw_metrics if isinstance(raw_metrics, dict) else {}
        percentage = _ratio(
            metrics.get("percentage")
            if "percentage" in metrics
            else metrics.get("ratio")
        )
        tgi = _number(
            metrics.get("preference")
            if "preference" in metrics
            else metrics.get("tgi")
        )
        kind = _bucket_kind(str(label))
        buckets.append(
            {
                "label": str(label),
                "kind": kind,
                "ratio": percentage,
                "tgi": tgi,
            }
        )
        if percentage is None:
            continue
        if kind == "older":
            older_ratio += percentage
            if tgi is not None:
                older_tgi_values.append((tgi, percentage))
        elif kind == "mature":
            mature_ratio += percentage

    weighted_tgi = None
    tgi_weight = sum(weight for _, weight in older_tgi_values)
    if tgi_weight:
        weighted_tgi = sum(tgi * weight for tgi, weight in older_tgi_values) / tgi_weight

    if not dimension:
        strength = "missing"
    elif (
        older_ratio >= 0.45
        or (older_ratio >= 0.35 and (weighted_tgi or 0) >= 100)
        or (older_ratio >= 0.20 and (weighted_tgi or 0) >= 130)
    ):
        strength = "strong"
    elif (
        older_ratio >= 0.20
        or (older_ratio > 0 and (weighted_tgi or 0) >= 100)
        or mature_ratio >= 0.30
    ):
        strength = "moderate"
    else:
        strength = "weak"

    return {
        "has_age_portrait": bool(dimension),
        "older_ratio": round(older_ratio, 6),
        "older_tgi": round(weighted_tgi, 4) if weighted_tgi is not None else None,
        "mature_ratio": round(mature_ratio, 6),
        "strength": strength,
        "buckets": buckets,
    }


def normalize_age_portrait_pair(
    content_portrait: dict[str, Any] | None,
    account_portrait: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Return the deterministic two-sided age normalization payload."""
    content = _normalize_portrait(content_portrait)
    account = _normalize_portrait(account_portrait)
    content_has = content["has_age_portrait"]
    account_has = account["has_age_portrait"]

    if content_has and account_has:
        strong_set = {"strong", "moderate"}
        consistency = (
            "aligned"
            if (content["strength"] in strong_set)
            == (account["strength"] in strong_set)
            else "conflict"
        )
        cap = 1.0
    elif account_has:
        consistency = "account_only"
        cap = 0.65
    elif content_has:
        consistency = "content_only"
        cap = 1.0
    else:
        consistency = "missing"
        cap = 0.35

    return {
        "content": content,
        "account": account,
        "consistency": consistency,
        "elder_score_cap": cap,
    }


def normalize_age_portraits(
    content_portrait: dict[str, Any],
    account_portrait: dict[str, Any] | None = None,
) -> str:
    """
    标准化视频点赞画像与作者粉丝画像中的年龄桶。

    能识别接口实际返回的 `50-`，以及 `50+ / 50岁以上 / >=50 / 41-50`
    等表达，统一输出直接老年比例、TGI、成熟人群代理比例和证据强度。
    本工具只标准化证据，不把点赞用户伪装成转发用户。

    Args:
        content_portrait: 视频 portrait_data，或其中的年龄字典。
        account_portrait: 可选作者 portrait_data，或其中的年龄字典。

    Returns:
        JSON，包含 content、account、consistency 和 elder_score_cap。
    """
    normalized = normalize_age_portrait_pair(content_portrait, account_portrait)
    content = normalized["content"]
    account = normalized["account"]
    consistency = normalized["consistency"]
    cap = normalized["elder_score_cap"]
    result = {
        "title": "年龄画像标准化",
        **normalized,
        "output": (
            f"视频侧={content['strength']}，作者侧={account['strength']}，"
            f"一致性={consistency}，E上限={cap}"
        ),
    }
    return json.dumps(result, ensure_ascii=False)