Coverage for src / local_deep_research / news / preference_manager / base_preference.py: 26%
70 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1"""
2Base class for preference management.
3Following LDR's pattern from BaseSearchStrategy.
4"""
6from abc import ABC, abstractmethod
7from typing import Dict, Any, List, Optional
8from datetime import timedelta
9from loguru import logger
11from ..core.utils import utc_now
14class BasePreferenceManager(ABC):
15 """Abstract base class for preference management."""
17 def __init__(self, storage_backend: Optional[Any] = None):
18 """
19 Initialize the base preference manager.
21 Args:
22 storage_backend: Optional storage backend for preferences
23 """
24 self.storage_backend = storage_backend
26 @abstractmethod
27 def get_preferences(self, user_id: str) -> Dict[str, Any]:
28 """
29 Get user preferences.
31 Args:
32 user_id: ID of the user
34 Returns:
35 Dictionary of user preferences
36 """
37 pass
39 @abstractmethod
40 def update_preferences(
41 self, user_id: str, preferences: Dict[str, Any]
42 ) -> Dict[str, Any]:
43 """
44 Update user preferences.
46 Args:
47 user_id: ID of the user
48 preferences: Dictionary of preferences to update
50 Returns:
51 Updated preferences
52 """
53 pass
55 def add_interest(
56 self, user_id: str, interest: str, weight: float = 1.0
57 ) -> None:
58 """
59 Add an interest to user preferences.
61 Args:
62 user_id: ID of the user
63 interest: The interest to add
64 weight: Weight/importance of this interest
65 """
66 prefs = self.get_preferences(user_id)
68 if "interests" not in prefs:
69 prefs["interests"] = {}
71 prefs["interests"][interest] = weight
72 prefs["interests_updated_at"] = utc_now().isoformat()
74 self.update_preferences(user_id, prefs)
75 logger.info(f"Added interest '{interest}' for user {user_id}")
77 def remove_interest(self, user_id: str, interest: str) -> None:
78 """
79 Remove an interest from user preferences.
81 Args:
82 user_id: ID of the user
83 interest: The interest to remove
84 """
85 prefs = self.get_preferences(user_id)
87 if "interests" in prefs and interest in prefs["interests"]:
88 del prefs["interests"][interest]
89 prefs["interests_updated_at"] = utc_now().isoformat()
91 self.update_preferences(user_id, prefs)
92 logger.info(f"Removed interest '{interest}' for user {user_id}")
94 def ignore_topic(self, user_id: str, topic: str) -> None:
95 """
96 Add a topic to the ignore list.
98 Args:
99 user_id: ID of the user
100 topic: Topic to ignore
101 """
102 prefs = self.get_preferences(user_id)
104 if "disliked_topics" not in prefs:
105 prefs["disliked_topics"] = []
107 if topic not in prefs["disliked_topics"]:
108 prefs["disliked_topics"].append(topic)
109 prefs["preferences_updated_at"] = utc_now().isoformat()
111 self.update_preferences(user_id, prefs)
112 logger.info(f"Added '{topic}' to ignore list for user {user_id}")
114 def boost_source(
115 self, user_id: str, source: str, weight: float = 1.5
116 ) -> None:
117 """
118 Boost a particular news source.
120 Args:
121 user_id: ID of the user
122 source: Source domain to boost
123 weight: Boost weight
124 """
125 prefs = self.get_preferences(user_id)
127 if "source_weights" not in prefs:
128 prefs["source_weights"] = {}
130 prefs["source_weights"][source] = weight
131 prefs["preferences_updated_at"] = utc_now().isoformat()
133 self.update_preferences(user_id, prefs)
134 logger.info(
135 f"Set source weight for '{source}' to {weight} for user {user_id}"
136 )
138 def get_default_preferences(self) -> Dict[str, Any]:
139 """
140 Get default preferences for new users.
142 Returns:
143 Dictionary of default preferences
144 """
145 return {
146 "liked_categories": [],
147 "disliked_categories": [],
148 "liked_topics": [],
149 "disliked_topics": [],
150 "interests": {},
151 "source_weights": {},
152 "impact_threshold": 5, # Default threshold
153 "focus_preferences": {
154 "surprising": False,
155 "breaking": True,
156 "positive": False,
157 "local": False,
158 },
159 "custom_search_terms": "",
160 "search_strategy": "news_aggregation",
161 "created_at": utc_now().isoformat(),
162 "preferences_updated_at": utc_now().isoformat(),
163 }
166class TopicRegistry:
167 """
168 Registry for dynamically discovered topics.
169 Not user-specific - tracks global topic trends.
170 """
172 def __init__(self, llm_client: Optional[Any] = None):
173 """
174 Initialize topic registry.
176 Args:
177 llm_client: LLM client for topic extraction
178 """
179 self.llm_client = llm_client
180 self.topics: Dict[str, Dict[str, Any]] = {}
182 def extract_topics(self, content: str, max_topics: int = 5) -> List[str]:
183 """
184 Extract topics from content using topic generator.
186 Args:
187 content: Text content to analyze
188 max_topics: Maximum number of topics to extract
190 Returns:
191 List of extracted topics
192 """
193 from ..utils.topic_generator import generate_topics
195 # Use topic generator to extract topics
196 topics = generate_topics(
197 query="", # No specific query, just analyzing content
198 findings=content,
199 category="",
200 max_topics=max_topics,
201 )
203 # Register discovered topics
204 for topic in topics:
205 self.register_topic(topic)
207 return topics
209 def register_topic(self, topic: str) -> None:
210 """
211 Register a discovered topic.
213 Args:
214 topic: Topic to register
215 """
216 if topic not in self.topics:
217 self.topics[topic] = {
218 "first_seen": utc_now(),
219 "count": 0,
220 "last_seen": utc_now(),
221 }
223 self.topics[topic]["count"] += 1
224 self.topics[topic]["last_seen"] = utc_now()
226 def get_trending_topics(
227 self, hours: int = 24, limit: int = 10
228 ) -> List[str]:
229 """
230 Get trending topics from the last N hours.
232 Args:
233 hours: Look back period in hours
234 limit: Maximum number of topics to return
236 Returns:
237 List of trending topic names
238 """
239 cutoff_time = utc_now() - timedelta(hours=hours)
241 # Filter topics seen recently
242 recent_topics = [
243 (topic, data)
244 for topic, data in self.topics.items()
245 if data["last_seen"] >= cutoff_time
246 ]
248 # Sort by count (most frequent first)
249 recent_topics.sort(key=lambda x: x[1]["count"], reverse=True)
251 # Return topic names only
252 return [topic for topic, _ in recent_topics[:limit]]
254 def get_topic_info(self, topic: str) -> Optional[Dict[str, Any]]:
255 """
256 Get information about a specific topic.
258 Args:
259 topic: Topic to look up
261 Returns:
262 Topic information or None if not found
263 """
264 return self.topics.get(topic)