Coverage for src / local_deep_research / news / core / base_card.py: 58%
147 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1"""
2Base card class for all news-related content.
3Following LDR's pattern from BaseSearchStrategy.
4"""
6from abc import ABC, abstractmethod
7from dataclasses import dataclass, field
8from datetime import datetime
9from typing import Dict, List, Optional, Any
11from .utils import generate_card_id, utc_now
12# Storage will be imported when needed to avoid circular import
15@dataclass
16class CardSource:
17 """Tracks the origin of each card."""
19 type: str # "news_item", "user_search", "subscription", "news_research"
20 source_id: Optional[str] = None
21 created_from: str = ""
22 metadata: Dict[str, Any] = field(default_factory=dict)
25@dataclass
26class CardVersion:
27 """Represents a version of research/content for a card."""
29 version_id: str
30 created_at: datetime
31 content: Dict[str, Any] # The actual research results
32 query_used: str
33 search_strategy: Optional[str] = None
35 def __post_init__(self):
36 if not self.version_id:
37 self.version_id = generate_card_id()
40@dataclass
41class BaseCard(ABC):
42 """
43 Abstract base class for all card types.
44 Following LDR's pattern of base classes with common functionality.
45 """
47 # Required fields
48 topic: str
49 source: CardSource
50 user_id: str
52 # Optional fields with defaults
53 card_id: Optional[str] = None
54 parent_card_id: Optional[str] = None
55 metadata: Dict[str, Any] = field(default_factory=dict)
57 # Automatically generated fields
58 id: str = field(init=False)
59 created_at: datetime = field(init=False)
60 updated_at: datetime = field(init=False)
61 versions: List[CardVersion] = field(default_factory=list, init=False)
62 interaction: Dict[str, Any] = field(init=False)
64 # Storage and callback fields
65 storage: Optional[Any] = field(default=None, init=False)
66 progress_callback: Optional[Any] = field(default=None, init=False)
68 def __post_init__(self):
69 """Initialize generated fields after dataclass initialization."""
70 self.id = self.card_id or generate_card_id()
71 self.created_at = utc_now()
72 self.updated_at = utc_now()
73 self.interaction = {
74 "votes_up": 0,
75 "votes_down": 0,
76 "views": 0,
77 "shares": 0,
78 }
80 def set_progress_callback(self, callback) -> None:
81 """Set a callback function to receive progress updates."""
82 self.progress_callback = callback
84 def _update_progress(
85 self,
86 message: str,
87 progress_percent: Optional[int] = None,
88 metadata: Optional[dict] = None,
89 ) -> None:
90 """Send a progress update via the callback if available."""
91 if self.progress_callback:
92 self.progress_callback(message, progress_percent, metadata or {})
94 def save(self) -> str:
95 """Save card to database"""
96 card_data = {
97 "id": self.id,
98 "user_id": self.user_id,
99 "topic": self.topic,
100 "card_type": self.get_card_type(),
101 "source_type": self.source.type,
102 "source_id": self.source.source_id,
103 "created_from": self.source.created_from,
104 "parent_card_id": self.parent_card_id,
105 "metadata": self.metadata,
106 }
107 return self.storage.create(card_data)
109 def add_version(
110 self, research_results: Dict[str, Any], query: str, strategy: str
111 ) -> str:
112 """Add a new version with research results"""
113 version_data = {
114 "search_query": query,
115 "research_result": research_results,
116 "headline": self._extract_headline(research_results),
117 "summary": self._extract_summary(research_results),
118 "findings": research_results.get("findings", []),
119 "sources": research_results.get("sources", []),
120 "impact_score": self._calculate_impact(research_results),
121 "topics": self._extract_topics(research_results),
122 "entities": self._extract_entities(research_results),
123 "strategy": strategy,
124 }
126 version_id = self.storage.add_version(self.id, version_data)
128 # Add to local versions list
129 version = CardVersion(
130 version_id=version_id,
131 created_at=utc_now(),
132 content=research_results,
133 query_used=query,
134 search_strategy=strategy,
135 )
136 self.versions.append(version)
137 self.updated_at = utc_now()
139 return version_id
141 def get_latest_version(self) -> Optional[CardVersion]:
142 """Get the most recent version of this card."""
143 if not self.versions:
144 return None
145 return max(self.versions, key=lambda v: v.created_at)
147 def to_base_dict(self) -> Dict[str, Any]:
148 """Convert base card attributes to dictionary."""
149 return {
150 "id": self.id,
151 "topic": self.topic,
152 "user_id": self.user_id,
153 "created_at": self.created_at.isoformat()
154 if self.created_at
155 else None,
156 "updated_at": self.updated_at.isoformat()
157 if self.updated_at
158 else None,
159 "source": {
160 "type": self.source.type,
161 "source_id": self.source.source_id,
162 "created_from": self.source.created_from,
163 "metadata": self.source.metadata,
164 },
165 "versions_count": len(self.versions),
166 "parent_card_id": self.parent_card_id,
167 "metadata": self.metadata,
168 "interaction": self.interaction,
169 "card_type": self.get_card_type(),
170 }
172 @abstractmethod
173 def get_card_type(self) -> str:
174 """Return the card type (news, research, update, overview)"""
175 pass
177 @abstractmethod
178 def to_dict(self) -> Dict[str, Any]:
179 """
180 Convert card to dictionary representation.
181 Must be implemented by subclasses.
182 """
183 pass
185 # Helper methods for extracting data from research results
186 def _extract_headline(self, result: Dict[str, Any]) -> str:
187 """Extract headline from research result"""
188 # Try different possible fields
189 return (
190 result.get("headline")
191 or result.get("title")
192 or result.get("query", "")[:100]
193 )
195 def _extract_summary(self, result: Dict[str, Any]) -> str:
196 """Extract summary from research result"""
197 return (
198 result.get("summary")
199 or result.get("current_knowledge")
200 or result.get("formatted_findings", "")[:500]
201 )
203 def _calculate_impact(self, result: Dict[str, Any]) -> int:
204 """Calculate impact score (1-10)"""
205 # Simple heuristic based on findings count and sources
206 findings_count = len(result.get("findings", []))
207 sources_count = len(result.get("sources", []))
209 score = min(10, 5 + (findings_count // 5) + (sources_count // 3))
210 return max(1, score)
212 def _extract_topics(self, result: Dict[str, Any]) -> List[str]:
213 """Extract topics from research result"""
214 # Could be enhanced with NLP
215 topics = result.get("topics", [])
216 if not topics and "query" in result:
217 # Simple keyword extraction from query
218 words = result["query"].lower().split()
219 topics = [w for w in words if len(w) > 4][:5]
220 return topics
222 def _extract_entities(self, result: Dict[str, Any]) -> Dict[str, List[str]]:
223 """Extract entities from research result"""
224 # Placeholder - would use NER in production
225 return result.get(
226 "entities", {"people": [], "places": [], "organizations": []}
227 )
230@dataclass
231class NewsCard(BaseCard):
232 """Card representing a news item with potential for research."""
234 # News-specific fields with defaults
235 headline: str = ""
236 summary: str = ""
237 category: str = "General"
238 impact_score: int = 5
239 entities: Dict[str, List[str]] = field(
240 default_factory=lambda: {
241 "people": [],
242 "places": [],
243 "organizations": [],
244 }
245 )
246 topics_extracted: List[str] = field(default_factory=list)
247 is_developing: bool = False
248 time_ago: str = "recent"
249 source_url: str = ""
250 analysis: str = ""
251 surprising_element: Optional[str] = None
253 def __post_init__(self):
254 """Initialize parent and set headline default."""
255 super().__post_init__()
256 if not self.headline:
257 self.headline = self.topic
259 def get_card_type(self) -> str:
260 """Return the card type"""
261 return "news"
263 def to_dict(self) -> Dict[str, Any]:
264 data = self.to_base_dict()
265 data.update(
266 {
267 "headline": self.headline,
268 "summary": self.summary,
269 "category": self.category,
270 "impact_score": self.impact_score,
271 "entities": self.entities,
272 "topics_extracted": self.topics_extracted,
273 "is_developing": self.is_developing,
274 "time_ago": self.time_ago,
275 }
276 )
277 return data
280@dataclass
281class ResearchCard(BaseCard):
282 """Card representing deeper research on a topic."""
284 # Research-specific fields
285 research_depth: str = "quick" # "quick", "detailed", "report"
286 key_findings: List[str] = field(default_factory=list)
287 sources_count: int = 0
289 def get_card_type(self) -> str:
290 """Return the card type"""
291 return "research"
293 def to_dict(self) -> Dict[str, Any]:
294 data = self.to_base_dict()
295 data.update(
296 {
297 "research_depth": self.research_depth,
298 "key_findings": self.key_findings,
299 "sources_count": self.sources_count,
300 }
301 )
302 return data
305@dataclass
306class UpdateCard(BaseCard):
307 """Card representing updates or notifications."""
309 # Update-specific fields
310 update_type: str = "new_stories" # "new_stories", "breaking", "follow_up"
311 count: int = 0
312 preview_items: List[Any] = field(default_factory=list)
313 since: datetime = field(init=False)
315 def __post_init__(self):
316 """Initialize parent and set since timestamp."""
317 super().__post_init__()
318 self.since = utc_now()
320 def get_card_type(self) -> str:
321 """Return the card type"""
322 return "update"
324 def to_dict(self) -> Dict[str, Any]:
325 data = self.to_base_dict()
326 data.update(
327 {
328 "update_type": self.update_type,
329 "count": self.count,
330 "preview_items": self.preview_items,
331 "since": self.since.isoformat(),
332 }
333 )
334 return data
337@dataclass
338class OverviewCard(BaseCard):
339 """Special card type for dashboard/overview display."""
341 # Override topic default for overview cards
342 topic: str = field(default="News Overview", init=False)
344 # Overview-specific fields
345 stats: Dict[str, Any] = field(
346 default_factory=lambda: {
347 "total_new": 0,
348 "breaking": 0,
349 "relevant": 0,
350 "categories": {},
351 }
352 )
353 summary: str = ""
354 top_stories: List[Any] = field(default_factory=list)
355 trend_analysis: str = ""
357 def get_card_type(self) -> str:
358 """Return the card type"""
359 return "overview"
361 def to_dict(self) -> Dict[str, Any]:
362 data = self.to_base_dict()
363 data.update(
364 {
365 "stats": self.stats,
366 "summary": self.summary,
367 "top_stories": self.top_stories,
368 "trend_analysis": self.trend_analysis,
369 }
370 )
371 return data