Public Access
111 lines
3.5 KiB
Python
111 lines
3.5 KiB
Python
"""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
|