Coverage for src/local_deep_research/news/core/news_analyzer.py: 99%
166 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +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_async
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 ``get_llm`` takes its snapshot-less path, which fails
43 closed with ``PolicyDeniedError`` for any non-local
44 provider — the analyzer cannot be constructed at all
45 instead of quietly reaching a cloud LLM.
46 """
47 self._settings_snapshot = settings_snapshot
48 self._owns_llm = llm_client is None
49 self.llm_client = llm_client or get_llm(
50 settings_snapshot=settings_snapshot
51 )
53 def close(self) -> None:
54 """Close the LLM client if this instance created it."""
55 from ...utilities.resource_utils import safe_close
57 if self._owns_llm:
58 safe_close(self.llm_client, "news analyzer LLM")
60 async def analyze_news(
61 self, search_results: List[Dict[str, Any]]
62 ) -> Dict[str, Any]:
63 """
64 Analyze news search results into modular components.
66 Args:
67 search_results: Raw search results
69 Returns:
70 Dictionary with modular analysis components
71 """
72 if not search_results:
73 return self._empty_analysis()
75 try:
76 # Step 1: Extract news items table
77 logger.debug("Extracting news items")
78 news_items = await self.extract_news_items(search_results)
80 # Step 2: Generate overview components (separate LLM calls for modularity)
81 logger.debug("Generating analysis components")
82 components = {
83 "items": news_items,
84 "item_count": len(news_items),
85 "search_result_count": len(search_results),
86 "timestamp": datetime.now(timezone.utc).isoformat(),
87 }
89 if news_items:
90 # Each component is generated independently
91 components["big_picture"] = await self.generate_big_picture(
92 news_items
93 )
94 components["watch_for"] = await self.generate_watch_for(
95 news_items
96 )
97 components["patterns"] = await self.generate_patterns(
98 news_items
99 )
100 components["topics"] = await self.extract_topics(news_items)
101 components["categories"] = self._count_categories(news_items)
102 components["impact_summary"] = self._summarize_impact(
103 news_items
104 )
106 topics_list = cast(List[Any], components.get("topics", []))
107 logger.info(
108 f"News analysis complete: {len(news_items)} items, {len(topics_list)} topics"
109 )
110 return components
112 except Exception:
113 logger.exception("Error analyzing news")
114 return self._empty_analysis()
116 async def extract_news_items(
117 self, search_results: List[Dict[str, Any]], max_items: int = 10
118 ) -> List[Dict[str, Any]]:
119 """
120 Extract structured news items from search results.
122 Args:
123 search_results: Raw search results
124 max_items: Maximum number of items to extract
126 Returns:
127 List of structured news items
128 """
129 if not self.llm_client:
130 logger.warning("No LLM client available for news extraction")
131 return []
133 # Prepare search results for LLM
134 snippets = self._prepare_snippets(
135 search_results # Use all results, let LLM handle token limits
136 )
138 prompt = f"""
139Extract up to {max_items} important news stories from these search results.
140Today's date: {datetime.now(UTC).strftime("%B %d, %Y")}
142{snippets}
144For each news story, extract:
1451. headline - 8 words max describing the story
1462. category - A descriptive category for this news (be specific, not limited to generic categories)
1473. summary - 3 clear sentences about what happened
1484. impact_score - 1-10 based on significance
1495. source_url - URL from the search results
1506. entities - people, places, organizations mentioned
1517. is_developing - true/false if story is still developing
1528. time_ago - when it happened (2 hours ago, yesterday, etc)
154Return as JSON array of news items.
155Focus on genuinely newsworthy stories.
156"""
158 try:
159 response = await self.llm_client.ainvoke(prompt)
160 content = get_llm_response_text(response)
162 # Parse JSON response
163 news_items = extract_json(content, expected_type=list)
164 if news_items is not None:
165 # Validate and clean items
166 valid_items = []
167 for item in news_items[:max_items]:
168 if self._validate_news_item(item):
169 # Generate ID
170 item["id"] = generate_card_id()
171 valid_items.append(item)
173 return valid_items
175 except Exception:
176 logger.exception("Error extracting news items")
178 return []
180 async def generate_big_picture(
181 self, news_items: List[Dict[str, Any]]
182 ) -> str:
183 """
184 Generate the big picture summary of how events connect.
186 Args:
187 news_items: Extracted news items
189 Returns:
190 Big picture summary (3-4 sentences)
191 """
192 if not self.llm_client or not news_items:
193 return ""
195 # Prepare news summaries
196 summaries = "\n".join(
197 [
198 f"- {item['headline']}: {item.get('summary', '')[:100]}..."
199 for item in news_items[:10]
200 ]
201 )
203 prompt = f"""
204Based on these news stories, write THE BIG PICTURE summary.
205Connect the dots between events. What's the larger narrative?
206Write 3-4 sentences maximum.
208News stories:
209{summaries}
211THE BIG PICTURE:"""
213 try:
214 response = await self.llm_client.ainvoke(prompt)
215 content = get_llm_response_text(response)
216 return content.strip()
217 except Exception:
218 logger.exception("Error generating big picture")
219 return ""
221 async def generate_watch_for(
222 self, news_items: List[Dict[str, Any]]
223 ) -> List[str]:
224 """
225 Generate list of developments to watch for in next 24-48 hours.
227 Args:
228 news_items: Extracted news items
230 Returns:
231 List of bullet points
232 """
233 if not self.llm_client or not news_items:
234 return []
236 # Focus on developing stories
237 developing = [
238 item for item in news_items if item.get("is_developing", False)
239 ]
240 if not developing:
241 developing = news_items[:5]
243 summaries = "\n".join(
244 [
245 f"- {item['headline']}: {item.get('summary', '')[:100]}..."
246 for item in developing
247 ]
248 )
250 prompt = f"""
251Based on these developing news stories, what should we watch for in the next 24-48 hours?
252Write 3-5 specific, actionable items.
254Developing stories:
255{summaries}
257WATCH FOR:
258-"""
260 try:
261 response = await self.llm_client.ainvoke(prompt)
262 content = get_llm_response_text(response)
264 # Parse bullet points
265 lines = content.strip().split("\n")
266 watch_items = []
267 for line in lines:
268 line = line.strip()
269 if line and line not in ["WATCH FOR:", "Watch for:"]:
270 # Remove bullet markers
271 line = line.lstrip("-•* ")
272 if line: 272 ↛ 267line 272 didn't jump to line 267 because the condition on line 272 was always true
273 watch_items.append(line)
275 return watch_items[:5]
277 except Exception:
278 logger.exception("Error generating watch items")
279 return []
281 async def generate_patterns(self, news_items: List[Dict[str, Any]]) -> str:
282 """
283 Identify emerging patterns from today's news.
285 Args:
286 news_items: Extracted news items
288 Returns:
289 Pattern recognition summary
290 """
291 if not self.llm_client or not news_items:
292 return ""
294 # Group by category
295 by_category: Dict[str, List[Any]] = {}
296 for item in news_items:
297 cat = item.get("category", "Other")
298 if cat not in by_category:
299 by_category[cat] = []
300 by_category[cat].append(item["headline"])
302 category_summary = "\n".join(
303 [
304 f"{cat}: {len(items)} stories"
305 for cat, items in by_category.items()
306 ]
307 )
309 prompt = f"""
310Identify emerging patterns from today's news distribution:
312{category_summary}
314Top headlines:
315{chr(10).join([f"- {item['headline']}" for item in news_items[:10]])}
317PATTERN RECOGNITION (1-2 sentences):"""
319 try:
320 response = await self.llm_client.ainvoke(prompt)
321 content = get_llm_response_text(response)
322 return content.strip()
323 except Exception:
324 logger.exception("Error generating patterns")
325 return ""
327 async def extract_topics(
328 self, news_items: List[Dict[str, Any]]
329 ) -> List[Dict[str, Any]]:
330 """
331 Extract subscribable topics from news items.
333 Args:
334 news_items: Extracted news items
336 Returns:
337 List of topic dictionaries with metadata
338 """
339 topics = []
341 # Use topic generator to extract from each item
342 for item in news_items:
343 # Use topic generator with headline as query and summary as findings
344 headline = item.get("headline", "")
345 summary = item.get("summary", "")
346 category = item.get("category", "")
348 extracted = await generate_topics_async(
349 query=headline,
350 findings=summary,
351 category=category,
352 max_topics=3,
353 settings_snapshot=self._settings_snapshot,
354 )
356 for topic in extracted:
357 topics.append(
358 {
359 "name": topic,
360 "source_item_id": item.get("id"),
361 "source_headline": item.get("headline"),
362 "category": item.get("category"),
363 "impact_score": item.get("impact_score", 5),
364 }
365 )
367 # Deduplicate and sort by frequency
368 topic_counts = {}
369 topic_metadata = {}
371 for topic_info in topics:
372 name = topic_info["name"]
373 if name not in topic_counts:
374 topic_counts[name] = 0
375 topic_metadata[name] = topic_info
376 topic_counts[name] += 1
378 # Keep highest impact score
379 if (
380 topic_info["impact_score"]
381 > topic_metadata[name]["impact_score"]
382 ):
383 topic_metadata[name] = topic_info
385 # Create final topic list
386 final_topics = []
387 for topic, count in sorted(
388 topic_counts.items(), key=lambda x: x[1], reverse=True
389 ):
390 metadata = topic_metadata[topic]
391 metadata["frequency"] = count
392 metadata["query"] = f"{topic} latest developments news"
393 final_topics.append(metadata)
395 return final_topics[:10] # Top 10 topics
397 def _prepare_snippets(self, search_results: List[Dict[str, Any]]) -> str:
398 """Prepare search result snippets for LLM processing."""
399 snippets = []
400 for i, result in enumerate(search_results):
401 snippet = f"[{i + 1}] "
402 if result.get("title"):
403 snippet += f"Title: {result['title']}\n"
404 if result.get("url"):
405 snippet += f"URL: {result['url']}\n"
406 if result.get("snippet"):
407 snippet += f"Snippet: {result['snippet'][:200]}...\n"
408 elif result.get("content"):
409 snippet += f"Content: {result['content'][:200]}...\n"
411 snippets.append(snippet)
413 return "\n".join(snippets)
415 def _validate_news_item(self, item: Dict[str, Any]) -> bool:
416 """Validate that a news item has required fields."""
417 required = ["headline", "summary"]
418 return all(field in item and item[field] for field in required)
420 def _count_categories(
421 self, news_items: List[Dict[str, Any]]
422 ) -> Dict[str, int]:
423 """Count items by category."""
424 counts: Dict[str, int] = {}
425 for item in news_items:
426 cat = item.get("category", "Other")
427 counts[cat] = counts.get(cat, 0) + 1
428 return counts
430 def _summarize_impact(
431 self, news_items: List[Dict[str, Any]]
432 ) -> Dict[str, Any]:
433 """Summarize impact scores."""
434 if not news_items:
435 return {"average": 0, "high_impact_count": 0}
437 scores = [item.get("impact_score", 5) for item in news_items]
438 return {
439 "average": sum(scores) / len(scores),
440 "high_impact_count": len([s for s in scores if s >= 8]),
441 "max": max(scores),
442 "min": min(scores),
443 }
445 def _empty_analysis(self) -> Dict[str, Any]:
446 """Return empty analysis structure."""
447 return {
448 "items": [],
449 "item_count": 0,
450 "big_picture": "",
451 "watch_for": [],
452 "patterns": "",
453 "topics": [],
454 "categories": {},
455 "impact_summary": {"average": 0, "high_impact_count": 0},
456 "timestamp": datetime.now(timezone.utc).isoformat(),
457 }