Coverage for src / local_deep_research / news / recommender / base_recommender.py: 99%

1""" 

2Base class for all news recommendation strategies. 

3Following LDR's pattern from BaseSearchStrategy. 

4""" 

5 

6from abc import ABC, abstractmethod 

7from typing import List, Dict, Any, Optional, Callable 

8from loguru import logger 

9 

10from ..core.base_card import NewsCard 

11from ..preference_manager.base_preference import BasePreferenceManager 

12from ..rating_system.base_rater import BaseRatingSystem 

13 

14 

15class BaseRecommender(ABC): 

16 """Abstract base class for all recommendation strategies.""" 

17 

18 def __init__( 

19 self, 

20 preference_manager: Optional[BasePreferenceManager] = None, 

21 rating_system: Optional[BaseRatingSystem] = None, 

22 topic_registry: Optional[Any] = None, 

23 search_system: Optional[Any] = None, 

24 ): 

25 """ 

26 Initialize the base recommender with common dependencies. 

27 

28 Args: 

29 preference_manager: Manager for user preferences 

30 rating_system: System for tracking ratings 

31 topic_registry: Registry of discovered topics 

32 search_system: LDR search system for executing queries 

33 """ 

34 self.preference_manager = preference_manager 

35 self.rating_system = rating_system 

36 self.topic_registry = topic_registry 

37 self.search_system = search_system 

38 

39 # Progress tracking (following LDR pattern) 

40 self.progress_callback: Optional[ 

41 Callable[[str, Optional[int], dict], None] 

42 ] = None 

43 

44 # Strategy name for identification 

45 self.strategy_name = self.__class__.__name__ 

46 

47 def set_progress_callback( 

48 self, callback: Callable[[str, Optional[int], dict], None] 

49 ) -> None: 

50 """Set a callback function to receive progress updates.""" 

51 self.progress_callback = callback 

52 

53 def _update_progress( 

54 self, 

55 message: str, 

56 progress_percent: Optional[int] = None, 

57 metadata: Optional[dict] = None, 

58 ) -> None: 

59 """Send a progress update via the callback if available.""" 

60 if self.progress_callback: 

61 self.progress_callback(message, progress_percent, metadata or {}) 

62 

63 @abstractmethod 

64 def generate_recommendations( 

65 self, user_id: str, context: Optional[Dict[str, Any]] = None 

66 ) -> List[NewsCard]: 

67 """ 

68 Generate news recommendations for a user. 

69 

70 Args: 

71 user_id: The user to generate recommendations for 

72 context: Optional context (e.g., current page, recent activity) 

73 

74 Returns: 

75 List of NewsCard objects representing recommendations 

76 """ 

77 pass 

78 

79 def _get_user_preferences(self, user_id: str) -> Dict[str, Any]: 

80 """Get user preferences if preference manager is available.""" 

81 if self.preference_manager: 

82 return self.preference_manager.get_preferences(user_id) 

83 return {} 

84 

85 def _get_user_ratings( 

86 self, user_id: str, limit: int = 50 

87 ) -> List[Dict[str, Any]]: 

88 """Get recent user ratings if rating system is available.""" 

89 if self.rating_system: 

90 return self.rating_system.get_recent_ratings(user_id, limit) 

91 return [] 

92 

93 def _execute_search( 

94 self, query: str, strategy: Optional[str] = None 

95 ) -> Dict[str, Any]: 

96 """ 

97 Execute a search using the LDR search system. 

98 

99 Args: 

100 query: The search query 

101 strategy: Optional search strategy to use 

102 

103 Returns: 

104 Search results dictionary 

105 """ 

106 if not self.search_system: 

107 logger.warning("No search system available for recommendations") 

108 return {"error": "Search system not configured"} 

109 

110 try: 

111 # Use news strategy by default if available 

112 if strategy is None: 

113 strategy = "news_aggregation" 

114 

115 # Execute search 

116 results: Dict[str, Any] = self.search_system.analyze_topic(query) 

117 return results 

118 

119 except Exception: 

120 logger.exception("Error executing search for recommendations") 

121 return {"error": "Recommendation search failed"} 

122 

123 def _filter_by_preferences( 

124 self, cards: List[NewsCard], preferences: Dict[str, Any] 

125 ) -> List[NewsCard]: 

126 """ 

127 Filter cards based on user preferences. 

128 

129 Args: 

130 cards: List of news cards to filter 

131 preferences: User preferences dictionary 

132 

133 Returns: 

134 Filtered list of cards 

135 """ 

136 filtered = cards 

137 

138 # Filter by categories if specified 

139 if ( 

140 "liked_categories" in preferences 

141 and preferences["liked_categories"] 

142 ): 

143 # Boost liked categories rather than filtering out others 

144 for card in filtered: 

145 if card.category in preferences["liked_categories"]: 

146 card.metadata["preference_boost"] = 1.2 

147 

148 # Filter by impact threshold 

149 if "impact_threshold" in preferences: 

150 threshold = preferences["impact_threshold"] 

151 filtered = [ 

152 card for card in filtered if card.impact_score >= threshold 

153 ] 

154 

155 # Apply disliked topics 

156 if "disliked_topics" in preferences and preferences["disliked_topics"]: 

157 filtered = [ 

158 card 

159 for card in filtered 

160 if not any( 

161 topic in card.topic.lower() 

162 for topic in preferences["disliked_topics"] 

163 ) 

164 ] 

165 

166 return filtered 

167 

168 def _sort_by_relevance( 

169 self, cards: List[NewsCard], user_id: str 

170 ) -> List[NewsCard]: 

171 """ 

172 Sort cards by relevance to the user. 

173 Default implementation uses impact score and preference boost. 

174 

175 Args: 

176 cards: List of cards to sort 

177 user_id: User ID for personalization 

178 

179 Returns: 

180 Sorted list of cards 

181 """ 

182 

183 def calculate_score(card: NewsCard) -> float: 

184 # Base score from impact 

185 score: float = card.impact_score / 10.0 

186 

187 # Apply preference boost if exists 

188 boost: float = float(card.metadata.get("preference_boost", 1.0)) 

189 score *= boost 

190 

191 # Could add more factors here (recency, etc.) 

192 return score 

193 

194 # Sort by calculated score (highest first) 

195 return sorted(cards, key=calculate_score, reverse=True) 

196 

197 def get_strategy_info(self) -> Dict[str, Any]: 

198 """Get information about this recommendation strategy.""" 

199 return { 

200 "name": self.strategy_name, 

201 "has_preference_manager": self.preference_manager is not None, 

202 "has_rating_system": self.rating_system is not None, 

203 "has_search_system": self.search_system is not None, 

204 "description": self.__doc__ or "No description available", 

205 }