"""Per-recipe scoring with the 5 weighted signals from the spec.""" from __future__ import annotations from datetime import date from decimal import Decimal from typing import Dict, Iterable, List, Optional from uuid import UUID from app.services.planner.config import PlannerConfig from app.services.planner.types import RecipeCost, ScoredRecipe def time_bonus(total_minutes: int, config: PlannerConfig) -> float: if total_minutes <= config.time_ideal_minutes: return 1.0 if total_minutes >= config.time_full_minutes: return 0.0 span = config.time_full_minutes - config.time_ideal_minutes over = total_minutes - config.time_ideal_minutes return max(0.0, min(1.0, 1.0 - over / span)) def recency_bonus( last_cooked: Optional[date], today: date, config: PlannerConfig, ) -> float: if last_cooked is None: return 1.0 weeks_ago = (today - last_cooked).days / 7 if weeks_ago >= config.recency_full_weeks: return 1.0 # Below recency_weeks the recipe wouldn't be in the feasible set, so we treat # exactly recency_weeks as score 0 and recency_full_weeks as score 1. if weeks_ago <= config.recency_weeks: return 0.0 span = config.recency_full_weeks - config.recency_weeks over = weeks_ago - config.recency_weeks return max(0.0, min(1.0, over / span)) def normalize_savings(values: List[Decimal]) -> List[float]: if not values: return [] floats = [float(v) for v in values] lo, hi = min(floats), max(floats) if hi == lo: return [0.0] * len(floats) return [(v - lo) / (hi - lo) for v in floats] def score_recipes( *, recipes: Iterable[dict], recipe_costs: Dict[UUID, RecipeCost], last_cooked_at: Dict[UUID, date], config: PlannerConfig, today: date, ) -> List[ScoredRecipe]: """Returns recipes scored DESC. Caller passes only the feasible set.""" materialized = list(recipes) if not materialized: return [] savings = [recipe_costs[r["id"]].total_savings for r in materialized] norm_savings = normalize_savings(savings) out: List[ScoredRecipe] = [] for r, ns in zip(materialized, norm_savings): rid = r["id"] if isinstance(rid, str): rid = UUID(rid) cost = recipe_costs[rid] total_min = int(r.get("prep_time_minutes") or 0) + int(r.get("cook_time_minutes") or 0) tb = time_bonus(total_min, config) rb = recency_bonus(last_cooked_at.get(rid), today, config) components = { "savings_normalized": ns, "sale_coverage_pct": cost.sale_coverage_pct, "pantry_hit_pct": cost.pantry_hit_pct, "time_bonus": tb, "recency_bonus": rb, "savings_dollars": float(cost.total_savings), } score = ( config.w_savings * ns + config.w_coverage * cost.sale_coverage_pct + config.w_pantry * cost.pantry_hit_pct + config.w_time * tb + config.w_recency * rb ) cuisine_tags = r.get("cuisine_tags") or [] primary_cuisine = cuisine_tags[0] if cuisine_tags else None out.append( ScoredRecipe( recipe_id=rid, score=score, components=components, cost=cost, protein=r.get("protein_type"), cuisine=primary_cuisine, ) ) out.sort(key=lambda s: s.score, reverse=True) return out