108 lines
2.6 KiB
Python
108 lines
2.6 KiB
Python
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any, Iterable
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def clamp(value: float, min_value: float, max_value: float) -> float:
|
||
if value != value: # NaN
|
||
return min_value
|
||
return max(min_value, min(max_value, value))
|
||
|
||
|
||
def as_finite_float(value: Any, *, default: float) -> float:
|
||
try:
|
||
f = float(value)
|
||
except Exception:
|
||
return float(default)
|
||
if f != f:
|
||
return float(default)
|
||
if f == float("inf") or f == float("-inf"):
|
||
return float(default)
|
||
return f
|
||
|
||
|
||
def normalize_int_id_set(values: Iterable[Any]) -> set[int]:
|
||
"""
|
||
将历史 ID 列表归一化为 int 集合（支持 str/int 混用）。
|
||
|
||
说明：
|
||
- 无法转换的值会被忽略，并记录 debug 日志（不影响主流程）
|
||
"""
|
||
|
||
out: set[int] = set()
|
||
for v in values:
|
||
try:
|
||
if isinstance(v, bool):
|
||
# 避免 True/False 被当作 1/0
|
||
raise ValueError("bool 不是合法 id")
|
||
out.add(int(v))
|
||
except Exception:
|
||
logger.debug("历史 id 无法转为 int，已忽略：%r", v)
|
||
return out
|
||
|
||
|
||
def jaccard(a: set[str], b: set[str]) -> float:
|
||
if not a and not b:
|
||
return 0.0
|
||
inter = len(a & b)
|
||
union = len(a | b)
|
||
return float(inter) / float(union) if union > 0 else 0.0
|
||
|
||
|
||
def argmax_key(d: dict[str, Any] | None) -> str | None:
|
||
"""
|
||
从 suitability 字典中取最大值 key（V1 用作代表标签）。
|
||
- 空字典/None -> None
|
||
- 值非法 -> 按 default=0 处理
|
||
"""
|
||
|
||
if not d:
|
||
return None
|
||
best_k: str | None = None
|
||
best_v = float("-inf")
|
||
for k, v in d.items():
|
||
fv = as_finite_float(v, default=0.0)
|
||
if fv > best_v:
|
||
best_v = fv
|
||
best_k = k
|
||
return best_k
|
||
|
||
|
||
def build_tags(candidate: Any) -> set[str]:
|
||
"""
|
||
构造离散标签集合（V1 写死）：
|
||
- stage:<stage>
|
||
- need:<argmax_key>
|
||
- context:<argmax_key>
|
||
|
||
说明：
|
||
- candidate 可能是 ScoredCandidate 或具备 content_profile 的对象
|
||
- 字段缺失时自动降级（只返回可得标签）
|
||
"""
|
||
|
||
tags: set[str] = set()
|
||
|
||
cp = getattr(candidate, "content_profile", None)
|
||
if cp is None:
|
||
return tags
|
||
|
||
stage = getattr(cp, "stage", None)
|
||
if stage:
|
||
tags.add(f"stage:{stage}")
|
||
|
||
need = getattr(cp, "need_suitability", None)
|
||
need_k = argmax_key(need)
|
||
if need_k:
|
||
tags.add(f"need:{need_k}")
|
||
|
||
ctx = getattr(cp, "context_suitability", None)
|
||
ctx_k = argmax_key(ctx)
|
||
if ctx_k:
|
||
tags.add(f"context:{ctx_k}")
|
||
|
||
return tags
|
||
|