Coverage for src / local_deep_research / news / core / card_factory.py: 94%
127 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:55 +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 # Try to get session from Flask context (lazy creation)
77 try:
78 from flask import has_app_context
80 if has_app_context(): 80 ↛ 81line 80 didn't jump to line 81 because the condition on line 80 was never true
81 from ...database.session_context import get_g_db_session
83 db_session = get_g_db_session()
84 if db_session:
85 return SQLCardStorage(db_session)
86 except ImportError:
87 pass
89 raise RuntimeError(
90 "No database session available. Pass a session to get_storage() "
91 "or ensure this is called within a Flask request context with "
92 "g.db_session set."
93 )
95 @classmethod
96 def create_card(
97 cls,
98 card_type: str,
99 topic: str,
100 source: CardSource,
101 user_id: str,
102 **kwargs,
103 ) -> BaseCard:
104 """
105 Create a new card of the specified type.
107 Args:
108 card_type: Type of card to create ('news', 'research', etc.)
109 topic: The card topic
110 source: Source information for the card
111 user_id: ID of the user creating the card
112 **kwargs: Additional arguments for the specific card type
114 Returns:
115 The created card instance
117 Raises:
118 ValueError: If card_type is not registered
119 """
120 if card_type not in cls._card_types:
121 raise ValueError(
122 f"Unknown card type: {card_type}. "
123 f"Available types: {list(cls._card_types.keys())}"
124 )
126 # Generate unique ID
127 card_id = str(uuid.uuid4())
129 # Create the card instance
130 card_class = cls._card_types[card_type]
131 card = card_class(
132 card_id=card_id,
133 topic=topic,
134 source=source,
135 user_id=user_id,
136 **kwargs,
137 )
139 # Save to storage
140 storage = cls.get_storage()
141 card_data = card.to_dict()
142 storage.create(card_data)
144 logger.info(f"Created {card_type} card: {card_id} - {topic}")
145 return card
147 @classmethod
148 def load_card(cls, card_id: str) -> Optional[BaseCard]:
149 """
150 Load a card from storage.
152 Args:
153 card_id: The ID of the card to load
155 Returns:
156 The card instance or None if not found
157 """
158 storage = cls.get_storage()
159 card_data = storage.get(card_id)
161 if not card_data:
162 logger.warning(f"Card not found: {card_id}")
163 return None
165 # Use the helper method to reconstruct the card
166 return cls._reconstruct_card(card_data)
168 @classmethod
169 def get_user_cards(
170 cls,
171 user_id: str,
172 card_types: Optional[List[str]] = None,
173 limit: int = 20,
174 offset: int = 0,
175 ) -> List[BaseCard]:
176 """
177 Get cards for a specific user.
179 Args:
180 user_id: The user ID
181 card_types: Optional list of card types to filter
182 limit: Maximum number of cards to return
183 offset: Offset for pagination
185 Returns:
186 List of card instances
187 """
188 storage = cls.get_storage()
190 # Build filter
191 filters: Dict[str, Any] = {"user_id": user_id}
192 if card_types:
193 filters["card_type"] = card_types
195 # Get card data
196 cards_data = storage.list(filters=filters, limit=limit, offset=offset)
198 # Reconstruct cards
199 cards = []
200 for data in cards_data:
201 card = cls._reconstruct_card(data)
202 if card:
203 cards.append(card)
205 return cards
207 @classmethod
208 def get_recent_cards(
209 cls,
210 hours: int = 24,
211 card_types: Optional[List[str]] = None,
212 limit: int = 50,
213 ) -> List[BaseCard]:
214 """
215 Get recent cards across all users.
217 Args:
218 hours: How many hours back to look
219 card_types: Optional list of types to filter
220 limit: Maximum number of cards
222 Returns:
223 List of recent cards
224 """
225 storage = cls.get_storage()
227 # Get recent card data
228 cards_data = storage.get_recent(
229 hours=hours, card_types=card_types, limit=limit
230 )
232 # Reconstruct cards
233 cards = []
234 for data in cards_data:
235 card = cls._reconstruct_card(data)
236 if card: 236 ↛ 234line 236 didn't jump to line 234 because the condition on line 236 was always true
237 cards.append(card)
239 return cards
241 @classmethod
242 def update_card(cls, card: BaseCard, session=None) -> bool:
243 """
244 Update a card in storage.
246 Args:
247 card: The card to update
248 session: Optional SQLAlchemy session
250 Returns:
251 True if successful
252 """
253 storage = cls.get_storage(session)
254 # Convert card to dict for storage.update() which expects (id, data)
255 card_data = card.to_dict()
256 # Include interaction data for updates
257 card_data["interaction"] = card.interaction
258 return storage.update(card.id, card_data)
260 @classmethod
261 def delete_card(cls, card_id: str) -> bool:
262 """
263 Delete a card from storage.
265 Args:
266 card_id: ID of the card to delete
268 Returns:
269 True if successful
270 """
271 storage = cls.get_storage()
272 return storage.delete(card_id)
274 @classmethod
275 def _reconstruct_card(cls, card_data: Dict[str, Any]) -> Optional[BaseCard]:
276 """
277 Helper to reconstruct a card from storage data.
279 Args:
280 card_data: Dictionary of card data
282 Returns:
283 Reconstructed card or None
284 """
285 from datetime import datetime
287 try:
288 # Handle enum card_type - could be string or enum
289 card_type = card_data.get("card_type", "news")
290 if hasattr(card_type, "value"):
291 card_type = card_type.value
293 if card_type not in cls._card_types:
294 logger.error(f"Unknown card type: {card_type}")
295 return None
297 # Extract source - may be dict or may need construction
298 source_data = card_data.get("source", {})
299 if not source_data:
300 # Construct from flat fields
301 source_data = {
302 "type": card_data.get("source_type", "unknown"),
303 "source_id": card_data.get("source_id"),
304 "created_from": card_data.get("created_from", ""),
305 "metadata": {},
306 }
307 source = CardSource(
308 type=source_data.get("type", "unknown"),
309 source_id=source_data.get("source_id"),
310 created_from=source_data.get("created_from", ""),
311 metadata=source_data.get("metadata", {}),
312 )
314 # Get user_id, handling None case
315 user_id = card_data.get("user_id") or "unknown"
317 # Create card
318 card_class = cls._card_types[card_type]
319 card = card_class(
320 card_id=card_data["id"],
321 topic=card_data.get("topic") or card_data.get("title", ""),
322 source=source,
323 user_id=user_id,
324 )
326 # Restore attributes - handle ISO string dates
327 created_at = card_data.get("created_at")
328 if created_at:
329 if isinstance(created_at, str):
330 card.created_at = datetime.fromisoformat(
331 created_at.replace("Z", "+00:00")
332 )
333 else:
334 card.created_at = created_at
336 updated_at = card_data.get("updated_at")
337 if updated_at:
338 if isinstance(updated_at, str):
339 card.updated_at = datetime.fromisoformat(
340 updated_at.replace("Z", "+00:00")
341 )
342 else:
343 card.updated_at = updated_at
345 card.versions = card_data.get("versions", [])
346 card.metadata = card_data.get("metadata", {})
347 card.interaction = card_data.get("interaction", {})
349 return card
351 except Exception:
352 logger.exception("Error reconstructing card")
353 return None
355 @classmethod
356 def create_news_card_from_analysis(
357 cls,
358 news_item: Dict[str, Any],
359 source_search_id: str,
360 user_id: str,
361 additional_metadata: Optional[Dict[str, Any]] = None,
362 ) -> NewsCard:
363 """
364 Create a news card from analyzed news data.
366 Args:
367 news_item: Dictionary with news data from analyzer
368 source_search_id: ID of the search that found this
369 user_id: User who initiated the search
371 Returns:
372 Created NewsCard
373 """
374 source = CardSource(
375 type="news_search",
376 source_id=source_search_id,
377 created_from="News analysis",
378 metadata={"analyzer_version": "1.0", "extraction_method": "llm"},
379 )
381 # Merge additional metadata if provided
382 metadata = news_item.get("metadata", {})
383 if additional_metadata:
384 metadata.update(additional_metadata)
386 # Create the news card with all the analyzed data
387 card = cls.create_card(
388 card_type="news",
389 topic=news_item.get("headline", "Untitled"),
390 source=source,
391 user_id=user_id,
392 category=news_item.get("category", "Other"),
393 summary=news_item.get("summary", ""),
394 analysis=news_item.get("analysis", ""),
395 impact_score=news_item.get("impact_score", 5),
396 entities=news_item.get("entities", {}),
397 topics=news_item.get("topics", []),
398 source_url=news_item.get("source_url", ""),
399 is_developing=news_item.get("is_developing", False),
400 surprising_element=news_item.get("surprising_element"),
401 metadata=metadata,
402 )
404 return cast(NewsCard, card)
407# Create convenience functions at module level
408def create_card(card_type: str, **kwargs) -> BaseCard:
409 """Convenience function to create a card."""
410 return CardFactory.create_card(card_type, **kwargs)
413def load_card(card_id: str) -> Optional[BaseCard]:
414 """Convenience function to load a card."""
415 return CardFactory.load_card(card_id)