Coverage for src / local_deep_research / news / core / card_factory.py: 29%
102 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +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): 50 ↛ 53line 50 didn't jump to line 53 because the condition on line 50 was always true
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) -> SQLCardStorage:
60 """Get or create the storage instance."""
61 if cls._storage is None:
62 # Tables are created automatically when needed
63 cls._storage = SQLCardStorage()
64 return cls._storage
66 @classmethod
67 def create_card(
68 cls,
69 card_type: str,
70 topic: str,
71 source: CardSource,
72 user_id: str,
73 **kwargs,
74 ) -> BaseCard:
75 """
76 Create a new card of the specified type.
78 Args:
79 card_type: Type of card to create ('news', 'research', etc.)
80 topic: The card topic
81 source: Source information for the card
82 user_id: ID of the user creating the card
83 **kwargs: Additional arguments for the specific card type
85 Returns:
86 The created card instance
88 Raises:
89 ValueError: If card_type is not registered
90 """
91 if card_type not in cls._card_types: 91 ↛ 98line 91 didn't jump to line 98 because the condition on line 91 was always true
92 raise ValueError(
93 f"Unknown card type: {card_type}. "
94 f"Available types: {list(cls._card_types.keys())}"
95 )
97 # Generate unique ID
98 card_id = str(uuid.uuid4())
100 # Create the card instance
101 card_class = cls._card_types[card_type]
102 card = card_class(
103 card_id=card_id,
104 topic=topic,
105 source=source,
106 user_id=user_id,
107 **kwargs,
108 )
110 # Save to storage
111 storage = cls.get_storage()
112 card_data = card.to_dict()
113 storage.create(card_data)
115 logger.info(f"Created {card_type} card: {card_id} - {topic}")
116 return card
118 @classmethod
119 def load_card(cls, card_id: str) -> Optional[BaseCard]:
120 """
121 Load a card from storage.
123 Args:
124 card_id: The ID of the card to load
126 Returns:
127 The card instance or None if not found
128 """
129 storage = cls.get_storage()
130 card_data = storage.get(card_id)
132 if not card_data:
133 logger.warning(f"Card not found: {card_id}")
134 return None
136 # Use the helper method to reconstruct the card
137 return cls._reconstruct_card(card_data)
139 @classmethod
140 def get_user_cards(
141 cls,
142 user_id: str,
143 card_types: Optional[List[str]] = None,
144 limit: int = 20,
145 offset: int = 0,
146 ) -> List[BaseCard]:
147 """
148 Get cards for a specific user.
150 Args:
151 user_id: The user ID
152 card_types: Optional list of card types to filter
153 limit: Maximum number of cards to return
154 offset: Offset for pagination
156 Returns:
157 List of card instances
158 """
159 storage = cls.get_storage()
161 # Build filter
162 filters = {"user_id": user_id}
163 if card_types:
164 filters["card_type"] = card_types
166 # Get card data
167 cards_data = storage.list(filters=filters, limit=limit, offset=offset)
169 # Reconstruct cards
170 cards = []
171 for data in cards_data:
172 card = cls._reconstruct_card(data)
173 if card:
174 cards.append(card)
176 return cards
178 @classmethod
179 def get_recent_cards(
180 cls,
181 hours: int = 24,
182 card_types: Optional[List[str]] = None,
183 limit: int = 50,
184 ) -> List[BaseCard]:
185 """
186 Get recent cards across all users.
188 Args:
189 hours: How many hours back to look
190 card_types: Optional list of types to filter
191 limit: Maximum number of cards
193 Returns:
194 List of recent cards
195 """
196 storage = cls.get_storage()
198 # Get recent card data
199 cards_data = storage.get_recent(
200 hours=hours, card_types=card_types, limit=limit
201 )
203 # Reconstruct cards
204 cards = []
205 for data in cards_data:
206 card = cls._reconstruct_card(data)
207 if card:
208 cards.append(card)
210 return cards
212 @classmethod
213 def update_card(cls, card: BaseCard) -> bool:
214 """
215 Update a card in storage.
217 Args:
218 card: The card to update
220 Returns:
221 True if successful
222 """
223 storage = cls.get_storage()
224 return storage.update(card)
226 @classmethod
227 def delete_card(cls, card_id: str) -> bool:
228 """
229 Delete a card from storage.
231 Args:
232 card_id: ID of the card to delete
234 Returns:
235 True if successful
236 """
237 storage = cls.get_storage()
238 return storage.delete(card_id)
240 @classmethod
241 def _reconstruct_card(cls, card_data: Dict[str, Any]) -> Optional[BaseCard]:
242 """
243 Helper to reconstruct a card from storage data.
245 Args:
246 card_data: Dictionary of card data
248 Returns:
249 Reconstructed card or None
250 """
251 try:
252 card_type = card_data.get("card_type", "news")
254 if card_type not in cls._card_types:
255 logger.error(f"Unknown card type: {card_type}")
256 return None
258 # Extract source
259 source_data = card_data.get("source", {})
260 source = CardSource(
261 type=source_data.get("type", "unknown"),
262 source_id=source_data.get("source_id"),
263 created_from=source_data.get("created_from", ""),
264 metadata=source_data.get("metadata", {}),
265 )
267 # Create card
268 card_class = cls._card_types[card_type]
269 card = card_class(
270 card_id=card_data["id"],
271 topic=card_data["topic"],
272 source=source,
273 user_id=card_data["user_id"],
274 )
276 # Restore attributes
277 card.created_at = card_data["created_at"]
278 card.updated_at = card_data["updated_at"]
279 card.versions = card_data.get("versions", [])
280 card.metadata = card_data.get("metadata", {})
281 card.interaction = card_data.get("interaction", {})
283 return card
285 except Exception:
286 logger.exception("Error reconstructing card")
287 return None
289 @classmethod
290 def create_news_card_from_analysis(
291 cls,
292 news_item: Dict[str, Any],
293 source_search_id: str,
294 user_id: str,
295 additional_metadata: Optional[Dict[str, Any]] = None,
296 ) -> NewsCard:
297 """
298 Create a news card from analyzed news data.
300 Args:
301 news_item: Dictionary with news data from analyzer
302 source_search_id: ID of the search that found this
303 user_id: User who initiated the search
305 Returns:
306 Created NewsCard
307 """
308 source = CardSource(
309 type="news_search",
310 source_id=source_search_id,
311 created_from="News analysis",
312 metadata={"analyzer_version": "1.0", "extraction_method": "llm"},
313 )
315 # Merge additional metadata if provided
316 metadata = news_item.get("metadata", {})
317 if additional_metadata:
318 metadata.update(additional_metadata)
320 # Create the news card with all the analyzed data
321 card = cls.create_card(
322 card_type="news",
323 topic=news_item.get("headline", "Untitled"),
324 source=source,
325 user_id=user_id,
326 category=news_item.get("category", "Other"),
327 summary=news_item.get("summary", ""),
328 analysis=news_item.get("analysis", ""),
329 impact_score=news_item.get("impact_score", 5),
330 entities=news_item.get("entities", {}),
331 topics=news_item.get("topics", []),
332 source_url=news_item.get("source_url", ""),
333 is_developing=news_item.get("is_developing", False),
334 surprising_element=news_item.get("surprising_element"),
335 metadata=metadata,
336 )
338 return card
341# Create convenience functions at module level
342def create_card(card_type: str, **kwargs) -> BaseCard:
343 """Convenience function to create a card."""
344 return CardFactory.create_card(card_type, **kwargs)
347def load_card(card_id: str) -> Optional[BaseCard]:
348 """Convenience function to load a card."""
349 return CardFactory.load_card(card_id)