Coverage for src/local_deep_research/news/core/storage_manager.py: 95%
150 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2Centralized storage manager for all news system data.
3Provides unified access to cards, subscriptions, ratings, and preferences.
4"""
6from enum import Enum
7from typing import List, Dict, Any, Optional
8from datetime import datetime, timedelta, timezone
9from loguru import logger
10from flask import has_app_context
12from .card_factory import CardFactory
13from .base_card import BaseCard
14from .card_storage import SQLCardStorage
15from .relevance_service import get_relevance_service
16from ..rating_system.storage import SQLRatingStorage
17from ..preference_manager.storage import SQLPreferenceStorage
20class InteractionType(Enum):
21 """Enum for card interaction types."""
23 VIEW = "view"
24 VOTE_UP = "vote_up"
25 VOTE_DOWN = "vote_down"
26 RESEARCH = "research"
27 SHARE = "share"
30class StorageManager:
31 """
32 Unified storage interface for the news system.
33 Manages all data access across modules.
34 """
36 def __init__(self):
37 """Initialize storage interfaces."""
38 # Storage interfaces will be created on demand
39 self._cards = None
40 self._ratings = None
41 self._preferences = None
43 # Card factory for reconstruction
44 self.card_factory = CardFactory
46 # Relevance service for business logic
47 self.relevance_service = get_relevance_service()
49 logger.info("StorageManager initialized")
51 def _get_current_session(self):
52 """Get the current database session from Flask context (lazy creation)."""
53 if has_app_context():
54 from ...database.session_context import get_g_db_session
56 db_session = get_g_db_session()
57 if db_session: 57 ↛ 61line 57 didn't jump to line 61 because the condition on line 57 was always true
58 return db_session
59 # If no session in context, we need to create one
60 # This will trigger register_activity
61 return None
63 @property
64 def cards(self):
65 """Get cards storage interface."""
66 session = self._get_current_session()
67 if session:
68 return SQLCardStorage(session)
69 # For now, create storage without session
70 # This will fail when trying to use it
71 if self._cards is None: 71 ↛ 73line 71 didn't jump to line 73 because the condition on line 71 was always true
72 raise RuntimeError("No database session available for news storage")
73 return self._cards
75 @property
76 def ratings(self):
77 """Get ratings storage interface."""
78 session = self._get_current_session()
79 if session:
80 return SQLRatingStorage(session)
81 if self._ratings is None: 81 ↛ 83line 81 didn't jump to line 83 because the condition on line 81 was always true
82 raise RuntimeError("No database session available for news storage")
83 return self._ratings
85 @property
86 def preferences(self):
87 """Get preferences storage interface."""
88 session = self._get_current_session()
89 if session:
90 return SQLPreferenceStorage(session)
91 if self._preferences is None: 91 ↛ 93line 91 didn't jump to line 93 because the condition on line 91 was always true
92 raise RuntimeError("No database session available for news storage")
93 return self._preferences
95 def get_user_feed(
96 self,
97 user_id: str,
98 limit: int = 20,
99 offset: int = 0,
100 include_seen: bool = True,
101 card_types: Optional[List[str]] = None,
102 ) -> List[BaseCard]:
103 """
104 Get personalized news feed for a user.
106 Args:
107 user_id: The user ID
108 limit: Maximum cards to return
109 offset: Pagination offset
110 include_seen: Whether to include already viewed cards
111 card_types: Filter by card types
113 Returns:
114 List of cards sorted by relevance
115 """
116 try:
117 # Get user preferences
118 user_prefs = self.preferences.get(user_id)
120 # Get recent cards
121 if user_prefs and user_prefs.get("liked_categories"):
122 # Filter by user's preferred categories
123 filters = {
124 "user_id": user_id,
125 "categories": user_prefs["liked_categories"],
126 }
127 else:
128 filters = {"user_id": user_id}
130 if card_types:
131 filters["card_type"] = card_types
133 # Get cards from storage
134 cards_data = self.cards.list(
135 filters=filters,
136 limit=limit * 2, # Get extra to filter
137 offset=offset,
138 )
140 # Reconstruct card objects
141 cards = []
142 for data in cards_data:
143 card = CardFactory.load_card(data["id"])
144 if card: 144 ↛ 142line 144 didn't jump to line 142 because the condition on line 144 was always true
145 cards.append(card)
147 # Apply personalization
148 cards = self.relevance_service.personalize_feed(
149 cards, user_prefs, include_seen=include_seen
150 )
152 return cards[:limit]
154 except Exception:
155 logger.exception("Error getting user feed")
156 return []
158 def get_trending_news(
159 self, hours: int = 24, limit: int = 10, min_impact: int = 7
160 ) -> List[BaseCard]:
161 """
162 Get trending news across all users.
164 Args:
165 hours: Look back period
166 limit: Maximum cards
167 min_impact: Minimum impact score
169 Returns:
170 List of trending news cards
171 """
172 try:
173 # Get recent high-impact cards
174 cards = CardFactory.get_recent_cards(
175 hours=hours, card_types=["news"], limit=limit * 2
176 )
178 # Use relevance service to filter and sort trending
179 return self.relevance_service.filter_trending(
180 cards, min_impact=min_impact, limit=limit
181 )
183 except Exception:
184 logger.exception("Error getting trending news")
185 return []
187 def record_interaction(
188 self,
189 user_id: str,
190 card_id: str,
191 interaction_type: InteractionType,
192 metadata: Optional[Dict] = None,
193 ) -> bool:
194 """
195 Record user interaction with a card.
197 Args:
198 user_id: The user
199 card_id: The card
200 interaction_type: Type of interaction (view, vote, share, etc.)
201 metadata: Additional data
203 Returns:
204 Success status
205 """
206 try:
207 # Load the card
208 card = CardFactory.load_card(card_id)
209 if not card:
210 return False
212 # Update interaction data
213 if interaction_type == InteractionType.VIEW:
214 card.interaction["viewed"] = True
215 card.interaction["last_viewed"] = datetime.now(timezone.utc)
216 card.interaction["views"] = card.interaction.get("views", 0) + 1
218 elif interaction_type == InteractionType.VOTE_UP:
219 card.interaction["voted"] = "up"
220 card.interaction["votes_up"] = (
221 card.interaction.get("votes_up", 0) + 1
222 )
223 # Record in ratings
224 self.ratings.save(
225 {
226 "user_id": user_id,
227 "card_id": card_id,
228 "rating_type": "relevance",
229 "value": 1,
230 }
231 )
233 elif interaction_type == InteractionType.VOTE_DOWN:
234 card.interaction["voted"] = "down"
235 card.interaction["votes_down"] = (
236 card.interaction.get("votes_down", 0) + 1
237 )
238 # Record in ratings
239 self.ratings.save(
240 {
241 "user_id": user_id,
242 "card_id": card_id,
243 "rating_type": "relevance",
244 "value": -1,
245 }
246 )
248 elif interaction_type == InteractionType.RESEARCH: 248 ↛ 255line 248 didn't jump to line 255 because the condition on line 248 was always true
249 card.interaction["researched"] = True
250 card.interaction["research_count"] = (
251 card.interaction.get("research_count", 0) + 1
252 )
254 # Add metadata if provided
255 if metadata:
256 card.interaction[f"{interaction_type}_metadata"] = metadata
258 # Save updated card
259 return CardFactory.update_card(card)
261 except Exception:
262 logger.exception("Error recording interaction")
263 return False
265 def get_card(self, card_id: str) -> Optional[BaseCard]:
266 """
267 Get a single card by ID.
269 Args:
270 card_id: The card ID
272 Returns:
273 Card object or None
274 """
275 try:
276 return CardFactory.load_card(card_id)
277 except Exception:
278 logger.exception("Error getting card")
279 return None
281 def get_card_interactions(self, card_id: str) -> List[Dict[str, Any]]:
282 """
283 Get all interactions for a card.
285 Args:
286 card_id: The card ID
288 Returns:
289 List of interaction records
290 """
291 try:
292 # Get ratings which represent votes
293 ratings = self.ratings.list({"card_id": card_id})
295 # Convert to interaction format
296 interactions = []
297 for rating in ratings:
298 interaction = {
299 "user_id": rating.get("user_id"),
300 "interaction_type": "vote",
301 "interaction_data": {
302 "vote": "up" if rating.get("value", 0) > 0 else "down"
303 },
304 "timestamp": rating.get("created_at"),
305 }
306 interactions.append(interaction)
308 return interactions
310 except Exception:
311 logger.exception("Error getting card interactions")
312 return []
314 def update_card(self, card: BaseCard) -> bool:
315 """
316 Update a card in storage.
318 Args:
319 card: The card to update
321 Returns:
322 Success status
323 """
324 try:
325 return CardFactory.update_card(card)
326 except Exception:
327 logger.exception("Error updating card")
328 return False
330 def cleanup_old_data(self, days: int = 30) -> Dict[str, int]:
331 """
332 Clean up old data from all storages.
334 Args:
335 days: Age threshold in days
337 Returns:
338 Counts of deleted items
339 """
340 try:
341 cutoff = datetime.now(timezone.utc) - timedelta(days=days)
342 counts = {}
344 # Clean old cards (except subscribed ones)
345 old_cards = self.cards.list(
346 {"created_before": cutoff, "not_subscribed": True}
347 )
348 deleted_cards = 0
349 for card_data in old_cards:
350 if self.cards.delete(card_data["id"]): 350 ↛ 349line 350 didn't jump to line 349 because the condition on line 350 was always true
351 deleted_cards += 1
352 counts["cards"] = deleted_cards
354 # Clean old ratings
355 old_ratings = self.ratings.list({"created_before": cutoff})
356 counts["ratings"] = len(old_ratings)
357 for rating in old_ratings:
358 self.ratings.delete(rating["id"])
360 logger.info(f"Cleanup complete: {counts}")
361 return counts
363 except Exception:
364 logger.exception("Error during cleanup")
365 return {}