Coverage for src / local_deep_research / news / recommender / topic_based.py: 12%
94 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1"""
2Topic-based recommender that generates recommendations from news topics.
3This is the primary recommender for v1.
4"""
6from typing import List, Dict, Any, Optional
7from loguru import logger
9from .base_recommender import BaseRecommender
10from ..core.base_card import NewsCard
11from ..core.card_factory import CardFactory
12from ...search_system import AdvancedSearchSystem
15class TopicBasedRecommender(BaseRecommender):
16 """
17 Recommends news based on topics extracted from recent news analysis.
19 This recommender:
20 1. Gets recent news topics from the topic registry
21 2. Filters based on user preferences
22 3. Generates search queries for interesting topics
23 4. Creates NewsCards from the results
24 """
26 def __init__(self, **kwargs):
27 """Initialize the topic-based recommender."""
28 super().__init__(**kwargs)
29 self.max_recommendations = 5 # Default limit
31 def generate_recommendations(
32 self, user_id: str, context: Optional[Dict[str, Any]] = None
33 ) -> List[NewsCard]:
34 """
35 Generate recommendations based on trending topics.
37 Args:
38 user_id: User to generate recommendations for
39 context: Optional context like current news being viewed
41 Returns:
42 List of NewsCard recommendations
43 """
44 logger.info(
45 f"Generating topic-based recommendations for user {user_id}"
46 )
48 recommendations = []
50 try:
51 # Update progress
52 self._update_progress("Getting trending topics", 10)
54 # Get trending topics
55 trending_topics = self._get_trending_topics(context)
57 # Filter by user preferences
58 self._update_progress("Applying user preferences", 30)
59 preferences = self._get_user_preferences(user_id)
60 filtered_topics = self._filter_topics_by_preferences(
61 trending_topics, preferences
62 )
64 # Generate recommendations for top topics
65 self._update_progress("Generating news searches", 50)
67 for i, topic in enumerate(
68 filtered_topics[: self.max_recommendations]
69 ):
70 progress = 50 + (
71 40 * i / len(filtered_topics[: self.max_recommendations])
72 )
73 self._update_progress(f"Searching for: {topic}", int(progress))
75 # Create search query
76 query = self._generate_topic_query(topic)
78 # Register with priority manager
79 try:
80 # Execute search
81 card = self._create_recommendation_card(
82 topic, query, user_id
83 )
84 if card:
85 recommendations.append(card)
87 except Exception as e:
88 logger.exception(
89 f"Error creating recommendation for topic '{topic}': {e}"
90 )
91 continue
93 self._update_progress("Recommendations complete", 100)
95 # Sort by relevance
96 recommendations = self._sort_by_relevance(recommendations, user_id)
98 logger.info(
99 f"Generated {len(recommendations)} recommendations for user {user_id}"
100 )
102 except Exception as e:
103 logger.exception("Error generating recommendations")
104 self._update_progress(f"Error: {str(e)}", 100)
106 return recommendations
108 def _get_trending_topics(
109 self, context: Optional[Dict[str, Any]]
110 ) -> List[str]:
111 """
112 Get trending topics to recommend.
114 Args:
115 context: Optional context
117 Returns:
118 List of trending topic strings
119 """
120 topics = []
122 # Get from topic registry if available
123 if self.topic_registry:
124 topics.extend(
125 self.topic_registry.get_trending_topics(hours=24, limit=20)
126 )
128 # Add context-based topics if provided
129 if context:
130 if "current_news_topics" in context:
131 topics.extend(context["current_news_topics"])
132 if "current_category" in context:
133 # Could fetch related topics based on category
134 pass
136 # Fallback topics if none found
137 if not topics:
138 logger.warning("No trending topics found, using defaults")
139 topics = [
140 "artificial intelligence developments",
141 "cybersecurity threats",
142 "climate change",
143 "economic policy",
144 "technology innovation",
145 ]
147 return topics
149 def _filter_topics_by_preferences(
150 self, topics: List[str], preferences: Dict[str, Any]
151 ) -> List[str]:
152 """
153 Filter topics based on user preferences.
155 Args:
156 topics: List of topics to filter
157 preferences: User preferences
159 Returns:
160 Filtered list of topics
161 """
162 filtered = []
164 # Get preference lists
165 disliked_topics = [
166 t.lower() for t in preferences.get("disliked_topics", [])
167 ]
168 interests = preferences.get("interests", {})
170 for topic in topics:
171 topic_lower = topic.lower()
173 # Skip disliked topics
174 if any(disliked in topic_lower for disliked in disliked_topics):
175 continue
177 # Boost topics matching interests
178 boost = 1.0
179 for interest, weight in interests.items():
180 if interest.lower() in topic_lower:
181 boost = weight
182 break
184 # Add with boost information
185 filtered.append((topic, boost))
187 # Sort by boost (highest first)
188 filtered.sort(key=lambda x: x[1], reverse=True)
190 # Return just the topics
191 return [topic for topic, _ in filtered]
193 def _generate_topic_query(self, topic: str) -> str:
194 """
195 Generate a search query for a topic.
197 Args:
198 topic: The topic to search for
200 Returns:
201 Search query string
202 """
203 # Add news-specific context
204 return f"{topic} latest news today breaking developments"
206 def _create_recommendation_card(
207 self, topic: str, query: str, user_id: str
208 ) -> Optional[NewsCard]:
209 """
210 Create a news card from a topic recommendation.
212 Args:
213 topic: The topic
214 query: The search query used
215 user_id: The user ID
217 Returns:
218 NewsCard or None if search fails
219 """
220 try:
221 # Use news search strategy
222 search_system = AdvancedSearchSystem(strategy_name="news")
224 # Mark as news search to use priority system
225 results = search_system.analyze_topic(query, is_news_search=True)
227 if "error" in results:
228 logger.error(
229 f"Search failed for topic '{topic}': {results['error']}"
230 )
231 return None
233 # Check if we have news items directly from the search
234 news_items = results.get("news_items", [])
236 # Use the news items from search results
237 news_data = {
238 "items": news_items,
239 "item_count": len(news_items),
240 "big_picture": results.get("formatted_findings", ""),
241 "topics": [],
242 }
244 if not news_items:
245 logger.warning(f"No news items found for topic '{topic}'")
246 return None
248 # Create card using factory
249 # Use the most impactful news item as the main content
250 main_item = max(news_items, key=lambda x: x.get("impact_score", 0))
252 card = CardFactory.create_news_card_from_analysis(
253 news_item=main_item,
254 source_search_id=results.get("search_id"),
255 user_id=user_id,
256 additional_metadata={
257 "recommender": self.strategy_name,
258 "original_topic": topic,
259 "query_used": query,
260 "total_items_found": len(news_items),
261 "big_picture": news_data.get("big_picture", ""),
262 "topics_extracted": news_data.get("topics", []),
263 },
264 )
266 # Add the full analysis as the first version
267 if card:
268 card.add_version(
269 research_results={
270 "search_results": results,
271 "news_analysis": news_data,
272 "query": query,
273 "strategy": "news_aggregation",
274 },
275 query=query,
276 strategy="news_aggregation",
277 )
279 return card
281 except Exception as e:
282 logger.exception(
283 f"Error creating recommendation card for topic '{topic}': {e}"
284 )
285 return None
288class SearchBasedRecommender(BaseRecommender):
289 """
290 Recommends news based on user's recent searches.
291 Only works if search tracking is enabled.
292 """
294 def generate_recommendations(
295 self, user_id: str, context: Optional[Dict[str, Any]] = None
296 ) -> List[NewsCard]:
297 """
298 Generate recommendations from user's search history.
300 Args:
301 user_id: User to generate recommendations for
302 context: Optional context
304 Returns:
305 List of NewsCard recommendations
306 """
307 logger.info(
308 f"Generating search-based recommendations for user {user_id}"
309 )
311 # This would need access to search history
312 # For now, return empty since search tracking is OFF by default
313 logger.warning(
314 "Search-based recommendations not available - search tracking is disabled"
315 )
316 return []
318 # Future implementation would:
319 # 1. Get user's recent searches
320 # 2. Transform them to news queries
321 # 3. Create recommendation cards