Coverage for src/local_deep_research/news/core/news_analyzer.py: 99%
165 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +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 = (
207 response.content
208 if hasattr(response, "content")
209 else str(response)
210 )
211 return content.strip()
212 except Exception:
213 logger.exception("Error generating big picture")
214 return ""
216 def generate_watch_for(self, news_items: List[Dict[str, Any]]) -> List[str]:
217 """
218 Generate list of developments to watch for in next 24-48 hours.
220 Args:
221 news_items: Extracted news items
223 Returns:
224 List of bullet points
225 """
226 if not self.llm_client or not news_items:
227 return []
229 # Focus on developing stories
230 developing = [
231 item for item in news_items if item.get("is_developing", False)
232 ]
233 if not developing:
234 developing = news_items[:5]
236 summaries = "\n".join(
237 [
238 f"- {item['headline']}: {item.get('summary', '')[:100]}..."
239 for item in developing
240 ]
241 )
243 prompt = f"""
244Based on these developing news stories, what should we watch for in the next 24-48 hours?
245Write 3-5 specific, actionable items.
247Developing stories:
248{summaries}
250WATCH FOR:
251-"""
253 try:
254 response = self.llm_client.invoke(prompt)
255 content = (
256 response.content
257 if hasattr(response, "content")
258 else str(response)
259 )
261 # Parse bullet points
262 lines = content.strip().split("\n")
263 watch_items = []
264 for line in lines:
265 line = line.strip()
266 if line and line not in ["WATCH FOR:", "Watch for:"]:
267 # Remove bullet markers
268 line = line.lstrip("-•* ")
269 if line: 269 ↛ 264line 269 didn't jump to line 264 because the condition on line 269 was always true
270 watch_items.append(line)
272 return watch_items[:5]
274 except Exception:
275 logger.exception("Error generating watch items")
276 return []
278 def generate_patterns(self, news_items: List[Dict[str, Any]]) -> str:
279 """
280 Identify emerging patterns from today's news.
282 Args:
283 news_items: Extracted news items
285 Returns:
286 Pattern recognition summary
287 """
288 if not self.llm_client or not news_items:
289 return ""
291 # Group by category
292 by_category: Dict[str, List[Any]] = {}
293 for item in news_items:
294 cat = item.get("category", "Other")
295 if cat not in by_category:
296 by_category[cat] = []
297 by_category[cat].append(item["headline"])
299 category_summary = "\n".join(
300 [
301 f"{cat}: {len(items)} stories"
302 for cat, items in by_category.items()
303 ]
304 )
306 prompt = f"""
307Identify emerging patterns from today's news distribution:
309{category_summary}
311Top headlines:
312{chr(10).join([f"- {item['headline']}" for item in news_items[:10]])}
314PATTERN RECOGNITION (1-2 sentences):"""
316 try:
317 response = self.llm_client.invoke(prompt)
318 content = (
319 response.content
320 if hasattr(response, "content")
321 else str(response)
322 )
323 return content.strip()
324 except Exception:
325 logger.exception("Error generating patterns")
326 return ""
328 def extract_topics(
329 self, news_items: List[Dict[str, Any]]
330 ) -> List[Dict[str, Any]]:
331 """
332 Extract subscribable topics from news items.
334 Args:
335 news_items: Extracted news items
337 Returns:
338 List of topic dictionaries with metadata
339 """
340 topics = []
342 # Use topic generator to extract from each item
343 for item in news_items:
344 # Use topic generator with headline as query and summary as findings
345 headline = item.get("headline", "")
346 summary = item.get("summary", "")
347 category = item.get("category", "")
349 extracted = generate_topics(
350 query=headline,
351 findings=summary,
352 category=category,
353 max_topics=3,
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 }