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: - need: - context: 说明: - 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