Coverage for apps / ai / services / ranking.py: 25%

1"""AI-powered search result ranking service.""" 

2 

3import logging 

4from typing import Any 

5 

6from apps.core.models import AppSettings 

7 

8from ..models import AIPrompt 

9from .openrouter import OpenRouterService, AIUnavailableError, AIResponseError 

10from .validator import AIResponseValidator, ValidationError 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15def is_ranking_available() -> bool: 

16 """Check if AI ranking is available (API key configured).""" 

17 try: 

18 settings = AppSettings.get() 

19 return bool(settings.openrouter_api_key) 

20 except Exception: 

21 return False 

22 

23 

24def _filter_valid(results: list[dict]) -> list[dict]: 

25 """Filter out results without titles (QA-053). 

26 

27 Results without titles will fail to import, so they should 

28 never be shown to users. 

29 """ 

30 return [r for r in results if r.get('title')] 

31 

32 

33def _sort_by_image(results: list[dict]) -> list[dict]: 

34 """Filter and sort results to prioritize those with images. 

35 

36 This provides a basic image-first sorting when AI ranking 

37 is unavailable or fails. Also filters out invalid results. 

38 """ 

39 valid_results = _filter_valid(results) 

40 return sorted(valid_results, key=lambda r: (0 if r.get('image_url') else 1)) 

41 

42 

43def rank_results(query: str, results: list[dict]) -> list[dict]: 

44 """Rank search results using AI. 

45 

46 The AI considers: 

47 - Relevance to the search query 

48 - Recipe completeness (image, ratings) 

49 - Source reliability 

50 - Title/description clarity 

51 

52 Args: 

53 query: The original search query. 

54 results: List of search result dicts (url, title, host, image_url, description). 

55 

56 Returns: 

57 The same results list, reordered by AI ranking. 

58 Falls back to original order if ranking fails. 

59 

60 Note: 

61 This function is non-blocking and will gracefully fall back 

62 to the original order if AI is unavailable or errors occur. 

63 """ 

64 # Filter out results without titles first (QA-053) 

65 results = _filter_valid(results) 

66 

67 if not results or len(results) <= 1: 

68 return results 

69 

70 # Check if ranking is available 

71 if not is_ranking_available(): 

72 logger.debug('AI ranking skipped: No API key configured, using image-first sorting') 

73 return _sort_by_image(results) 

74 

75 try: 

76 prompt = AIPrompt.get_prompt('search_ranking') 

77 except AIPrompt.DoesNotExist: 

78 logger.warning('search_ranking prompt not found, using image-first sorting') 

79 return _sort_by_image(results) 

80 

81 # Format results for the AI 

82 # Limit to first 40 results to keep prompt size manageable 

83 results_to_rank = results[:40] 

84 remaining = results[40:] if len(results) > 40 else [] 

85 

86 results_text = '\n'.join( 

87 f'{i}. "{r.get("title", "Unknown")}" from {r.get("host", "unknown")} ' 

88 f'- {r.get("description", "No description")[:100]}' 

89 f'{" [has image]" if r.get("image_url") else ""}' 

90 for i, r in enumerate(results_to_rank) 

91 ) 

92 

93 user_prompt = prompt.format_user_prompt( 

94 query=query, 

95 results=results_text, 

96 count=len(results_to_rank), 

97 ) 

98 

99 try: 

100 service = OpenRouterService() 

101 response = service.complete( 

102 system_prompt=prompt.system_prompt, 

103 user_prompt=user_prompt, 

104 model=prompt.model, 

105 json_response=True, 

106 ) 

107 

108 # Validate response - expects array of integers (indices) 

109 validator = AIResponseValidator() 

110 ranking = validator.validate('search_ranking', response) 

111 

112 # Apply ranking 

113 ranked_results = _apply_ranking(results_to_rank, ranking) 

114 

115 # Sort remaining results to prioritize those with images 

116 if remaining: 

117 remaining_sorted = sorted( 

118 remaining, 

119 key=lambda r: (0 if r.get('image_url') else 1), 

120 ) 

121 ranked_results.extend(remaining_sorted) 

122 

123 logger.info(f'AI ranked {len(results_to_rank)} results for query "{query}"') 

124 return ranked_results 

125 

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

127 logger.warning(f'AI ranking failed for query "{query}": {e}, using image-first sorting') 

128 return _sort_by_image(results) 

129 except Exception as e: 

130 logger.error(f'Unexpected error in AI ranking: {e}, using image-first sorting') 

131 return _sort_by_image(results) 

132 

133 

134def _apply_ranking(results: list[dict], ranking: list[int]) -> list[dict]: 

135 """Apply the ranking indices to reorder results. 

136 

137 Args: 

138 results: Original list of results. 

139 ranking: List of indices in desired order. 

140 

141 Returns: 

142 Reordered list of results. 

143 """ 

144 # Validate indices are within bounds 

145 valid_indices = set(range(len(results))) 

146 filtered_ranking = [i for i in ranking if i in valid_indices] 

147 

148 # Build reordered list 

149 ranked = [] 

150 seen = set() 

151 

152 for idx in filtered_ranking: 

153 if idx not in seen: 

154 ranked.append(results[idx]) 

155 seen.add(idx) 

156 

157 # Append any results not included in ranking (shouldn't happen, but safety) 

158 for i, result in enumerate(results): 

159 if i not in seen: 

160 ranked.append(result) 

161 

162 return ranked