Coverage for src / local_deep_research / news / core / base_card.py: 99%

1""" 

2Base card class for all news-related content. 

3Following LDR's pattern from BaseSearchStrategy. 

4""" 

5 

6from abc import ABC, abstractmethod 

7from dataclasses import dataclass, field 

8from datetime import datetime 

9from typing import Dict, List, Optional, Any 

10 

11from .utils import generate_card_id, utc_now 

12# Storage will be imported when needed to avoid circular import 

13 

14 

15@dataclass 

16class CardSource: 

17 """Tracks the origin of each card.""" 

18 

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) 

23 

24 

25@dataclass 

26class CardVersion: 

27 """Represents a version of research/content for a card.""" 

28 

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 

34 

35 def __post_init__(self): 

36 if not self.version_id: 

37 self.version_id = generate_card_id() 

38 

39 

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 """ 

46 

47 # Required fields 

48 topic: str 

49 source: CardSource 

50 user_id: str 

51 

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) 

56 

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) 

63 

64 # Storage and callback fields 

65 storage: Optional[Any] = field(default=None, init=False) 

66 progress_callback: Optional[Any] = field(default=None, init=False) 

67 

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 } 

79 

80 def set_progress_callback(self, callback) -> None: 

81 """Set a callback function to receive progress updates.""" 

82 self.progress_callback = callback 

83 

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 {}) 

93 

94 def save(self) -> str: 

95 """Save card to database""" 

96 if self.storage is None: 

97 raise RuntimeError("Storage must be set before calling save()") 

98 card_data = { 

99 "id": self.id, 

100 "user_id": self.user_id, 

101 "topic": self.topic, 

102 "card_type": self.get_card_type(), 

103 "source_type": self.source.type, 

104 "source_id": self.source.source_id, 

105 "created_from": self.source.created_from, 

106 "parent_card_id": self.parent_card_id, 

107 "metadata": self.metadata, 

108 } 

109 result: str = self.storage.create(card_data) 

110 return result 

111 

112 def add_version( 

113 self, research_results: Dict[str, Any], query: str, strategy: str 

114 ) -> str: 

115 """Add a new version with research results""" 

116 version_data = { 

117 "search_query": query, 

118 "research_result": research_results, 

119 "headline": self._extract_headline(research_results), 

120 "summary": self._extract_summary(research_results), 

121 "findings": research_results.get("findings", []), 

122 "sources": research_results.get("sources", []), 

123 "impact_score": self._calculate_impact(research_results), 

124 "topics": self._extract_topics(research_results), 

125 "entities": self._extract_entities(research_results), 

126 "strategy": strategy, 

127 } 

128 

129 if self.storage is None: 

130 raise RuntimeError( 

131 "Storage must be set before calling add_version()" 

132 ) 

133 version_id: str = self.storage.add_version(self.id, version_data) 

134 

135 # Add to local versions list 

136 version = CardVersion( 

137 version_id=version_id, 

138 created_at=utc_now(), 

139 content=research_results, 

140 query_used=query, 

141 search_strategy=strategy, 

142 ) 

143 self.versions.append(version) 

144 self.updated_at = utc_now() 

145 

146 return version_id 

147 

148 def get_latest_version(self) -> Optional[CardVersion]: 

149 """Get the most recent version of this card.""" 

150 if not self.versions: 

151 return None 

152 return max(self.versions, key=lambda v: v.created_at) 

153 

154 def to_base_dict(self) -> Dict[str, Any]: 

155 """Convert base card attributes to dictionary.""" 

156 return { 

157 "id": self.id, 

158 "topic": self.topic, 

159 "user_id": self.user_id, 

160 "created_at": self.created_at.isoformat() 

161 if self.created_at 

162 else None, 

163 "updated_at": self.updated_at.isoformat() 

164 if self.updated_at 

165 else None, 

166 "source": { 

167 "type": self.source.type, 

168 "source_id": self.source.source_id, 

169 "created_from": self.source.created_from, 

170 "metadata": self.source.metadata, 

171 }, 

172 "versions_count": len(self.versions), 

173 "parent_card_id": self.parent_card_id, 

174 "metadata": self.metadata, 

175 "interaction": self.interaction, 

176 "card_type": self.get_card_type(), 

177 } 

178 

179 @abstractmethod 

180 def get_card_type(self) -> str: 

181 """Return the card type (news, research, update, overview)""" 

182 pass 

183 

184 @abstractmethod 

185 def to_dict(self) -> Dict[str, Any]: 

186 """ 

187 Convert card to dictionary representation. 

188 Must be implemented by subclasses. 

189 """ 

190 pass 

191 

192 # Helper methods for extracting data from research results 

193 def _extract_headline(self, result: Dict[str, Any]) -> str: 

194 """Extract headline from research result""" 

195 # Try different possible fields 

196 return str( 

197 result.get("headline") 

198 or result.get("title") 

199 or result.get("query", "")[:100] 

200 ) 

201 

202 def _extract_summary(self, result: Dict[str, Any]) -> str: 

203 """Extract summary from research result""" 

204 return str( 

205 result.get("summary") 

206 or result.get("current_knowledge") 

207 or result.get("formatted_findings", "")[:500] 

208 ) 

209 

210 def _calculate_impact(self, result: Dict[str, Any]) -> int: 

211 """Calculate impact score (1-10)""" 

212 # Simple heuristic based on findings count and sources 

213 findings_count = len(result.get("findings", [])) 

214 sources_count = len(result.get("sources", [])) 

215 

216 score = min(10, 5 + (findings_count // 5) + (sources_count // 3)) 

217 return max(1, score) 

218 

219 def _extract_topics(self, result: Dict[str, Any]) -> List[str]: 

220 """Extract topics from research result""" 

221 # Could be enhanced with NLP 

222 topics: List[str] = result.get("topics", []) 

223 if not topics and "query" in result: 

224 # Simple keyword extraction from query 

225 words = result["query"].lower().split() 

226 topics = [w for w in words if len(w) > 4][:5] 

227 return topics 

228 

229 def _extract_entities(self, result: Dict[str, Any]) -> Dict[str, List[str]]: 

230 """Extract entities from research result""" 

231 # Placeholder - would use NER in production 

232 entities: Dict[str, List[str]] = result.get( 

233 "entities", {"people": [], "places": [], "organizations": []} 

234 ) 

235 return entities 

236 

237 

238@dataclass 

239class NewsCard(BaseCard): 

240 """Card representing a news item with potential for research.""" 

241 

242 # News-specific fields with defaults 

243 headline: str = "" 

244 summary: str = "" 

245 category: str = "General" 

246 impact_score: int = 5 

247 entities: Dict[str, List[str]] = field( 

248 default_factory=lambda: { 

249 "people": [], 

250 "places": [], 

251 "organizations": [], 

252 } 

253 ) 

254 topics_extracted: List[str] = field(default_factory=list) 

255 is_developing: bool = False 

256 time_ago: str = "recent" 

257 source_url: str = "" 

258 analysis: str = "" 

259 surprising_element: Optional[str] = None 

260 

261 def __post_init__(self): 

262 """Initialize parent and set headline default.""" 

263 super().__post_init__() 

264 if not self.headline: 

265 self.headline = self.topic 

266 

267 def get_card_type(self) -> str: 

268 """Return the card type""" 

269 return "news" 

270 

271 def to_dict(self) -> Dict[str, Any]: 

272 data = self.to_base_dict() 

273 data.update( 

274 { 

275 "headline": self.headline, 

276 "summary": self.summary, 

277 "category": self.category, 

278 "impact_score": self.impact_score, 

279 "entities": self.entities, 

280 "topics_extracted": self.topics_extracted, 

281 "is_developing": self.is_developing, 

282 "time_ago": self.time_ago, 

283 } 

284 ) 

285 return data 

286 

287 

288@dataclass 

289class ResearchCard(BaseCard): 

290 """Card representing deeper research on a topic.""" 

291 

292 # Research-specific fields 

293 research_depth: str = "quick" # "quick", "detailed", "report" 

294 key_findings: List[str] = field(default_factory=list) 

295 sources_count: int = 0 

296 

297 def get_card_type(self) -> str: 

298 """Return the card type""" 

299 return "research" 

300 

301 def to_dict(self) -> Dict[str, Any]: 

302 data = self.to_base_dict() 

303 data.update( 

304 { 

305 "research_depth": self.research_depth, 

306 "key_findings": self.key_findings, 

307 "sources_count": self.sources_count, 

308 } 

309 ) 

310 return data 

311 

312 

313@dataclass 

314class UpdateCard(BaseCard): 

315 """Card representing updates or notifications.""" 

316 

317 # Update-specific fields 

318 update_type: str = "new_stories" # "new_stories", "breaking", "follow_up" 

319 count: int = 0 

320 preview_items: List[Any] = field(default_factory=list) 

321 since: datetime = field(init=False) 

322 

323 def __post_init__(self): 

324 """Initialize parent and set since timestamp.""" 

325 super().__post_init__() 

326 self.since = utc_now() 

327 

328 def get_card_type(self) -> str: 

329 """Return the card type""" 

330 return "update" 

331 

332 def to_dict(self) -> Dict[str, Any]: 

333 data = self.to_base_dict() 

334 data.update( 

335 { 

336 "update_type": self.update_type, 

337 "count": self.count, 

338 "preview_items": self.preview_items, 

339 "since": self.since.isoformat(), 

340 } 

341 ) 

342 return data 

343 

344 

345@dataclass 

346class OverviewCard(BaseCard): 

347 """Special card type for dashboard/overview display.""" 

348 

349 # Override topic default for overview cards 

350 topic: str = field(default="News Overview", init=False) 

351 

352 # Overview-specific fields 

353 stats: Dict[str, Any] = field( 

354 default_factory=lambda: { 

355 "total_new": 0, 

356 "breaking": 0, 

357 "relevant": 0, 

358 "categories": {}, 

359 } 

360 ) 

361 summary: str = "" 

362 top_stories: List[Any] = field(default_factory=list) 

363 trend_analysis: str = "" 

364 

365 def get_card_type(self) -> str: 

366 """Return the card type""" 

367 return "overview" 

368 

369 def to_dict(self) -> Dict[str, Any]: 

370 data = self.to_base_dict() 

371 data.update( 

372 { 

373 "stats": self.stats, 

374 "summary": self.summary, 

375 "top_stories": self.top_stories, 

376 "trend_analysis": self.trend_analysis, 

377 } 

378 ) 

379 return data