Coverage for src/local_deep_research/news/recommender/topic_based.py: 97%
115 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +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
13from ...security import sanitize_for_log
14from ...security.egress.policy import PolicyDeniedError
17class TopicBasedRecommender(BaseRecommender):
18 """
19 Recommends news based on topics extracted from recent news analysis.
21 This recommender:
22 1. Gets recent news topics from the topic registry
23 2. Filters based on user preferences
24 3. Generates search queries for interesting topics
25 4. Creates NewsCards from the results
26 """
28 def __init__(self, **kwargs):
29 """Initialize the topic-based recommender."""
30 super().__init__(**kwargs)
31 self.max_recommendations = 5 # Default limit
33 def generate_recommendations(
34 self, user_id: str, context: Optional[Dict[str, Any]] = None
35 ) -> List[NewsCard]:
36 """
37 Generate recommendations based on trending topics.
39 Args:
40 user_id: User to generate recommendations for
41 context: Optional context like current news being viewed
43 Returns:
44 List of NewsCard recommendations
45 """
46 logger.info(
47 f"Generating topic-based recommendations for user {user_id}"
48 )
50 recommendations = []
52 try:
53 # Update progress
54 self._update_progress("Getting trending topics", 10)
56 # Get trending topics
57 trending_topics = self._get_trending_topics(context)
59 # Filter by user preferences
60 self._update_progress("Applying user preferences", 30)
61 preferences = self._get_user_preferences(user_id)
62 filtered_topics = self._filter_topics_by_preferences(
63 trending_topics, preferences
64 )
66 # Generate recommendations for top topics
67 self._update_progress("Generating news searches", 50)
69 for i, topic in enumerate(
70 filtered_topics[: self.max_recommendations]
71 ):
72 progress = 50 + (
73 40 * i / len(filtered_topics[: self.max_recommendations])
74 )
75 self._update_progress(f"Searching for: {topic}", int(progress))
77 # Create search query
78 query = self._generate_topic_query(topic)
80 # Register with priority manager
81 try:
82 # Execute search
83 card = self._create_recommendation_card(
84 topic, query, user_id
85 )
86 if card:
87 recommendations.append(card)
89 except Exception:
90 logger.exception(
91 f"Error creating recommendation for topic '{sanitize_for_log(topic)}'"
92 )
93 continue
95 self._update_progress("Recommendations complete", 100)
97 # Sort by relevance
98 recommendations = self._sort_by_relevance(recommendations, user_id)
100 logger.info(
101 f"Generated {len(recommendations)} recommendations for user {user_id}"
102 )
104 except Exception as e:
105 logger.exception("Error generating recommendations")
106 self._update_progress(f"Error: {str(e)}", 100)
108 return recommendations
110 def _get_trending_topics(
111 self, context: Optional[Dict[str, Any]]
112 ) -> List[str]:
113 """
114 Get trending topics to recommend.
116 Args:
117 context: Optional context
119 Returns:
120 List of trending topic strings
121 """
122 topics = []
124 # Get from topic registry if available
125 if self.topic_registry:
126 topics.extend(
127 self.topic_registry.get_trending_topics(hours=24, limit=20)
128 )
130 # Add context-based topics if provided
131 if context:
132 if "current_news_topics" in context:
133 topics.extend(context["current_news_topics"])
134 if "current_category" in context:
135 # Could fetch related topics based on category
136 pass
138 # Fallback topics if none found
139 if not topics:
140 logger.warning("No trending topics found, using defaults")
141 topics = [
142 "artificial intelligence developments",
143 "cybersecurity threats",
144 "climate change",
145 "economic policy",
146 "technology innovation",
147 ]
149 return topics
151 def _filter_topics_by_preferences(
152 self, topics: List[str], preferences: Dict[str, Any]
153 ) -> List[str]:
154 """
155 Filter topics based on user preferences.
157 Args:
158 topics: List of topics to filter
159 preferences: User preferences
161 Returns:
162 Filtered list of topics
163 """
164 filtered = []
166 # Get preference lists
167 disliked_topics = [
168 t.lower() for t in preferences.get("disliked_topics", [])
169 ]
170 interests = preferences.get("interests", {})
172 for topic in topics:
173 topic_lower = topic.lower()
175 # Skip disliked topics
176 if any(disliked in topic_lower for disliked in disliked_topics):
177 continue
179 # Boost topics matching interests
180 boost = 1.0
181 for interest, weight in interests.items():
182 if interest.lower() in topic_lower:
183 boost = weight
184 break
186 # Add with boost information
187 filtered.append((topic, boost))
189 # Sort by boost (highest first)
190 filtered.sort(key=lambda x: x[1], reverse=True)
192 # Return just the topics
193 return [topic for topic, _ in filtered]
195 def _generate_topic_query(self, topic: str) -> str:
196 """
197 Generate a search query for a topic.
199 Args:
200 topic: The topic to search for
202 Returns:
203 Search query string
204 """
205 # Add news-specific context
206 return f"{topic} latest news today breaking developments"
208 def _create_recommendation_card(
209 self, topic: str, query: str, user_id: str
210 ) -> Optional[NewsCard]:
211 """
212 Create a news card from a topic recommendation.
214 Args:
215 topic: The topic
216 query: The search query used
217 user_id: The user ID
219 Returns:
220 NewsCard or None if search fails
221 """
222 llm = None
223 search = None
224 search_system = None
225 try:
226 # Use news search strategy
227 from ...config.llm_config import get_llm
228 from ...config.search_config import get_search
230 try:
231 llm = get_llm()
232 except ValueError:
233 # Configuration not set (e.g. llm.model empty). User issue,
234 # not a runtime fault. Log a single concise warning per
235 # scheduled topic — no stack trace, since this scheduler
236 # runs repeatedly and we don't want to spam the log.
237 # Topic is externally derived (news content); sanitize
238 # before interpolation to prevent log injection.
239 logger.warning(
240 f"Skipping news recommendation for topic "
241 f"'{sanitize_for_log(topic)}': "
242 "LLM not configured. Set llm.model in Settings."
243 )
244 return None
245 except PolicyDeniedError as exc:
246 # The egress policy refused the LLM (no snapshot in this
247 # background path + non-local provider). Recurring
248 # scheduler — log concisely, do not retry.
249 # Decision.reason is a short machine code, safe to log.
250 denial_reason = exc.decision.reason
251 logger.warning(
252 f"Skipping news recommendation for topic "
253 f"'{sanitize_for_log(topic)}': "
254 f"LLM blocked by egress policy ({denial_reason}). "
255 "Use a local provider (ollama/lmstudio/llamacpp) or "
256 "thread settings_snapshot through this caller."
257 )
258 return None
259 search = get_search(llm_instance=llm)
260 search_system = AdvancedSearchSystem(
261 llm=llm, search=search, strategy_name="news"
262 )
264 # Mark as news search to use priority system
265 results = search_system.analyze_topic(query, is_news_search=True)
267 if "error" in results:
268 logger.error(
269 f"Search failed for topic '{sanitize_for_log(topic)}': "
270 f"{sanitize_for_log(str(results['error']))}"
271 )
272 return None
274 # Check if we have news items directly from the search
275 news_items = results.get("news_items", [])
277 # Use the news items from search results
278 news_data = {
279 "items": news_items,
280 "item_count": len(news_items),
281 "big_picture": results.get("formatted_findings", ""),
282 "topics": [],
283 }
285 if not news_items:
286 logger.warning(
287 f"No news items found for topic '{sanitize_for_log(topic)}'"
288 )
289 return None
291 # Create card using factory
292 # Use the most impactful news item as the main content
293 main_item = max(news_items, key=lambda x: x.get("impact_score", 0))
295 card = CardFactory.create_news_card_from_analysis(
296 news_item=main_item,
297 source_search_id=str(results.get("search_id") or ""),
298 user_id=user_id,
299 additional_metadata={
300 "recommender": self.strategy_name,
301 "original_topic": topic,
302 "query_used": query,
303 "total_items_found": len(news_items),
304 "big_picture": news_data.get("big_picture", ""),
305 "topics_extracted": news_data.get("topics", []),
306 },
307 )
309 # Add the full analysis as the first version
310 if card: 310 ↛ 322line 310 didn't jump to line 322 because the condition on line 310 was always true
311 card.add_version(
312 research_results={
313 "search_results": results,
314 "news_analysis": news_data,
315 "query": query,
316 "strategy": "news_aggregation",
317 },
318 query=query,
319 strategy="news_aggregation",
320 )
322 return card
324 except Exception:
325 logger.exception(
326 f"Error creating recommendation card for topic "
327 f"'{sanitize_for_log(topic)}'"
328 )
329 return None
330 finally:
331 from ...utilities.resource_utils import safe_close
333 safe_close(search_system, "news search system", allow_none=True)
334 safe_close(search, "news search engine", allow_none=True)
335 safe_close(llm, "news LLM", allow_none=True)
338class SearchBasedRecommender(BaseRecommender):
339 """
340 Recommends news based on user's recent searches.
341 Only works if search tracking is enabled.
342 """
344 def generate_recommendations(
345 self, user_id: str, context: Optional[Dict[str, Any]] = None
346 ) -> List[NewsCard]:
347 """
348 Generate recommendations from user's search history.
350 Args:
351 user_id: User to generate recommendations for
352 context: Optional context
354 Returns:
355 List of NewsCard recommendations
356 """
357 logger.info(
358 f"Generating search-based recommendations for user {user_id}"
359 )
361 # This would need access to search history
362 # For now, return empty since search tracking is OFF by default
363 logger.warning(
364 "Search-based recommendations not available - search tracking is disabled"
365 )
366 return []
368 # Future implementation would:
369 # 1. Get user's recent searches
370 # 2. Transform them to news queries
371 # 3. Create recommendation cards