Coverage for src/local_deep_research/news/core/news_analyzer.py: 99%
165 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2News analyzer that produces modular output components.
3Breaks down news analysis into separate, reusable pieces.
4"""
6from typing import List, Dict, Any, Optional, cast
7from datetime import datetime, timezone, UTC
8from loguru import logger
10from .utils import generate_card_id
11from ..utils.topic_generator import generate_topics
12from ...config.llm_config import get_llm
13from ...utilities.json_utils import extract_json, get_llm_response_text
16class NewsAnalyzer:
17 """
18 Analyzes news search results to produce modular components.
20 Instead of one big analysis, produces:
21 - News items table
22 - Big picture summary
23 - Watch for (next 24-48h)
24 - Pattern recognition
25 - Extractable topics for subscriptions
26 """
28 def __init__(
29 self,
30 llm_client: Optional[Any] = None,
31 settings_snapshot: Optional[Dict[str, Any]] = None,
32 ):
33 """
34 Initialize the news analyzer.
36 Args:
37 llm_client: LLM client for analysis
38 settings_snapshot: Optional saved settings snapshot so the
39 LLM is constructed under the user's egress policy. The
40 news scheduler runs in background threads without a
41 live request context, so without an explicit snapshot
42 the PEP guard at the top of ``get_llm`` skips and a
43 cloud LLM can fire under require_local_endpoint.
44 """
45 self._owns_llm = llm_client is None
46 self.llm_client = llm_client or get_llm(
47 settings_snapshot=settings_snapshot
48 )
50 def close(self) -> None:
51 """Close the LLM client if this instance created it."""
52 from ...utilities.resource_utils import safe_close
54 if self._owns_llm:
55 safe_close(self.llm_client, "news analyzer LLM")
57 def analyze_news(
58 self, search_results: List[Dict[str, Any]]
59 ) -> Dict[str, Any]:
60 """
61 Analyze news search results into modular components.
63 Args:
64 search_results: Raw search results
66 Returns:
67 Dictionary with modular analysis components
68 """
69 if not search_results:
70 return self._empty_analysis()
72 try:
73 # Step 1: Extract news items table
74 logger.debug("Extracting news items")
75 news_items = self.extract_news_items(search_results)
77 # Step 2: Generate overview components (separate LLM calls for modularity)
78 logger.debug("Generating analysis components")
79 components = {
80 "items": news_items,
81 "item_count": len(news_items),
82 "search_result_count": len(search_results),
83 "timestamp": datetime.now(timezone.utc).isoformat(),
84 }
86 if news_items:
87 # Each component is generated independently
88 components["big_picture"] = self.generate_big_picture(
89 news_items
90 )
91 components["watch_for"] = self.generate_watch_for(news_items)
92 components["patterns"] = self.generate_patterns(news_items)
93 components["topics"] = self.extract_topics(news_items)
94 components["categories"] = self._count_categories(news_items)
95 components["impact_summary"] = self._summarize_impact(
96 news_items
97 )
99 topics_list = cast(List[Any], components.get("topics", []))
100 logger.info(
101 f"News analysis complete: {len(news_items)} items, {len(topics_list)} topics"
102 )
103 return components
105 except Exception:
106 logger.exception("Error analyzing news")
107 return self._empty_analysis()
109 def extract_news_items(
110 self, search_results: List[Dict[str, Any]], max_items: int = 10
111 ) -> List[Dict[str, Any]]:
112 """
113 Extract structured news items from search results.
115 Args:
116 search_results: Raw search results
117 max_items: Maximum number of items to extract
119 Returns:
120 List of structured news items
121 """
122 if not self.llm_client:
123 logger.warning("No LLM client available for news extraction")
124 return []
126 # Prepare search results for LLM
127 snippets = self._prepare_snippets(
128 search_results # Use all results, let LLM handle token limits
129 )
131 prompt = f"""
132Extract up to {max_items} important news stories from these search results.
133Today's date: {datetime.now(UTC).strftime("%B %d, %Y")}
135{snippets}
137For each news story, extract:
1381. headline - 8 words max describing the story
1392. category - A descriptive category for this news (be specific, not limited to generic categories)
1403. summary - 3 clear sentences about what happened
1414. impact_score - 1-10 based on significance
1425. source_url - URL from the search results
1436. entities - people, places, organizations mentioned
1447. is_developing - true/false if story is still developing
1458. time_ago - when it happened (2 hours ago, yesterday, etc)
147Return as JSON array of news items.
148Focus on genuinely newsworthy stories.
149"""
151 try:
152 response = self.llm_client.invoke(prompt)
153 content = get_llm_response_text(response)
155 # Parse JSON response
156 news_items = extract_json(content, expected_type=list)
157 if news_items is not None:
158 # Validate and clean items
159 valid_items = []
160 for item in news_items[:max_items]:
161 if self._validate_news_item(item):
162 # Generate ID
163 item["id"] = generate_card_id()
164 valid_items.append(item)
166 return valid_items
168 except Exception:
169 logger.exception("Error extracting news items")
171 return []
173 def generate_big_picture(self, news_items: List[Dict[str, Any]]) -> str:
174 """
175 Generate the big picture summary of how events connect.
177 Args:
178 news_items: Extracted news items
180 Returns:
181 Big picture summary (3-4 sentences)
182 """
183 if not self.llm_client or not news_items:
184 return ""
186 # Prepare news summaries
187 summaries = "\n".join(
188 [
189 f"- {item['headline']}: {item.get('summary', '')[:100]}..."
190 for item in news_items[:10]
191 ]
192 )
194 prompt = f"""
195Based on these news stories, write THE BIG PICTURE summary.
196Connect the dots between events. What's the larger narrative?
197Write 3-4 sentences maximum.
199News stories:
200{summaries}
202THE BIG PICTURE:"""
204 try:
205 response = self.llm_client.invoke(prompt)
206 content = get_llm_response_text(response)
207 return content.strip()
208 except Exception:
209 logger.exception("Error generating big picture")
210 return ""
212 def generate_watch_for(self, news_items: List[Dict[str, Any]]) -> List[str]:
213 """
214 Generate list of developments to watch for in next 24-48 hours.
216 Args:
217 news_items: Extracted news items
219 Returns:
220 List of bullet points
221 """
222 if not self.llm_client or not news_items:
223 return []
225 # Focus on developing stories
226 developing = [
227 item for item in news_items if item.get("is_developing", False)
228 ]
229 if not developing:
230 developing = news_items[:5]
232 summaries = "\n".join(
233 [
234 f"- {item['headline']}: {item.get('summary', '')[:100]}..."
235 for item in developing
236 ]
237 )
239 prompt = f"""
240Based on these developing news stories, what should we watch for in the next 24-48 hours?
241Write 3-5 specific, actionable items.
243Developing stories:
244{summaries}
246WATCH FOR:
247-"""
249 try:
250 response = self.llm_client.invoke(prompt)
251 content = get_llm_response_text(response)
253 # Parse bullet points
254 lines = content.strip().split("\n")
255 watch_items = []
256 for line in lines:
257 line = line.strip()
258 if line and line not in ["WATCH FOR:", "Watch for:"]:
259 # Remove bullet markers
260 line = line.lstrip("-•* ")
261 if line: 261 ↛ 256line 261 didn't jump to line 256 because the condition on line 261 was always true
262 watch_items.append(line)
264 return watch_items[:5]
266 except Exception:
267 logger.exception("Error generating watch items")
268 return []
270 def generate_patterns(self, news_items: List[Dict[str, Any]]) -> str:
271 """
272 Identify emerging patterns from today's news.
274 Args:
275 news_items: Extracted news items
277 Returns:
278 Pattern recognition summary
279 """
280 if not self.llm_client or not news_items:
281 return ""
283 # Group by category
284 by_category: Dict[str, List[Any]] = {}
285 for item in news_items:
286 cat = item.get("category", "Other")
287 if cat not in by_category:
288 by_category[cat] = []
289 by_category[cat].append(item["headline"])
291 category_summary = "\n".join(
292 [
293 f"{cat}: {len(items)} stories"
294 for cat, items in by_category.items()
295 ]
296 )
298 prompt = f"""
299Identify emerging patterns from today's news distribution:
301{category_summary}
303Top headlines:
304{chr(10).join([f"- {item['headline']}" for item in news_items[:10]])}
306PATTERN RECOGNITION (1-2 sentences):"""
308 try:
309 response = self.llm_client.invoke(prompt)
310 content = get_llm_response_text(response)
311 return content.strip()
312 except Exception:
313 logger.exception("Error generating patterns")
314 return ""
316 def extract_topics(
317 self, news_items: List[Dict[str, Any]]
318 ) -> List[Dict[str, Any]]:
319 """
320 Extract subscribable topics from news items.
322 Args:
323 news_items: Extracted news items
325 Returns:
326 List of topic dictionaries with metadata
327 """
328 topics = []
330 # Use topic generator to extract from each item
331 for item in news_items:
332 # Use topic generator with headline as query and summary as findings
333 headline = item.get("headline", "")
334 summary = item.get("summary", "")
335 category = item.get("category", "")
337 extracted = generate_topics(
338 query=headline,
339 findings=summary,
340 category=category,
341 max_topics=3,
342 )
344 for topic in extracted:
345 topics.append(
346 {
347 "name": topic,
348 "source_item_id": item.get("id"),
349 "source_headline": item.get("headline"),
350 "category": item.get("category"),
351 "impact_score": item.get("impact_score", 5),
352 }
353 )
355 # Deduplicate and sort by frequency
356 topic_counts = {}
357 topic_metadata = {}
359 for topic_info in topics:
360 name = topic_info["name"]
361 if name not in topic_counts:
362 topic_counts[name] = 0
363 topic_metadata[name] = topic_info
364 topic_counts[name] += 1
366 # Keep highest impact score
367 if (
368 topic_info["impact_score"]
369 > topic_metadata[name]["impact_score"]
370 ):
371 topic_metadata[name] = topic_info
373 # Create final topic list
374 final_topics = []
375 for topic, count in sorted(
376 topic_counts.items(), key=lambda x: x[1], reverse=True
377 ):
378 metadata = topic_metadata[topic]
379 metadata["frequency"] = count
380 metadata["query"] = f"{topic} latest developments news"
381 final_topics.append(metadata)
383 return final_topics[:10] # Top 10 topics
385 def _prepare_snippets(self, search_results: List[Dict[str, Any]]) -> str:
386 """Prepare search result snippets for LLM processing."""
387 snippets = []
388 for i, result in enumerate(search_results):
389 snippet = f"[{i + 1}] "
390 if result.get("title"):
391 snippet += f"Title: {result['title']}\n"
392 if result.get("url"):
393 snippet += f"URL: {result['url']}\n"
394 if result.get("snippet"):
395 snippet += f"Snippet: {result['snippet'][:200]}...\n"
396 elif result.get("content"):
397 snippet += f"Content: {result['content'][:200]}...\n"
399 snippets.append(snippet)
401 return "\n".join(snippets)
403 def _validate_news_item(self, item: Dict[str, Any]) -> bool:
404 """Validate that a news item has required fields."""
405 required = ["headline", "summary"]
406 return all(field in item and item[field] for field in required)
408 def _count_categories(
409 self, news_items: List[Dict[str, Any]]
410 ) -> Dict[str, int]:
411 """Count items by category."""
412 counts: Dict[str, int] = {}
413 for item in news_items:
414 cat = item.get("category", "Other")
415 counts[cat] = counts.get(cat, 0) + 1
416 return counts
418 def _summarize_impact(
419 self, news_items: List[Dict[str, Any]]
420 ) -> Dict[str, Any]:
421 """Summarize impact scores."""
422 if not news_items:
423 return {"average": 0, "high_impact_count": 0}
425 scores = [item.get("impact_score", 5) for item in news_items]
426 return {
427 "average": sum(scores) / len(scores),
428 "high_impact_count": len([s for s in scores if s >= 8]),
429 "max": max(scores),
430 "min": min(scores),
431 }
433 def _empty_analysis(self) -> Dict[str, Any]:
434 """Return empty analysis structure."""
435 return {
436 "items": [],
437 "item_count": 0,
438 "big_picture": "",
439 "watch_for": [],
440 "patterns": "",
441 "topics": [],
442 "categories": {},
443 "impact_summary": {"average": 0, "high_impact_count": 0},
444 "timestamp": datetime.now(timezone.utc).isoformat(),
445 }