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