Public Access
306 lines
10 KiB
Python
306 lines
10 KiB
Python
"""Feedback Analyzer — turns family feedback signals into recipe discovery queries.
|
|
|
|
Reads the past N weeks of feedback (ratings, text, denial reasons, never-suggest)
|
|
and produces structured positive/negative signals plus external API search queries.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import date, timedelta
|
|
from typing import List, Optional, Dict, Any, Set
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import (
|
|
FamilyProfile,
|
|
Feedback,
|
|
MealPlanItem,
|
|
Recipe,
|
|
DenialReason,
|
|
NeverSuggest,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_DEFAULT_LOOKBACK_WEEKS = 4
|
|
_MIN_FEEDBACK_COUNT = 3
|
|
_MIN_AVG_RATING_FOR_POSITIVE = 4.0
|
|
_TOP_RATED_COUNT = 3
|
|
|
|
|
|
@dataclass
|
|
class NegativeSignal:
|
|
type: str # e.g. "avoid_ingredient", "avoid_tag", "too_spicy"
|
|
value: str
|
|
count: int
|
|
sources: Set[str] = field(default_factory=set)
|
|
|
|
|
|
@dataclass
|
|
class PositiveSignal:
|
|
type: str # e.g. "prefer_cuisine", "prefer_protein"
|
|
value: str
|
|
count: int
|
|
avg_rating: float
|
|
|
|
|
|
@dataclass
|
|
class FeedbackAnalysis:
|
|
family_id: UUID
|
|
lookback_start: date
|
|
lookback_end: date
|
|
total_feedback_count: int
|
|
positive_signals: List[PositiveSignal]
|
|
negative_signals: List[NegativeSignal]
|
|
top_rated_recipe_ids: List[UUID]
|
|
top_rated_recipe_names: List[str]
|
|
discovery_queries: List[str]
|
|
confidence: float = 0.0
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"family_id": str(self.family_id),
|
|
"lookback_start": self.lookback_start.isoformat(),
|
|
"lookback_end": self.lookback_end.isoformat(),
|
|
"total_feedback_count": self.total_feedback_count,
|
|
"positive_signals": [
|
|
{"type": s.type, "value": s.value, "count": s.count, "avg_rating": s.avg_rating}
|
|
for s in self.positive_signals
|
|
],
|
|
"negative_signals": [
|
|
{"type": s.type, "value": s.value, "count": s.count}
|
|
for s in self.negative_signals
|
|
],
|
|
"top_rated_recipe_ids": [str(r) for r in self.top_rated_recipe_ids],
|
|
"top_rated_recipe_names": self.top_rated_recipe_names,
|
|
"discovery_queries": self.discovery_queries,
|
|
"confidence": self.confidence,
|
|
}
|
|
|
|
|
|
class FeedbackAnalyzer:
|
|
"""Analyze family feedback and produce recipe discovery signals."""
|
|
|
|
def __init__(self, lookback_weeks: int = _DEFAULT_LOOKBACK_WEEKS) -> None:
|
|
self.lookback_weeks = lookback_weeks
|
|
|
|
def analyze(self, db: Session, family_id: UUID, today: date | None = None) -> FeedbackAnalysis:
|
|
today = today or date.today()
|
|
lookback_start = today - timedelta(weeks=self.lookback_weeks)
|
|
|
|
# Pull feedback from lookback window with recipe context
|
|
feedbacks = (
|
|
db.query(Feedback, MealPlanItem, Recipe)
|
|
.join(MealPlanItem, MealPlanItem.id == Feedback.meal_plan_item_id)
|
|
.join(Recipe, Recipe.id == MealPlanItem.recipe_id)
|
|
.filter(
|
|
Feedback.family_profile_id == family_id,
|
|
Feedback.created_at >= lookback_start,
|
|
)
|
|
.all()
|
|
)
|
|
|
|
total = len(feedbacks)
|
|
|
|
if total < _MIN_FEEDBACK_COUNT:
|
|
logger.info(
|
|
"FeedbackAnalyzer: only %d feedbacks in last %d weeks (< %d minimum). "
|
|
"Skipping discovery.",
|
|
total, self.lookback_weeks, _MIN_FEEDBACK_COUNT,
|
|
)
|
|
return FeedbackAnalysis(
|
|
family_id=family_id,
|
|
lookback_start=lookback_start,
|
|
lookback_end=today,
|
|
total_feedback_count=total,
|
|
positive_signals=[],
|
|
negative_signals=[],
|
|
top_rated_recipe_ids=[],
|
|
top_rated_recipe_names=[],
|
|
discovery_queries=[],
|
|
confidence=0.0,
|
|
)
|
|
|
|
# Aggregate ratings per recipe
|
|
recipe_ratings: dict[UUID, list[int]] = {}
|
|
recipe_names: dict[UUID, str] = {}
|
|
tag_ratings: dict[str, list[int]] = {}
|
|
protein_ratings: dict[str, list[int]] = {}
|
|
never_suggest_recipe_ids: set[UUID] = set()
|
|
never_suggest_ingredient_ids: set[UUID] = set()
|
|
|
|
for feedback, meal_item, recipe in feedbacks:
|
|
rid = recipe.id
|
|
recipe_names[rid] = recipe.name
|
|
recipe_ratings.setdefault(rid, []).append(feedback.rating or 3)
|
|
|
|
# Aggregate cuisine tags
|
|
for tag in (recipe.cuisine_tags or []):
|
|
tag.lower()
|
|
tag_ratings.setdefault(tag.lower(), []).append(feedback.rating or 3)
|
|
|
|
# Aggregate protein
|
|
if recipe.protein_type:
|
|
protein_ratings.setdefault(recipe.protein_type.lower(), []).append(feedback.rating or 3)
|
|
|
|
# Pull never_suggest rules
|
|
ns_rows = (
|
|
db.query(NeverSuggest)
|
|
.filter(NeverSuggest.family_profile_id == family_id)
|
|
.all()
|
|
)
|
|
for ns in ns_rows:
|
|
if ns.recipe_id:
|
|
never_suggest_recipe_ids.add(ns.recipe_id)
|
|
if ns.ingredient_id:
|
|
never_suggest_ingredient_ids.add(ns.ingredient_id)
|
|
|
|
# Build positive signals (high-rated cuisines/proteins)
|
|
positive_signals: list[PositiveSignal] = []
|
|
|
|
for tag, ratings in tag_ratings.items():
|
|
avg = sum(ratings) / len(ratings)
|
|
if avg >= _MIN_AVG_RATING_FOR_POSITIVE and len(ratings) >= 2:
|
|
positive_signals.append(
|
|
PositiveSignal(
|
|
type="prefer_cuisine",
|
|
value=tag,
|
|
count=len(ratings),
|
|
avg_rating=round(avg, 2),
|
|
)
|
|
)
|
|
|
|
for protein, ratings in protein_ratings.items():
|
|
avg = sum(ratings) / len(ratings)
|
|
if avg >= _MIN_AVG_RATING_FOR_POSITIVE and len(ratings) >= 2:
|
|
positive_signals.append(
|
|
PositiveSignal(
|
|
type="prefer_protein",
|
|
value=protein,
|
|
count=len(ratings),
|
|
avg_rating=round(avg, 2),
|
|
)
|
|
)
|
|
|
|
# Sort by avg_rating desc, then count desc
|
|
positive_signals.sort(key=lambda s: (-s.avg_rating, -s.count))
|
|
|
|
# Build negative signals from denial reasons + never_suggest
|
|
negative_signals: list[NegativeSignal] = []
|
|
|
|
denial_counts: dict[str, int] = {}
|
|
for feedback, meal_item, recipe in feedbacks:
|
|
if feedback.denial_reason:
|
|
key = f"denial_{feedback.denial_reason.value}"
|
|
denial_counts[key] = denial_counts.get(key, 0) + 1
|
|
|
|
for reason, count in denial_counts.items():
|
|
sig_type = reason.replace("denial_", "")
|
|
negative_signals.append(
|
|
NegativeSignal(
|
|
type=f"denial_{sig_type}",
|
|
value=sig_type,
|
|
count=count,
|
|
)
|
|
)
|
|
|
|
# Top rated recipes (for "similar to X" queries)
|
|
recipe_avgs = {
|
|
rid: sum(ratings) / len(ratings)
|
|
for rid, ratings in recipe_ratings.items()
|
|
}
|
|
top_rated = sorted(
|
|
recipe_avgs.items(), key=lambda x: -x[1]
|
|
)[:_TOP_RATED_COUNT]
|
|
top_rated_ids = [rid for rid, _ in top_rated]
|
|
top_rated_names = [recipe_names[rid] for rid in top_rated_ids]
|
|
|
|
# Build discovery queries
|
|
queries = _build_discovery_queries(
|
|
positive_signals=positive_signals,
|
|
top_rated_names=top_rated_names,
|
|
negative_signals=negative_signals,
|
|
)
|
|
|
|
# Confidence = proportion of positive signals that are well-supported
|
|
confidence = 0.0
|
|
if positive_signals:
|
|
well_supported = sum(1 for s in positive_signals if s.count >= 2)
|
|
confidence = well_supported / len(positive_signals)
|
|
|
|
logger.info(
|
|
"FeedbackAnalyzer: family=%s feedbacks=%d positives=%d negatives=%d queries=%d confidence=%.2f",
|
|
family_id, total, len(positive_signals), len(negative_signals), len(queries), confidence,
|
|
)
|
|
|
|
return FeedbackAnalysis(
|
|
family_id=family_id,
|
|
lookback_start=lookback_start,
|
|
lookback_end=today,
|
|
total_feedback_count=total,
|
|
positive_signals=positive_signals,
|
|
negative_signals=negative_signals,
|
|
top_rated_recipe_ids=top_rated_ids,
|
|
top_rated_recipe_names=top_rated_names,
|
|
discovery_queries=queries,
|
|
confidence=confidence,
|
|
)
|
|
|
|
|
|
def _build_discovery_queries(
|
|
positive_signals: List[PositiveSignal],
|
|
top_rated_names: List[str],
|
|
negative_signals: List[NegativeSignal],
|
|
) -> List[str]:
|
|
"""Convert signals into Spoonacular search queries.
|
|
|
|
Strategy:
|
|
1. Combine top cuisine + top protein → "mexican shrimp"
|
|
2. High-rated specific recipes → search by name
|
|
3. Cap at 5 queries to stay within free quota
|
|
"""
|
|
queries = []
|
|
seen = set()
|
|
|
|
# Extract top cuisines and proteins
|
|
cuisines = [s.value for s in positive_signals if s.type == "prefer_cuisine"]
|
|
proteins = [s.value for s in positive_signals if s.type == "prefer_protein"]
|
|
|
|
# Cross product of top cuisine + protein
|
|
for cuisine in cuisines[:2]:
|
|
for protein in proteins[:2]:
|
|
q = f"{cuisine} {protein}"
|
|
if q not in seen:
|
|
queries.append(q)
|
|
seen.add(q)
|
|
if len(queries) >= 5:
|
|
return queries
|
|
|
|
# Single cuisine or protein queries
|
|
for cuisine in cuisines[:2]:
|
|
if cuisine not in seen:
|
|
queries.append(cuisine)
|
|
seen.add(cuisine)
|
|
if len(queries) >= 5:
|
|
return queries
|
|
|
|
for protein in proteins[:2]:
|
|
if protein not in seen:
|
|
queries.append(protein)
|
|
seen.add(protein)
|
|
if len(queries) >= 5:
|
|
return queries
|
|
|
|
# Top-rated recipe names (people liked these, find similar)
|
|
for name in top_rated_names[:2]:
|
|
if name not in seen:
|
|
queries.append(name)
|
|
seen.add(name)
|
|
if len(queries) >= 5:
|
|
return queries
|
|
|
|
return queries
|