Coverage for src / local_deep_research / news / subscription_manager / search_subscription.py: 26%
62 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +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 elif 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 else:
136 # General queries - latest news
137 return f"{query} latest news developments"
139 def evolve_query(self, new_terms: Optional[str] = None) -> None:
140 """
141 Evolve the query based on emerging trends.
142 Future feature: LLM-based query evolution.
144 Args:
145 new_terms: New terms to incorporate
146 """
147 if new_terms:
148 # Simple evolution for now
149 evolved_query = f"{self.original_query} {new_terms}"
150 self.current_query = evolved_query
151 self.query_history.append(evolved_query)
153 logger.info(f"Evolved search query to: {evolved_query}")
155 def get_statistics(self) -> Dict[str, Any]:
156 """Get statistics about this subscription."""
157 stats = {
158 "original_query": self.original_query,
159 "current_query": self.current_query,
160 "query_evolution_count": len(self.query_history) - 1,
161 "total_refreshes": self.refresh_count,
162 "success_rate": (
163 self.refresh_count / (self.refresh_count + self.error_count)
164 if (self.refresh_count + self.error_count) > 0
165 else 0
166 ),
167 }
168 return stats
170 def to_dict(self) -> Dict[str, Any]:
171 """Convert to dictionary representation."""
172 data = super().to_dict()
173 data.update(
174 {
175 "original_query": self.original_query,
176 "current_query": self.current_query,
177 "transform_to_news_query": self.transform_to_news_query,
178 "query_history": self.query_history,
179 "statistics": self.get_statistics(),
180 }
181 )
182 return data
185class SearchSubscriptionFactory:
186 """
187 Factory for creating search subscriptions from various sources.
188 """
190 @staticmethod
191 def from_user_search(
192 user_id: str,
193 search_query: str,
194 search_result_id: Optional[str] = None,
195 **kwargs,
196 ) -> SearchSubscription:
197 """
198 Create a subscription from a user's search.
200 Args:
201 user_id: The user who performed the search
202 search_query: The original search query
203 search_result_id: Optional ID of the search result
204 **kwargs: Additional arguments for SearchSubscription
206 Returns:
207 SearchSubscription instance
208 """
209 source = CardSource(
210 type="user_search",
211 source_id=search_result_id,
212 created_from=f"Your search: '{search_query}'",
213 metadata={
214 "search_timestamp": kwargs.get("search_timestamp"),
215 "search_strategy": kwargs.get("search_strategy"),
216 },
217 )
219 return SearchSubscription(
220 user_id=user_id, query=search_query, source=source, **kwargs
221 )
223 @staticmethod
224 def from_recommendation(
225 user_id: str,
226 recommended_query: str,
227 recommendation_source: str,
228 **kwargs,
229 ) -> SearchSubscription:
230 """
231 Create a subscription from a system recommendation.
233 Args:
234 user_id: The user to create subscription for
235 recommended_query: The recommended search query
236 recommendation_source: What generated this recommendation
237 **kwargs: Additional arguments
239 Returns:
240 SearchSubscription instance
241 """
242 source = CardSource(
243 type="recommendation",
244 created_from=f"Recommended based on: {recommendation_source}",
245 metadata={
246 "recommendation_type": kwargs.get(
247 "recommendation_type", "topic_based"
248 )
249 },
250 )
252 return SearchSubscription(
253 user_id=user_id, query=recommended_query, source=source, **kwargs
254 )