Coverage for src / local_deep_research / news / subscription_manager / search_subscription.py: 100%
61 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:55 +0000
1"""
2Search subscription - allows users to subscribe to their LDR searches for news updates.
3This is the killer feature: turning searches into living news feeds.
4"""
6from typing import Optional, Dict, Any
7from loguru import logger
9from .base_subscription import BaseSubscription
10from ..core.base_card import CardSource
13class SearchSubscription(BaseSubscription):
14 """
15 Subscription based on a user's search query.
16 Transforms the original search into news-focused queries.
17 """
19 def __init__(
20 self,
21 user_id: str,
22 query: str,
23 source: Optional[CardSource] = None,
24 refresh_interval_minutes: int = 360, # Default 6 hours
25 transform_to_news_query: bool = True,
26 subscription_id: Optional[str] = None,
27 ):
28 """
29 Initialize a search subscription.
31 Args:
32 user_id: ID of the user
33 query: The original search query
34 source: Source information (auto-created if not provided)
35 refresh_interval_minutes: How often to check for news in minutes
36 transform_to_news_query: Whether to add news context to query
37 subscription_id: Optional ID
38 """
39 # Create source if not provided
40 if source is None:
41 source = CardSource(
42 type="user_search", created_from=f"Search subscription: {query}"
43 )
45 super().__init__(
46 user_id, source, query, refresh_interval_minutes, subscription_id
47 )
49 self.original_query = query
50 self.transform_to_news_query = transform_to_news_query
52 # Track query evolution over time
53 self.query_history = [query]
54 self.current_query = query
56 # Set subscription type
57 self.subscription_type = "search"
59 # Metadata specific to search subscriptions
60 self.metadata.update(
61 {
62 "subscription_type": "search",
63 "original_query": query,
64 "transform_enabled": transform_to_news_query,
65 }
66 )
68 logger.info(f"Created search subscription for query: {query}")
70 @property
71 def query(self) -> str:
72 """Get the original query for backward compatibility."""
73 return self.original_query
75 def get_subscription_type(self) -> str:
76 """Return the subscription type identifier."""
77 return "search_subscription"
79 def generate_search_query(self) -> str:
80 """
81 Generate the news search query from the original search.
83 Returns:
84 str: The transformed search query
85 """
86 # Update any date placeholders with current date in user's timezone
87 from ..core.utils import get_local_date_string
89 current_date = get_local_date_string()
90 updated_query = self.current_query
92 # Replace YYYY-MM-DD placeholder ONLY (not all dates)
93 updated_query = updated_query.replace("YYYY-MM-DD", current_date)
95 if self.transform_to_news_query:
96 # Add news context to the updated query
97 news_query = self._transform_to_news_query(updated_query)
98 else:
99 news_query = updated_query
101 logger.debug(f"Generated news query: {news_query}")
102 return news_query
104 def _transform_to_news_query(self, query: str) -> str:
105 """
106 Transform a regular search query into a news-focused query.
108 Args:
109 query: The original query
111 Returns:
112 str: News-focused version of the query
113 """
114 # Don't double-add news terms
115 query_lower = query.lower()
116 if any(
117 term in query_lower
118 for term in ["news", "latest", "recent", "today"]
119 ):
120 return query
122 # Add temporal and news context
123 # This could be more sophisticated with LLM in the future
125 # Choose appropriate term based on query type
126 if any(term in query_lower for term in ["how to", "tutorial", "guide"]):
127 # Technical queries - look for updates
128 return f"{query} latest updates developments"
129 if any(
130 term in query_lower
131 for term in ["vulnerability", "security", "breach"]
132 ):
133 # Security queries - urgent news
134 return f"{query} breaking news alerts today"
135 # General queries - latest news
136 return f"{query} latest news developments"
138 def evolve_query(self, new_terms: Optional[str] = None) -> None:
139 """
140 Evolve the query based on emerging trends.
141 Future feature: LLM-based query evolution.
143 Args:
144 new_terms: New terms to incorporate
145 """
146 if new_terms:
147 # Simple evolution for now
148 evolved_query = f"{self.original_query} {new_terms}"
149 self.current_query = evolved_query
150 self.query_history.append(evolved_query)
152 logger.info(f"Evolved search query to: {evolved_query}")
154 def get_statistics(self) -> Dict[str, Any]:
155 """Get statistics about this subscription."""
156 return {
157 "original_query": self.original_query,
158 "current_query": self.current_query,
159 "query_evolution_count": len(self.query_history) - 1,
160 "total_refreshes": self.refresh_count,
161 "success_rate": (
162 self.refresh_count / (self.refresh_count + self.error_count)
163 if (self.refresh_count + self.error_count) > 0
164 else 0
165 ),
166 }
168 def to_dict(self) -> Dict[str, Any]:
169 """Convert to dictionary representation."""
170 data = super().to_dict()
171 data.update(
172 {
173 "original_query": self.original_query,
174 "current_query": self.current_query,
175 "transform_to_news_query": self.transform_to_news_query,
176 "query_history": self.query_history,
177 "statistics": self.get_statistics(),
178 }
179 )
180 return data
183class SearchSubscriptionFactory:
184 """
185 Factory for creating search subscriptions from various sources.
186 """
188 @staticmethod
189 def from_user_search(
190 user_id: str,
191 search_query: str,
192 search_result_id: Optional[str] = None,
193 **kwargs,
194 ) -> SearchSubscription:
195 """
196 Create a subscription from a user's search.
198 Args:
199 user_id: The user who performed the search
200 search_query: The original search query
201 search_result_id: Optional ID of the search result
202 **kwargs: Additional arguments for SearchSubscription
204 Returns:
205 SearchSubscription instance
206 """
207 source = CardSource(
208 type="user_search",
209 source_id=search_result_id,
210 created_from=f"Your search: '{search_query}'",
211 metadata={
212 "search_timestamp": kwargs.get("search_timestamp"),
213 "search_strategy": kwargs.get("search_strategy"),
214 },
215 )
217 return SearchSubscription(
218 user_id=user_id, query=search_query, source=source, **kwargs
219 )
221 @staticmethod
222 def from_recommendation(
223 user_id: str,
224 recommended_query: str,
225 recommendation_source: str,
226 **kwargs,
227 ) -> SearchSubscription:
228 """
229 Create a subscription from a system recommendation.
231 Args:
232 user_id: The user to create subscription for
233 recommended_query: The recommended search query
234 recommendation_source: What generated this recommendation
235 **kwargs: Additional arguments
237 Returns:
238 SearchSubscription instance
239 """
240 source = CardSource(
241 type="recommendation",
242 created_from=f"Recommended based on: {recommendation_source}",
243 metadata={
244 "recommendation_type": kwargs.get(
245 "recommendation_type", "topic_based"
246 )
247 },
248 )
250 return SearchSubscription(
251 user_id=user_id, query=recommended_query, source=source, **kwargs
252 )