Coverage for src / local_deep_research / news / recommender / topic_based.py: 99%
105 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:55 +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:
88 logger.exception(
89 f"Error creating recommendation for topic '{topic}'"
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 llm = None
221 search = None
222 search_system = None
223 try:
224 # Use news search strategy
225 from ...config.llm_config import get_llm
226 from ...config.search_config import get_search
228 llm = get_llm()
229 search = get_search(llm_instance=llm)
230 search_system = AdvancedSearchSystem(
231 llm=llm, search=search, strategy_name="news"
232 )
234 # Mark as news search to use priority system
235 results = search_system.analyze_topic(query, is_news_search=True)
237 if "error" in results:
238 logger.error(
239 f"Search failed for topic '{topic}': {results['error']}"
240 )
241 return None
243 # Check if we have news items directly from the search
244 news_items = results.get("news_items", [])
246 # Use the news items from search results
247 news_data = {
248 "items": news_items,
249 "item_count": len(news_items),
250 "big_picture": results.get("formatted_findings", ""),
251 "topics": [],
252 }
254 if not news_items:
255 logger.warning(f"No news items found for topic '{topic}'")
256 return None
258 # Create card using factory
259 # Use the most impactful news item as the main content
260 main_item = max(news_items, key=lambda x: x.get("impact_score", 0))
262 card = CardFactory.create_news_card_from_analysis(
263 news_item=main_item,
264 source_search_id=str(results.get("search_id") or ""),
265 user_id=user_id,
266 additional_metadata={
267 "recommender": self.strategy_name,
268 "original_topic": topic,
269 "query_used": query,
270 "total_items_found": len(news_items),
271 "big_picture": news_data.get("big_picture", ""),
272 "topics_extracted": news_data.get("topics", []),
273 },
274 )
276 # Add the full analysis as the first version
277 if card: 277 ↛ 289line 277 didn't jump to line 289 because the condition on line 277 was always true
278 card.add_version(
279 research_results={
280 "search_results": results,
281 "news_analysis": news_data,
282 "query": query,
283 "strategy": "news_aggregation",
284 },
285 query=query,
286 strategy="news_aggregation",
287 )
289 return card
291 except Exception:
292 logger.exception(
293 f"Error creating recommendation card for topic '{topic}'"
294 )
295 return None
296 finally:
297 from ...utilities.resource_utils import safe_close
299 safe_close(search_system, "news search system", allow_none=True)
300 safe_close(search, "news search engine", allow_none=True)
301 safe_close(llm, "news LLM", allow_none=True)
304class SearchBasedRecommender(BaseRecommender):
305 """
306 Recommends news based on user's recent searches.
307 Only works if search tracking is enabled.
308 """
310 def generate_recommendations(
311 self, user_id: str, context: Optional[Dict[str, Any]] = None
312 ) -> List[NewsCard]:
313 """
314 Generate recommendations from user's search history.
316 Args:
317 user_id: User to generate recommendations for
318 context: Optional context
320 Returns:
321 List of NewsCard recommendations
322 """
323 logger.info(
324 f"Generating search-based recommendations for user {user_id}"
325 )
327 # This would need access to search history
328 # For now, return empty since search tracking is OFF by default
329 logger.warning(
330 "Search-based recommendations not available - search tracking is disabled"
331 )
332 return []
334 # Future implementation would:
335 # 1. Get user's recent searches
336 # 2. Transform them to news queries
337 # 3. Create recommendation cards