from __future__ import annotations from typing import Optional from app.features.personalized_reco.observability.types import EmptyReason, MissingFields def compute_missing_fields(user_profile: object) -> MissingFields: """ 判定用户画像缺失字段(对齐算法规则 V1.2 口径)。 规则: - need:user_profile.need 为空对象 {} 或不存在 - context:user_profile.context 为空对象 {} 或不存在 - emotion:user_profile.emotion_score 为 None 或不存在 """ need = getattr(user_profile, "need", None) context = getattr(user_profile, "context", None) emotion_score = getattr(user_profile, "emotion_score", None) need_missing = not bool(need) context_missing = not bool(context) emotion_missing = emotion_score is None return MissingFields(need=need_missing, context=context_missing, emotion=emotion_missing) def compute_empty_reason( *, served_k: int, candidate_pool_size_raw: int, candidate_pool_size_after_hard_filter: int, candidate_pool_size_after_freqcap: int, ) -> Optional[EmptyReason]: """ 判定 empty_reason(served_k=0 必填)。 规则(对齐 plan): - served_k>0 -> None - raw==0 -> pool_empty - raw>0 且 after_hard_filter==0 -> hard_filter_all - after_freqcap==0 -> freqcap_all - 其他 -> unknown """ if int(served_k) > 0: return None raw = int(candidate_pool_size_raw) after_hard = int(candidate_pool_size_after_hard_filter) after_freqcap = int(candidate_pool_size_after_freqcap) if raw == 0: return "pool_empty" if raw > 0 and after_hard == 0: return "hard_filter_all" if after_freqcap == 0: return "freqcap_all" return "unknown"