SupplyAgent/agents/demand_grade_agent/tools/shared.py

"""demand_grade_agent 工具共享辅助函数（非 @tool，不对外暴露为工具）。"""
from __future__ import annotations

import json
from decimal import Decimal
from typing import Any

from supply_infra.db.models.global_tree_category import GlobalTreeCategory
from supply_infra.db.models.multi_demand_pool_di import MultiDemandPoolDi

_VIDEO_LIST_LIMIT = 10

VALID_GRADES: tuple[str, ...] = ("S", "A", "B", "C", "D")


def normalize_biz_dt(biz_dt: str | None) -> tuple[str | None, str | None]:
    """校验并规范化 biz_dt（YYYYMMDD）；空值返回 (None, None) 表示未指定。"""
    if biz_dt is None:
        return None, None
    text = str(biz_dt).strip()
    if not text:
        return None, None
    if len(text) != 8 or not text.isdigit():
        return None, f"biz_dt 格式无效，应为 YYYYMMDD: {biz_dt!r}"
    return text, None


def to_float(value: Any) -> float | None:
    """将 Decimal/None/数值统一转换为 float，None 原样返回。"""
    if value is None:
        return None
    return float(value)


def format_score(avg: Decimal | float | int | None) -> str:
    """格式化分数为易读文本，None 显示为 —。"""
    if avg is None:
        return "—"
    value = float(avg)
    if value >= 100:
        return f"{value:.0f}"
    if value >= 1:
        return f"{value:.2f}"
    return f"{value:.4f}"


def percentile(sorted_values: list[float], pct: float) -> float | None:
    """对已排序（升序）的数值列表求分位数（线性插值），pct 取 0~100。"""
    if not sorted_values:
        return None
    if len(sorted_values) == 1:
        return sorted_values[0]

    rank = (pct / 100) * (len(sorted_values) - 1)
    lower_idx = int(rank)
    upper_idx = min(lower_idx + 1, len(sorted_values) - 1)
    frac = rank - lower_idx
    return sorted_values[lower_idx] + (sorted_values[upper_idx] - sorted_values[lower_idx]) * frac


def distribution_summary(values: list[float]) -> dict[str, float | int | None]:
    """返回一组数值的 min/p25/p50/p75/p90/max/count 分布摘要。"""
    if not values:
        return {"count": 0, "min": None, "p25": None, "p50": None, "p75": None, "p90": None, "max": None}
    ordered = sorted(values)
    return {
        "count": len(ordered),
        "min": ordered[0],
        "p25": percentile(ordered, 25),
        "p50": percentile(ordered, 50),
        "p75": percentile(ordered, 75),
        "p90": percentile(ordered, 90),
        "max": ordered[-1],
    }


def _normalize_parent_id(parent_id: int | None) -> int | None:
    if parent_id is None or parent_id == 0:
        return None
    return parent_id


def build_category_path(
    category_id: int,
    by_id: dict[int, GlobalTreeCategory],
) -> str | None:
    """从根到指定节点的名称路径，如「美妆护肤 > 护肤 > 防晒」。"""
    cat = by_id.get(category_id)
    if cat is None:
        return None
    names: list[str] = []
    current: GlobalTreeCategory | None = cat
    seen: set[int] = set()
    while current is not None:
        cid = int(current.id)
        if cid in seen:
            break
        seen.add(cid)
        names.append(current.name or str(cid))
        parent_key = _normalize_parent_id(current.parent_id)
        current = by_id.get(parent_key) if parent_key is not None else None
    names.reverse()
    return " > ".join(names)


def normalize_str_list(raw: Any, field_name: str = "items") -> tuple[list[str], str | None]:
    """将 JSON 文本/列表/单个字符串统一解析为非空 str 列表（去重保序）。"""
    if raw is None:
        return [], f"{field_name} 不能为空"

    items: list[Any]
    if isinstance(raw, str):
        text = raw.strip()
        if not text:
            return [], f"{field_name} 不能为空"
        try:
            parsed = json.loads(text)
            items = list(parsed) if isinstance(parsed, list) else [text]
        except (ValueError, TypeError):
            items = [text]
    elif isinstance(raw, (list, tuple)):
        items = list(raw)
    else:
        items = [raw]

    out: list[str] = []
    seen: set[str] = set()
    for item in items:
        text = str(item).strip() if item is not None else ""
        if not text or text in seen:
            continue
        seen.add(text)
        out.append(text)

    if not out:
        return [], f"{field_name} 不能为空"
    return out, None


def parse_int_list(raw: Any) -> list[int]:
    """将 JSON 文本/列表统一解析为 int 列表，解析失败返回空列表。"""
    if raw is None:
        return []
    if isinstance(raw, str):
        try:
            raw = json.loads(raw)
        except (ValueError, TypeError):
            return []
    if not isinstance(raw, list):
        return []
    out: list[int] = []
    for item in raw:
        try:
            out.append(int(item))
        except (TypeError, ValueError):
            continue
    return out


def dump_int_list(values: list[int] | None) -> str | None:
    """将 int 列表序列化为 JSON 文本，空列表/None 返回 None。"""
    if not values:
        return None
    return json.dumps(values, ensure_ascii=False)


def _parse_video_ids(raw: Any) -> list[str]:
    """解析单个 multi_demand_pool_di.video_list（JSON数组/逗号分隔文本）为 vid 字符串列表。"""
    if raw is None:
        return []
    items: list[Any]
    if isinstance(raw, str):
        text = raw.strip()
        if not text:
            return []
        try:
            parsed = json.loads(text)
            items = list(parsed) if isinstance(parsed, list) else [text]
        except (ValueError, TypeError):
            items = [part.strip() for part in text.split(",") if part.strip()]
    elif isinstance(raw, (list, tuple)):
        items = list(raw)
    else:
        return []
    return [str(v).strip() for v in items if v is not None and str(v).strip()]


def merge_video_ids(pool_rows: list[MultiDemandPoolDi], limit: int = _VIDEO_LIST_LIMIT) -> str | None:
    """合并多条原始需求行的 video_list，去重保序，最多取前 limit 个，返回 JSON 文本或 None。"""
    merged: list[str] = []
    seen: set[str] = set()
    for row in pool_rows:
        for vid in _parse_video_ids(row.video_list):
            if vid not in seen:
                seen.add(vid)
                merged.append(vid)
    if not merged:
        return None
    return json.dumps(merged[:limit], ensure_ascii=False)


def collect_strategies(pool_rows: list[MultiDemandPoolDi]) -> str | None:
    """收集多条原始需求行的策略名，去重排序，返回 JSON 文本或 None。"""
    strategies = sorted({row.strategy.strip() for row in pool_rows if row.strategy and row.strategy.strip()})
    if not strategies:
        return None
    return json.dumps(strategies, ensure_ascii=False)