Coverage for src/local_deep_research/news/recommender/topic_based.py: 99%
109 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-03 23:15 +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 try:
229 llm = get_llm()
230 except ValueError:
231 # Configuration not set (e.g. llm.model empty). User issue,
232 # not a runtime fault. Log a single concise warning per
233 # scheduled topic — no stack trace, since this scheduler
234 # runs repeatedly and we don't want to spam the log.
235 logger.warning(
236 f"Skipping news recommendation for topic '{topic}': "
237 "LLM not configured. Set llm.model in Settings."
238 )
239 return None
240 search = get_search(llm_instance=llm)
241 search_system = AdvancedSearchSystem(
242 llm=llm, search=search, strategy_name="news"
243 )
245 # Mark as news search to use priority system
246 results = search_system.analyze_topic(query, is_news_search=True)
248 if "error" in results:
249 logger.error(
250 f"Search failed for topic '{topic}': {results['error']}"
251 )
252 return None
254 # Check if we have news items directly from the search
255 news_items = results.get("news_items", [])
257 # Use the news items from search results
258 news_data = {
259 "items": news_items,
260 "item_count": len(news_items),
261 "big_picture": results.get("formatted_findings", ""),
262 "topics": [],
263 }
265 if not news_items:
266 logger.warning(f"No news items found for topic '{topic}'")
267 return None
269 # Create card using factory
270 # Use the most impactful news item as the main content
271 main_item = max(news_items, key=lambda x: x.get("impact_score", 0))
273 card = CardFactory.create_news_card_from_analysis(
274 news_item=main_item,
275 source_search_id=str(results.get("search_id") or ""),
276 user_id=user_id,
277 additional_metadata={
278 "recommender": self.strategy_name,
279 "original_topic": topic,
280 "query_used": query,
281 "total_items_found": len(news_items),
282 "big_picture": news_data.get("big_picture", ""),
283 "topics_extracted": news_data.get("topics", []),
284 },
285 )
287 # Add the full analysis as the first version
288 if card: 288 ↛ 300line 288 didn't jump to line 300 because the condition on line 288 was always true
289 card.add_version(
290 research_results={
291 "search_results": results,
292 "news_analysis": news_data,
293 "query": query,
294 "strategy": "news_aggregation",
295 },
296 query=query,
297 strategy="news_aggregation",
298 )
300 return card
302 except Exception:
303 logger.exception(
304 f"Error creating recommendation card for topic '{topic}'"
305 )
306 return None
307 finally:
308 from ...utilities.resource_utils import safe_close
310 safe_close(search_system, "news search system", allow_none=True)
311 safe_close(search, "news search engine", allow_none=True)
312 safe_close(llm, "news LLM", allow_none=True)
315class SearchBasedRecommender(BaseRecommender):
316 """
317 Recommends news based on user's recent searches.
318 Only works if search tracking is enabled.
319 """
321 def generate_recommendations(
322 self, user_id: str, context: Optional[Dict[str, Any]] = None
323 ) -> List[NewsCard]:
324 """
325 Generate recommendations from user's search history.
327 Args:
328 user_id: User to generate recommendations for
329 context: Optional context
331 Returns:
332 List of NewsCard recommendations
333 """
334 logger.info(
335 f"Generating search-based recommendations for user {user_id}"
336 )
338 # This would need access to search history
339 # For now, return empty since search tracking is OFF by default
340 logger.warning(
341 "Search-based recommendations not available - search tracking is disabled"
342 )
343 return []
345 # Future implementation would:
346 # 1. Get user's recent searches
347 # 2. Transform them to news queries
348 # 3. Create recommendation cards