SupplyAgent/agents/find_agent/tools/decision_support.py

"""find_agent 的确定性年龄画像标准化与流程审计工具。"""
from __future__ import annotations

import json
import re
from typing import Any

from supply_agent.tools import tool

_ROOT_SOURCE_TYPES = {"demand", "seed", "point", "mixed"}


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,
    }


@tool
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)


def _float_score(candidate: dict[str, Any], key: str) -> float | None:
    value = candidate.get(key)
    try:
        number = float(value)
    except (TypeError, ValueError):
        return None
    return number if 0 <= number <= 1 else None


@tool
def audit_video_discovery_process(
    searches: list[dict[str, Any]],
    candidates: list[dict[str, Any]],
    intended_status: str = "finished",
) -> str:
    """
    在结束找片前审计搜索树、翻页、标签扩展、证据完备性和双池分流。

    Args:
        searches: 已执行搜索页。建议包含 search_id、keyword、source_type、
            parent_search_id、cursor、page_no、has_more、new_candidate_count。
        candidates: 已评估候选。建议包含 aweme_id、decision_bucket、R/E/S 分数、
            detail_verified、content_portrait_attempted、account_portrait_attempted、
            content_analysis_verified、has_video_url、expansion_worthy_tags。
        intended_status: 准备设置的运行状态，通常为 finished。

    Returns:
        JSON，包含 can_finish、critical_violations、warnings 和 coverage。
    """
    critical: list[str] = []
    warnings: list[str] = []
    coverage_violations: list[str] = []
    valid_searches = [item for item in searches if isinstance(item, dict)]
    valid_candidates = [item for item in candidates if isinstance(item, dict)]

    roots = [
        item
        for item in valid_searches
        if item.get("source_type") in _ROOT_SOURCE_TYPES
        and not item.get("parent_search_id")
        and int(item.get("page_no") or 1) == 1
    ]
    root_keywords = {
        str(item.get("keyword") or "").strip() for item in roots if item.get("keyword")
    }
    if len(root_keywords) < 2:
        coverage_violations.append("独立根搜索词少于2个")

    by_parent = {
        int(item["parent_search_id"])
        for item in valid_searches
        if item.get("parent_search_id") is not None
    }
    keyword_pages = {
        (str(item.get("keyword") or ""), int(item.get("page_no") or 1))
        for item in valid_searches
    }
    for item in valid_searches:
        page_no = int(item.get("page_no") or 1)
        if (
            page_no != 1
            or not item.get("has_more")
            or int(item.get("new_candidate_count") or 0) <= 0
        ):
            continue
        search_id = item.get("search_id")
        keyword = str(item.get("keyword") or "")
        followed = (
            search_id is not None and int(search_id) in by_parent
        ) or (keyword, page_no + 1) in keyword_pages
        if not followed:
            coverage_violations.append(
                f"生产性搜索页未翻页: search_id={search_id}, keyword={keyword}"
            )

    tag_searches = [
        item for item in valid_searches if item.get("source_type") == "tag"
    ]
    worthy_tags = {
        str(tag)
        for candidate in valid_candidates
        for tag in (candidate.get("expansion_worthy_tags") or [])
        if str(tag).strip()
    }
    if worthy_tags and not tag_searches:
        coverage_violations.append("存在值得扩展的标签，但没有 tag 搜索分支")

    bucket_counts = {
        "primary": 0,
        "backup": 0,
        "rejected": 0,
        "pending_evaluation": 0,
    }
    for candidate in valid_candidates:
        aweme_id = str(candidate.get("aweme_id") or "unknown")
        bucket = str(
            candidate.get("decision_bucket") or "pending_evaluation"
        )
        if bucket == "unreviewed":
            bucket = "pending_evaluation"
        bucket_counts[bucket] = bucket_counts.get(bucket, 0) + 1
        relevance = _float_score(candidate, "relevance_score")
        elder = _float_score(candidate, "elder_score")
        share = _float_score(candidate, "share_score")

        if bucket in {"primary", "backup"}:
            if not candidate.get("detail_verified"):
                critical.append(f"{aweme_id} 未核验详情")
            if not candidate.get("content_portrait_attempted"):
                critical.append(f"{aweme_id} 未尝试视频画像")
            if not candidate.get("account_portrait_attempted"):
                critical.append(f"{aweme_id} 未尝试作者画像")
            if not candidate.get("age_portraits_normalized"):
                critical.append(f"{aweme_id} 未标准化年龄画像")
            if candidate.get("has_video_url") and not candidate.get(
                "content_analysis_verified"
            ):
                critical.append(f"{aweme_id} 有播放地址但未核验视频内容")

        if bucket == "primary" and not (
            relevance is not None
            and relevance >= 0.55
            and elder is not None
            and elder >= 0.55
            and share is not None
            and share >= 0.50
        ):
            critical.append(f"{aweme_id} 不满足 primary 阈值")
        meets_primary = (
            relevance is not None
            and relevance >= 0.55
            and elder is not None
            and elder >= 0.55
            and share is not None
            and share >= 0.50
        )
        meets_backup = (
            not meets_primary
            and elder is not None
            and elder >= 0.50
            and share is not None
            and share >= 0.50
        )
        if bucket == "backup" and not meets_backup:
            critical.append(f"{aweme_id} 不满足 backup 阈值")
        if bucket == "rejected" and (meets_primary or meets_backup):
            critical.append(f"{aweme_id} 达到保留线却被错误淘汰")
        if bucket == "pending_evaluation":
            warnings.append(f"{aweme_id} 等待 Agent 补证和评估")

    retained_count = (
        bucket_counts.get("primary", 0)
        + bucket_counts.get("backup", 0)
    )
    if retained_count < 5:
        warnings.append(
            f"当前保留 {retained_count} 条，低于优先目标 5 条；"
            "若仍有高价值搜索前沿应继续探索，候选确实不足时允许结束"
        )
    if retained_count > 0:
        exploration_note = (
            "已达到 5 条优先目标"
            if retained_count >= 5
            else "尚未达到 5 条优先目标；仅在剩余前沿价值较低时允许结束"
        )
        warnings.extend(
            f"{exploration_note}；未继续探索: {item}"
            for item in coverage_violations
        )
    else:
        critical.extend(coverage_violations)

    if not valid_candidates:
        warnings.append("没有候选；应确认是搜索无结果而非提前停止")
    can_finish = intended_status != "finished" or not critical
    result = {
        "title": "视频发现流程审计",
        "can_finish": can_finish,
        "critical_violations": list(dict.fromkeys(critical)),
        "warnings": list(dict.fromkeys(warnings)),
        "coverage": {
            "search_pages": len(valid_searches),
            "root_keywords": sorted(root_keywords),
            "tag_search_count": len(tag_searches),
            "candidate_count": len(valid_candidates),
            "retained_candidate_count": retained_count,
            "bucket_counts": bucket_counts,
        },
        "output": (
            f"can_finish={can_finish}，严重问题 {len(set(critical))} 个，"
            f"警告 {len(set(warnings))} 个"
        ),
    }
    return json.dumps(result, ensure_ascii=False)