Coverage for apps / ai / services / discover.py: 14%

1"""AI discovery suggestions service.""" 

2 

3import logging 

4from datetime import datetime, timedelta 

5from typing import Optional, List 

6 

7from django.utils import timezone 

8 

9from apps.profiles.models import Profile 

10from apps.recipes.models import RecipeFavorite, RecipeViewHistory 

11 

12from ..models import AIPrompt, AIDiscoverySuggestion 

13from .openrouter import OpenRouterService, AIUnavailableError, AIResponseError 

14from .validator import AIResponseValidator, ValidationError 

15 

16logger = logging.getLogger(__name__) 

17 

18# Suggestions are cached for 24 hours 

19CACHE_DURATION_HOURS = 24 

20 

21 

22def get_discover_suggestions(profile_id: int) -> dict: 

23 """Get AI discovery suggestions for a profile. 

24 

25 Returns cached suggestions if still valid (within 24 hours), 

26 otherwise generates new suggestions via AI. 

27 

28 For new users (no favorites), only seasonal suggestions are returned. 

29 

30 Args: 

31 profile_id: The ID of the profile to get suggestions for. 

32 

33 Returns: 

34 Dict with suggestions array and refresh timestamp. 

35 

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) 

41 

42 # Check for cached suggestions (within 24 hours) 

43 cache_cutoff = timezone.now() - timedelta(hours=CACHE_DURATION_HOURS) 

44 cached = AIDiscoverySuggestion.objects.filter( 

45 profile=profile, 

46 created_at__gte=cache_cutoff 

47 ) 

48 

49 if cached.exists(): 

50 logger.info(f'Returning cached discover suggestions for profile {profile_id}') 

51 return _format_suggestions(cached) 

52 

53 # Clear old suggestions 

54 AIDiscoverySuggestion.objects.filter(profile=profile).delete() 

55 

56 # Check if user has viewing history 

57 has_history = RecipeViewHistory.objects.filter(profile=profile).exists() 

58 

59 # Generate new suggestions 

60 suggestions = [] 

61 

62 # Always generate seasonal suggestions 

63 seasonal_suggestions = _generate_seasonal_suggestions(profile) 

64 suggestions.extend(seasonal_suggestions) 

65 

66 # Only generate personalized suggestions if user has history 

67 if has_history: 

68 # Recommended based on what they've viewed 

69 recommended_suggestions = _generate_recommended_suggestions(profile) 

70 suggestions.extend(recommended_suggestions) 

71 

72 # Try something new/different 

73 new_suggestions = _generate_new_suggestions(profile) 

74 suggestions.extend(new_suggestions) 

75 

76 if not suggestions: 

77 # If no suggestions generated, return empty 

78 return { 

79 'suggestions': [], 

80 'refreshed_at': timezone.now().isoformat(), 

81 } 

82 

83 return _format_suggestions(suggestions) 

84 

85 

86def _format_suggestions(suggestions) -> dict: 

87 """Format suggestions for API response.""" 

88 result = [] 

89 

90 for suggestion in suggestions: 

91 result.append({ 

92 'type': suggestion.suggestion_type, 

93 'title': suggestion.title, 

94 'description': suggestion.description, 

95 'search_query': suggestion.search_query, 

96 }) 

97 

98 # Get the most recent created_at for refreshed_at 

99 if hasattr(suggestions, 'first'): 

100 # QuerySet 

101 first = suggestions.first() 

102 refreshed_at = first.created_at.isoformat() if first else timezone.now().isoformat() 

103 else: 

104 # List 

105 refreshed_at = suggestions[0].created_at.isoformat() if suggestions else timezone.now().isoformat() 

106 

107 return { 

108 'suggestions': result, 

109 'refreshed_at': refreshed_at, 

110 } 

111 

112 

113def _generate_seasonal_suggestions(profile: Profile) -> List[AIDiscoverySuggestion]: 

114 """Generate seasonal/holiday suggestions.""" 

115 try: 

116 prompt = AIPrompt.get_prompt('discover_seasonal') 

117 except AIPrompt.DoesNotExist: 

118 logger.warning('discover_seasonal prompt not found') 

119 return [] 

120 

121 # Get current date info 

122 now = datetime.now() 

123 date_str = now.strftime('%B %d, %Y') # e.g., "January 15, 2024" 

124 season = _get_season(now) 

125 

126 user_prompt = prompt.format_user_prompt( 

127 date=date_str, 

128 season=season, 

129 ) 

130 

131 try: 

132 service = OpenRouterService() 

133 response = service.complete( 

134 system_prompt=prompt.system_prompt, 

135 user_prompt=user_prompt, 

136 model=prompt.model, 

137 json_response=True, 

138 ) 

139 

140 validator = AIResponseValidator() 

141 validated = validator.validate('discover_seasonal', response) 

142 

143 # Create and save suggestions 

144 suggestions = [] 

145 for item in validated: 

146 suggestion = AIDiscoverySuggestion.objects.create( 

147 profile=profile, 

148 suggestion_type='seasonal', 

149 search_query=item['search_query'], 

150 title=item['title'], 

151 description=item['description'], 

152 ) 

153 suggestions.append(suggestion) 

154 

155 logger.info(f'Generated {len(suggestions)} seasonal suggestions for profile {profile.id}') 

156 return suggestions 

157 

158 except (AIUnavailableError, AIResponseError, ValidationError) as e: 

159 logger.warning(f'Failed to generate seasonal suggestions: {e}') 

160 return [] 

161 

162 

163def _generate_recommended_suggestions(profile: Profile) -> List[AIDiscoverySuggestion]: 

164 """Generate suggestions based on user's viewing history.""" 

165 try: 

166 prompt = AIPrompt.get_prompt('discover_favorites') 

167 except AIPrompt.DoesNotExist: 

168 logger.warning('discover_favorites prompt not found') 

169 return [] 

170 

171 # Get user's recently viewed recipes 

172 history = RecipeViewHistory.objects.filter(profile=profile).select_related('recipe').order_by('-viewed_at')[:15] 

173 if not history: 

174 return [] 

175 

176 # Format history list 

177 history_list = '\n'.join( 

178 f'- {item.recipe.title} ({item.recipe.cuisine or item.recipe.category or "uncategorized"})' 

179 for item in history 

180 ) 

181 

182 user_prompt = prompt.format_user_prompt(favorites=history_list) 

183 

184 try: 

185 service = OpenRouterService() 

186 response = service.complete( 

187 system_prompt=prompt.system_prompt, 

188 user_prompt=user_prompt, 

189 model=prompt.model, 

190 json_response=True, 

191 ) 

192 

193 validator = AIResponseValidator() 

194 validated = validator.validate('discover_favorites', response) 

195 

196 # Create and save suggestions 

197 suggestions = [] 

198 for item in validated: 

199 suggestion = AIDiscoverySuggestion.objects.create( 

200 profile=profile, 

201 suggestion_type='favorites', # Keep type for frontend compatibility 

202 search_query=item['search_query'], 

203 title=item['title'], 

204 description=item['description'], 

205 ) 

206 suggestions.append(suggestion) 

207 

208 logger.info(f'Generated {len(suggestions)} recommended suggestions for profile {profile.id}') 

209 return suggestions 

210 

211 except (AIUnavailableError, AIResponseError, ValidationError) as e: 

212 logger.warning(f'Failed to generate recommended suggestions: {e}') 

213 return [] 

214 

215 

216def _generate_new_suggestions(profile: Profile) -> List[AIDiscoverySuggestion]: 

217 """Generate suggestions for trying something new.""" 

218 try: 

219 prompt = AIPrompt.get_prompt('discover_new') 

220 except AIPrompt.DoesNotExist: 

221 logger.warning('discover_new prompt not found') 

222 return [] 

223 

224 # Get user's recent recipes from history to understand their preferences 

225 history = RecipeViewHistory.objects.filter(profile=profile).select_related('recipe').order_by('-viewed_at')[:15] 

226 if not history: 

227 return [] 

228 

229 recent_recipes = '\n'.join( 

230 f'- {item.recipe.title} ({item.recipe.cuisine or "unknown cuisine"}, {item.recipe.category or "unknown category"})' 

231 for item in history 

232 ) 

233 

234 user_prompt = prompt.format_user_prompt(history=recent_recipes) 

235 

236 try: 

237 service = OpenRouterService() 

238 response = service.complete( 

239 system_prompt=prompt.system_prompt, 

240 user_prompt=user_prompt, 

241 model=prompt.model, 

242 json_response=True, 

243 ) 

244 

245 validator = AIResponseValidator() 

246 validated = validator.validate('discover_new', response) 

247 

248 # Create and save suggestions 

249 suggestions = [] 

250 for item in validated: 

251 suggestion = AIDiscoverySuggestion.objects.create( 

252 profile=profile, 

253 suggestion_type='new', 

254 search_query=item['search_query'], 

255 title=item['title'], 

256 description=item['description'], 

257 ) 

258 suggestions.append(suggestion) 

259 

260 logger.info(f'Generated {len(suggestions)} try-new suggestions for profile {profile.id}') 

261 return suggestions 

262 

263 except (AIUnavailableError, AIResponseError, ValidationError) as e: 

264 logger.warning(f'Failed to generate try-new suggestions: {e}') 

265 return [] 

266 

267 

268def _get_season(dt: datetime) -> str: 

269 """Get the season for a given date (Northern Hemisphere).""" 

270 month = dt.month 

271 if month in (12, 1, 2): 

272 return 'winter' 

273 elif month in (3, 4, 5): 

274 return 'spring' 

275 elif month in (6, 7, 8): 

276 return 'summer' 

277 else: 

278 return 'autumn'