Coverage for apps / ai / services / discover.py: 14%
123 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-14 19:13 +0000
1"""AI discovery suggestions service."""
3import logging
4from datetime import datetime, timedelta
5from typing import Optional, List
7from django.utils import timezone
9from apps.profiles.models import Profile
10from apps.recipes.models import RecipeFavorite, RecipeViewHistory
12from ..models import AIPrompt, AIDiscoverySuggestion
13from .openrouter import OpenRouterService, AIUnavailableError, AIResponseError
14from .validator import AIResponseValidator, ValidationError
16logger = logging.getLogger(__name__)
18# Suggestions are cached for 24 hours
19CACHE_DURATION_HOURS = 24
22def get_discover_suggestions(profile_id: int) -> dict:
23 """Get AI discovery suggestions for a profile.
25 Returns cached suggestions if still valid (within 24 hours),
26 otherwise generates new suggestions via AI.
28 For new users (no favorites), only seasonal suggestions are returned.
30 Args:
31 profile_id: The ID of the profile to get suggestions for.
33 Returns:
34 Dict with suggestions array and refresh timestamp.
36 Raises:
37 Profile.DoesNotExist: If profile not found.
38 AIUnavailableError: If AI service is not available.
39 """
40 profile = Profile.objects.get(id=profile_id)
42 # Check for cached suggestions (within 24 hours)
43 cache_cutoff = timezone.now() - timedelta(hours=CACHE_DURATION_HOURS)
44 cached = AIDiscoverySuggestion.objects.filter(profile=profile, created_at__gte=cache_cutoff)
46 if cached.exists():
47 logger.info(f"Returning cached discover suggestions for profile {profile_id}")
48 return _format_suggestions(cached)
50 # Clear old suggestions
51 AIDiscoverySuggestion.objects.filter(profile=profile).delete()
53 # Check if user has viewing history
54 has_history = RecipeViewHistory.objects.filter(profile=profile).exists()
56 # Generate new suggestions
57 suggestions = []
59 # Always generate seasonal suggestions
60 seasonal_suggestions = _generate_seasonal_suggestions(profile)
61 suggestions.extend(seasonal_suggestions)
63 # Only generate personalized suggestions if user has history
64 if has_history:
65 # Recommended based on what they've viewed
66 recommended_suggestions = _generate_recommended_suggestions(profile)
67 suggestions.extend(recommended_suggestions)
69 # Try something new/different
70 new_suggestions = _generate_new_suggestions(profile)
71 suggestions.extend(new_suggestions)
73 if not suggestions:
74 # If no suggestions generated, return empty
75 return {
76 "suggestions": [],
77 "refreshed_at": timezone.now().isoformat(),
78 }
80 return _format_suggestions(suggestions)
83def _format_suggestions(suggestions) -> dict:
84 """Format suggestions for API response."""
85 result = []
87 for suggestion in suggestions:
88 result.append(
89 {
90 "type": suggestion.suggestion_type,
91 "title": suggestion.title,
92 "description": suggestion.description,
93 "search_query": suggestion.search_query,
94 }
95 )
97 # Get the most recent created_at for refreshed_at
98 if hasattr(suggestions, "first"):
99 # QuerySet
100 first = suggestions.first()
101 refreshed_at = first.created_at.isoformat() if first else timezone.now().isoformat()
102 else:
103 # List
104 refreshed_at = suggestions[0].created_at.isoformat() if suggestions else timezone.now().isoformat()
106 return {
107 "suggestions": result,
108 "refreshed_at": refreshed_at,
109 }
112def _generate_seasonal_suggestions(profile: Profile) -> List[AIDiscoverySuggestion]:
113 """Generate seasonal/holiday suggestions."""
114 try:
115 prompt = AIPrompt.get_prompt("discover_seasonal")
116 except AIPrompt.DoesNotExist:
117 logger.warning("discover_seasonal prompt not found")
118 return []
120 # Get current date info
121 now = datetime.now()
122 date_str = now.strftime("%B %d, %Y") # e.g., "January 15, 2024"
123 season = _get_season(now)
125 user_prompt = prompt.format_user_prompt(
126 date=date_str,
127 season=season,
128 )
130 try:
131 service = OpenRouterService()
132 response = service.complete(
133 system_prompt=prompt.system_prompt,
134 user_prompt=user_prompt,
135 model=prompt.model,
136 json_response=True,
137 )
139 validator = AIResponseValidator()
140 validated = validator.validate("discover_seasonal", response)
142 # Create and save suggestions
143 suggestions = []
144 for item in validated:
145 suggestion = AIDiscoverySuggestion.objects.create(
146 profile=profile,
147 suggestion_type="seasonal",
148 search_query=item["search_query"],
149 title=item["title"],
150 description=item["description"],
151 )
152 suggestions.append(suggestion)
154 logger.info(f"Generated {len(suggestions)} seasonal suggestions for profile {profile.id}")
155 return suggestions
157 except (AIUnavailableError, AIResponseError, ValidationError) as e:
158 logger.warning(f"Failed to generate seasonal suggestions: {e}")
159 return []
162def _generate_recommended_suggestions(profile: Profile) -> List[AIDiscoverySuggestion]:
163 """Generate suggestions based on user's viewing history."""
164 try:
165 prompt = AIPrompt.get_prompt("discover_favorites")
166 except AIPrompt.DoesNotExist:
167 logger.warning("discover_favorites prompt not found")
168 return []
170 # Get user's recently viewed recipes
171 history = RecipeViewHistory.objects.filter(profile=profile).select_related("recipe").order_by("-viewed_at")[:15]
172 if not history:
173 return []
175 # Format history list
176 history_list = "\n".join(
177 f"- {item.recipe.title} ({item.recipe.cuisine or item.recipe.category or 'uncategorized'})" for item in history
178 )
180 user_prompt = prompt.format_user_prompt(favorites=history_list)
182 try:
183 service = OpenRouterService()
184 response = service.complete(
185 system_prompt=prompt.system_prompt,
186 user_prompt=user_prompt,
187 model=prompt.model,
188 json_response=True,
189 )
191 validator = AIResponseValidator()
192 validated = validator.validate("discover_favorites", response)
194 # Create and save suggestions
195 suggestions = []
196 for item in validated:
197 suggestion = AIDiscoverySuggestion.objects.create(
198 profile=profile,
199 suggestion_type="favorites", # Keep type for frontend compatibility
200 search_query=item["search_query"],
201 title=item["title"],
202 description=item["description"],
203 )
204 suggestions.append(suggestion)
206 logger.info(f"Generated {len(suggestions)} recommended suggestions for profile {profile.id}")
207 return suggestions
209 except (AIUnavailableError, AIResponseError, ValidationError) as e:
210 logger.warning(f"Failed to generate recommended suggestions: {e}")
211 return []
214def _generate_new_suggestions(profile: Profile) -> List[AIDiscoverySuggestion]:
215 """Generate suggestions for trying something new."""
216 try:
217 prompt = AIPrompt.get_prompt("discover_new")
218 except AIPrompt.DoesNotExist:
219 logger.warning("discover_new prompt not found")
220 return []
222 # Get user's recent recipes from history to understand their preferences
223 history = RecipeViewHistory.objects.filter(profile=profile).select_related("recipe").order_by("-viewed_at")[:15]
224 if not history:
225 return []
227 recent_recipes = "\n".join(
228 f"- {item.recipe.title} ({item.recipe.cuisine or 'unknown cuisine'}, {item.recipe.category or 'unknown category'})"
229 for item in history
230 )
232 user_prompt = prompt.format_user_prompt(history=recent_recipes)
234 try:
235 service = OpenRouterService()
236 response = service.complete(
237 system_prompt=prompt.system_prompt,
238 user_prompt=user_prompt,
239 model=prompt.model,
240 json_response=True,
241 )
243 validator = AIResponseValidator()
244 validated = validator.validate("discover_new", response)
246 # Create and save suggestions
247 suggestions = []
248 for item in validated:
249 suggestion = AIDiscoverySuggestion.objects.create(
250 profile=profile,
251 suggestion_type="new",
252 search_query=item["search_query"],
253 title=item["title"],
254 description=item["description"],
255 )
256 suggestions.append(suggestion)
258 logger.info(f"Generated {len(suggestions)} try-new suggestions for profile {profile.id}")
259 return suggestions
261 except (AIUnavailableError, AIResponseError, ValidationError) as e:
262 logger.warning(f"Failed to generate try-new suggestions: {e}")
263 return []
266def _get_season(dt: datetime) -> str:
267 """Get the season for a given date (Northern Hemisphere)."""
268 month = dt.month
269 if month in (12, 1, 2):
270 return "winter"
271 elif month in (3, 4, 5):
272 return "spring"
273 elif month in (6, 7, 8):
274 return "summer"
275 else:
276 return "autumn"