Coverage for src/local_deep_research/news/core/card_factory.py: 98%
124 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Factory for creating and managing different types of cards.
3Handles card creation, loading, and type registration.
4"""
6import uuid
7from typing import Dict, Type, Optional, Any, List, cast
8from loguru import logger
10from .base_card import (
11 BaseCard,
12 CardSource,
13 NewsCard,
14 ResearchCard,
15 UpdateCard,
16 OverviewCard,
17)
18from .card_storage import SQLCardStorage
19# from ..database import init_news_database # Not needed - tables created on demand
22class CardFactory:
23 """
24 Factory for creating and managing cards.
25 Provides a unified interface for card operations.
26 """
28 # Registry of card types
29 _card_types: Dict[str, Type[BaseCard]] = {
30 "news": NewsCard,
31 "research": ResearchCard,
32 "update": UpdateCard,
33 "overview": OverviewCard,
34 }
36 # Storage instance (singleton)
37 _storage: Optional[SQLCardStorage] = None
39 @classmethod
40 def register_card_type(
41 cls, type_name: str, card_class: Type[BaseCard]
42 ) -> None:
43 """
44 Register a new card type.
46 Args:
47 type_name: The name identifier for this card type
48 card_class: The class to use for this type
49 """
50 if not issubclass(card_class, BaseCard):
51 raise ValueError(f"{card_class} must be a subclass of BaseCard")
53 cls._card_types[type_name] = card_class
54 logger.info(
55 f"Registered card type: {type_name} -> {card_class.__name__}"
56 )
58 @classmethod
59 def get_storage(cls, session=None) -> SQLCardStorage:
60 """Get or create the storage instance.
62 Args:
63 session: SQLAlchemy session. If not provided, attempts to get
64 session from Flask context (g.db_session).
66 Returns:
67 SQLCardStorage instance
69 Raises:
70 RuntimeError: If no session is available
71 """
72 if session is not None:
73 # Return a fresh storage with the provided session
74 return SQLCardStorage(session)
76 # Resolve the current user from the FastAPI request contextvar
77 # and open their encrypted DB session.
78 from ...utilities.request_context import get_current_username
79 from ...database.session_context import get_user_db_session
81 username = get_current_username()
82 if username: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 with get_user_db_session(username) as db_session:
84 return SQLCardStorage(db_session)
86 raise RuntimeError(
87 "No database session available. Pass a session to get_storage() "
88 "or ensure this is called within a request with the "
89 "authenticated user contextvar set."
90 )
92 @classmethod
93 def create_card(
94 cls,
95 card_type: str,
96 topic: str,
97 source: CardSource,
98 user_id: str,
99 **kwargs,
100 ) -> BaseCard:
101 """
102 Create a new card of the specified type.
104 Args:
105 card_type: Type of card to create ('news', 'research', etc.)
106 topic: The card topic
107 source: Source information for the card
108 user_id: ID of the user creating the card
109 **kwargs: Additional arguments for the specific card type
111 Returns:
112 The created card instance
114 Raises:
115 ValueError: If card_type is not registered
116 """
117 if card_type not in cls._card_types:
118 raise ValueError(
119 f"Unknown card type: {card_type}. "
120 f"Available types: {list(cls._card_types.keys())}"
121 )
123 # Generate unique ID
124 card_id = str(uuid.uuid4())
126 # Create the card instance
127 card_class = cls._card_types[card_type]
128 card = card_class(
129 card_id=card_id,
130 topic=topic,
131 source=source,
132 user_id=user_id,
133 **kwargs,
134 )
136 # Save to storage
137 storage = cls.get_storage()
138 card_data = card.to_dict()
139 storage.create(card_data)
141 logger.info(f"Created {card_type} card: {card_id} - {topic}")
142 return card
144 @classmethod
145 def load_card(cls, card_id: str) -> Optional[BaseCard]:
146 """
147 Load a card from storage.
149 Args:
150 card_id: The ID of the card to load
152 Returns:
153 The card instance or None if not found
154 """
155 storage = cls.get_storage()
156 card_data = storage.get(card_id)
158 if not card_data:
159 logger.warning(f"Card not found: {card_id}")
160 return None
162 # Use the helper method to reconstruct the card
163 return cls._reconstruct_card(card_data)
165 @classmethod
166 def get_user_cards(
167 cls,
168 user_id: str,
169 card_types: Optional[List[str]] = None,
170 limit: int = 20,
171 offset: int = 0,
172 ) -> List[BaseCard]:
173 """
174 Get cards for a specific user.
176 Args:
177 user_id: The user ID
178 card_types: Optional list of card types to filter
179 limit: Maximum number of cards to return
180 offset: Offset for pagination
182 Returns:
183 List of card instances
184 """
185 storage = cls.get_storage()
187 # Build filter
188 filters: Dict[str, Any] = {"user_id": user_id}
189 if card_types:
190 filters["card_type"] = card_types
192 # Get card data
193 cards_data = storage.list(filters=filters, limit=limit, offset=offset)
195 # Reconstruct cards
196 cards = []
197 for data in cards_data:
198 card = cls._reconstruct_card(data)
199 if card:
200 cards.append(card)
202 return cards
204 @classmethod
205 def get_recent_cards(
206 cls,
207 hours: int = 24,
208 card_types: Optional[List[str]] = None,
209 limit: int = 50,
210 ) -> List[BaseCard]:
211 """
212 Get recent cards across all users.
214 Args:
215 hours: How many hours back to look
216 card_types: Optional list of types to filter
217 limit: Maximum number of cards
219 Returns:
220 List of recent cards
221 """
222 storage = cls.get_storage()
224 # Get recent card data
225 cards_data = storage.get_recent(
226 hours=hours, card_types=card_types, limit=limit
227 )
229 # Reconstruct cards
230 cards = []
231 for data in cards_data:
232 card = cls._reconstruct_card(data)
233 if card: 233 ↛ 231line 233 didn't jump to line 231 because the condition on line 233 was always true
234 cards.append(card)
236 return cards
238 @classmethod
239 def update_card(cls, card: BaseCard, session=None) -> bool:
240 """
241 Update a card in storage.
243 Args:
244 card: The card to update
245 session: Optional SQLAlchemy session
247 Returns:
248 True if successful
249 """
250 storage = cls.get_storage(session)
251 # Convert card to dict for storage.update() which expects (id, data)
252 card_data = card.to_dict()
253 # Include interaction data for updates
254 card_data["interaction"] = card.interaction
255 return storage.update(card.id, card_data)
257 @classmethod
258 def delete_card(cls, card_id: str) -> bool:
259 """
260 Delete a card from storage.
262 Args:
263 card_id: ID of the card to delete
265 Returns:
266 True if successful
267 """
268 storage = cls.get_storage()
269 return storage.delete(card_id)
271 @classmethod
272 def _reconstruct_card(cls, card_data: Dict[str, Any]) -> Optional[BaseCard]:
273 """
274 Helper to reconstruct a card from storage data.
276 Args:
277 card_data: Dictionary of card data
279 Returns:
280 Reconstructed card or None
281 """
282 from datetime import datetime
284 try:
285 # Handle enum card_type - could be string or enum
286 card_type = card_data.get("card_type", "news")
287 if hasattr(card_type, "value"):
288 card_type = card_type.value
290 if card_type not in cls._card_types:
291 logger.error(f"Unknown card type: {card_type}")
292 return None
294 # Extract source - may be dict or may need construction
295 source_data = card_data.get("source", {})
296 if not source_data:
297 # Construct from flat fields
298 source_data = {
299 "type": card_data.get("source_type", "unknown"),
300 "source_id": card_data.get("source_id"),
301 "created_from": card_data.get("created_from", ""),
302 "metadata": {},
303 }
304 source = CardSource(
305 type=source_data.get("type", "unknown"),
306 source_id=source_data.get("source_id"),
307 created_from=source_data.get("created_from", ""),
308 metadata=source_data.get("metadata", {}),
309 )
311 # Get user_id, handling None case
312 user_id = card_data.get("user_id") or "unknown"
314 # Create card
315 card_class = cls._card_types[card_type]
316 card = card_class(
317 card_id=card_data["id"],
318 topic=card_data.get("topic") or card_data.get("title", ""),
319 source=source,
320 user_id=user_id,
321 )
323 # Restore attributes - handle ISO string dates
324 created_at = card_data.get("created_at")
325 if created_at:
326 if isinstance(created_at, str):
327 card.created_at = datetime.fromisoformat(
328 created_at.replace("Z", "+00:00")
329 )
330 else:
331 card.created_at = created_at
333 updated_at = card_data.get("updated_at")
334 if updated_at:
335 if isinstance(updated_at, str):
336 card.updated_at = datetime.fromisoformat(
337 updated_at.replace("Z", "+00:00")
338 )
339 else:
340 card.updated_at = updated_at
342 card.versions = card_data.get("versions", [])
343 card.metadata = card_data.get("metadata", {})
344 card.interaction = card_data.get("interaction", {})
346 return card
348 except Exception:
349 logger.exception("Error reconstructing card")
350 return None
352 @classmethod
353 def create_news_card_from_analysis(
354 cls,
355 news_item: Dict[str, Any],
356 source_search_id: str,
357 user_id: str,
358 additional_metadata: Optional[Dict[str, Any]] = None,
359 ) -> NewsCard:
360 """
361 Create a news card from analyzed news data.
363 Args:
364 news_item: Dictionary with news data from analyzer
365 source_search_id: ID of the search that found this
366 user_id: User who initiated the search
368 Returns:
369 Created NewsCard
370 """
371 source = CardSource(
372 type="news_search",
373 source_id=source_search_id,
374 created_from="News analysis",
375 metadata={"analyzer_version": "1.0", "extraction_method": "llm"},
376 )
378 # Merge additional metadata if provided
379 metadata = news_item.get("metadata", {})
380 if additional_metadata:
381 metadata.update(additional_metadata)
383 # Create the news card with all the analyzed data
384 card = cls.create_card(
385 card_type="news",
386 topic=news_item.get("headline", "Untitled"),
387 source=source,
388 user_id=user_id,
389 category=news_item.get("category", "Other"),
390 summary=news_item.get("summary", ""),
391 analysis=news_item.get("analysis", ""),
392 impact_score=news_item.get("impact_score", 5),
393 entities=news_item.get("entities", {}),
394 topics=news_item.get("topics", []),
395 source_url=news_item.get("source_url", ""),
396 is_developing=news_item.get("is_developing", False),
397 surprising_element=news_item.get("surprising_element"),
398 metadata=metadata,
399 )
401 return cast(NewsCard, card)
404# Create convenience functions at module level
405def create_card(card_type: str, **kwargs) -> BaseCard:
406 """Convenience function to create a card."""
407 return CardFactory.create_card(card_type, **kwargs)
410def load_card(card_id: str) -> Optional[BaseCard]:
411 """Convenience function to load a card."""
412 return CardFactory.load_card(card_id)