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

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 = None 

41 

42 # Strategy name for identification 

43 self.strategy_name = self.__class__.__name__ 

44 

45 def set_progress_callback( 

46 self, callback: Callable[[str, int, dict], None] 

47 ) -> None: 

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

49 self.progress_callback = callback 

50 

51 def _update_progress( 

52 self, 

53 message: str, 

54 progress_percent: Optional[int] = None, 

55 metadata: Optional[dict] = None, 

56 ) -> None: 

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

58 if self.progress_callback: 

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

60 

61 @abstractmethod 

62 def generate_recommendations( 

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

64 ) -> List[NewsCard]: 

65 """ 

66 Generate news recommendations for a user. 

67 

68 Args: 

69 user_id: The user to generate recommendations for 

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

71 

72 Returns: 

73 List of NewsCard objects representing recommendations 

74 """ 

75 pass 

76 

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

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

79 if self.preference_manager: 

80 return self.preference_manager.get_preferences(user_id) 

81 return {} 

82 

83 def _get_user_ratings( 

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

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

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

87 if self.rating_system: 

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

89 return [] 

90 

91 def _execute_search( 

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

93 ) -> Dict[str, Any]: 

94 """ 

95 Execute a search using the LDR search system. 

96 

97 Args: 

98 query: The search query 

99 strategy: Optional search strategy to use 

100 

101 Returns: 

102 Search results dictionary 

103 """ 

104 if not self.search_system: 

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

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

107 

108 try: 

109 # Use news strategy by default if available 

110 if strategy is None: 

111 strategy = "news_aggregation" 

112 

113 # Execute search 

114 results = self.search_system.analyze_topic(query) 

115 return results 

116 

117 except Exception: 

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

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

120 

121 def _filter_by_preferences( 

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

123 ) -> List[NewsCard]: 

124 """ 

125 Filter cards based on user preferences. 

126 

127 Args: 

128 cards: List of news cards to filter 

129 preferences: User preferences dictionary 

130 

131 Returns: 

132 Filtered list of cards 

133 """ 

134 filtered = cards 

135 

136 # Filter by categories if specified 

137 if ( 

138 "liked_categories" in preferences 

139 and preferences["liked_categories"] 

140 ): 

141 # Boost liked categories rather than filtering out others 

142 for card in filtered: 

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

144 card.metadata["preference_boost"] = 1.2 

145 

146 # Filter by impact threshold 

147 if "impact_threshold" in preferences: 

148 threshold = preferences["impact_threshold"] 

149 filtered = [ 

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

151 ] 

152 

153 # Apply disliked topics 

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

155 filtered = [ 

156 card 

157 for card in filtered 

158 if not any( 

159 topic in card.topic.lower() 

160 for topic in preferences["disliked_topics"] 

161 ) 

162 ] 

163 

164 return filtered 

165 

166 def _sort_by_relevance( 

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

168 ) -> List[NewsCard]: 

169 """ 

170 Sort cards by relevance to the user. 

171 Default implementation uses impact score and preference boost. 

172 

173 Args: 

174 cards: List of cards to sort 

175 user_id: User ID for personalization 

176 

177 Returns: 

178 Sorted list of cards 

179 """ 

180 

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

182 # Base score from impact 

183 score = card.impact_score / 10.0 

184 

185 # Apply preference boost if exists 

186 boost = card.metadata.get("preference_boost", 1.0) 

187 score *= boost 

188 

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

190 return score 

191 

192 # Sort by calculated score (highest first) 

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

194 

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

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

197 return { 

198 "name": self.strategy_name, 

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

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

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

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

203 }