Coverage for src / local_deep_research / news / rating_system / base_rater.py: 41%
77 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1"""
2Base class for rating systems.
3Following LDR's pattern from BaseSearchStrategy.
5NOTE: The news rating system is intentionally separate from research session ratings.
6News ratings serve a different purpose - they are used for:
71. Recommending new topics to users based on their preferences
82. Improving the news recommendation algorithm
93. Understanding which types of news content are most valuable to users
11This is distinct from research session ratings which evaluate the quality of
12research output. As per PR review discussion with djpetti.
13"""
15from abc import ABC, abstractmethod
16from enum import Enum
17from typing import List, Dict, Any, Optional
18from loguru import logger
20from ..core.utils import utc_now
23class RelevanceRating(Enum):
24 """Enum for relevance rating values."""
26 UP = "up"
27 DOWN = "down"
30class QualityRating(Enum):
31 """Enum for quality rating values."""
33 ONE_STAR = 1
34 TWO_STARS = 2
35 THREE_STARS = 3
36 FOUR_STARS = 4
37 FIVE_STARS = 5
40class BaseRatingSystem(ABC):
41 """Abstract base class for all rating systems."""
43 def __init__(self, storage_backend: Optional[Any] = None):
44 """
45 Initialize the base rating system.
47 Args:
48 storage_backend: Optional storage backend for ratings
49 """
50 self.storage_backend = storage_backend
51 self.rating_type = self.__class__.__name__
53 @abstractmethod
54 def rate(
55 self,
56 user_id: str,
57 card_id: str,
58 rating_value: Any,
59 metadata: Optional[Dict[str, Any]] = None,
60 ) -> Dict[str, Any]:
61 """
62 Record a rating from a user.
64 Args:
65 user_id: ID of the user making the rating
66 card_id: ID of the card being rated
67 rating_value: The rating value (type depends on implementation)
68 metadata: Optional additional metadata
70 Returns:
71 Dict containing rating confirmation and any updates
72 """
73 pass
75 @abstractmethod
76 def get_rating(
77 self, user_id: str, card_id: str
78 ) -> Optional[Dict[str, Any]]:
79 """
80 Get a user's rating for a specific card.
82 Args:
83 user_id: ID of the user
84 card_id: ID of the card
86 Returns:
87 Rating information or None if not rated
88 """
89 pass
91 @abstractmethod
92 def get_rating_type(self) -> str:
93 """
94 Get the type of rating this system handles.
96 Returns:
97 str: Rating type identifier
98 """
99 pass
101 def get_recent_ratings(
102 self, user_id: str, limit: int = 50
103 ) -> List[Dict[str, Any]]:
104 """
105 Get recent ratings by a user.
107 Args:
108 user_id: ID of the user
109 limit: Maximum number of ratings to return
111 Returns:
112 List of recent ratings
113 """
114 # Default implementation - should be overridden if storage is available
115 logger.warning(
116 f"get_recent_ratings not implemented for {self.rating_type}"
117 )
118 return []
120 def get_card_ratings(
121 self, card_id: str, rating_type: Optional[str] = None
122 ) -> Dict[str, Any]:
123 """
124 Get aggregated ratings for a card.
126 Args:
127 card_id: ID of the card
128 rating_type: Optional specific rating type to filter
130 Returns:
131 Aggregated rating information
132 """
133 # Default implementation - should be overridden
134 logger.warning(
135 f"get_card_ratings not implemented for {self.rating_type}"
136 )
137 return {"total": 0, "average": None}
139 def remove_rating(self, user_id: str, card_id: str) -> bool:
140 """
141 Remove a user's rating for a card.
143 Args:
144 user_id: ID of the user
145 card_id: ID of the card
147 Returns:
148 bool: True if rating was removed
149 """
150 # Default implementation
151 logger.warning(f"remove_rating not implemented for {self.rating_type}")
152 return False
154 def _create_rating_record(
155 self,
156 user_id: str,
157 card_id: str,
158 rating_value: Any,
159 metadata: Optional[Dict[str, Any]] = None,
160 ) -> Dict[str, Any]:
161 """
162 Create a standard rating record.
164 Helper method for subclasses to create consistent rating records.
165 """
166 return {
167 "user_id": user_id,
168 "card_id": card_id,
169 "rating_type": self.get_rating_type(),
170 "rating_value": rating_value,
171 "rated_at": utc_now().isoformat(),
172 "metadata": metadata or {},
173 }
175 def _validate_rating_value(self, rating_value: Any) -> None:
176 """
177 Validate that a rating value is acceptable.
178 Should be overridden by subclasses with specific validation.
180 Args:
181 rating_value: The value to validate
183 Raises:
184 ValueError: If the rating value is invalid
185 """
186 if rating_value is None:
187 raise ValueError("Rating value cannot be None")
190class QualityRatingSystem(BaseRatingSystem):
191 """Rating system for article quality (1-5 stars)."""
193 def get_rating_type(self) -> str:
194 return "quality"
196 def rate(
197 self,
198 user_id: str,
199 card_id: str,
200 rating_value: QualityRating,
201 metadata: Optional[Dict[str, Any]] = None,
202 ) -> Dict[str, Any]:
203 """
204 Rate the quality of an article.
206 Args:
207 rating_value: QualityRating enum value (ONE_STAR through FIVE_STARS)
208 """
209 self._validate_rating_value(rating_value)
211 record = self._create_rating_record(
212 user_id, card_id, rating_value, metadata
213 )
215 # Store the rating (implementation depends on storage backend)
216 if self.storage_backend:
217 # Storage implementation would go here
218 pass
220 logger.info(
221 f"User {user_id} rated card {card_id} quality: {rating_value.value} stars"
222 )
224 return {
225 "success": True,
226 "rating": record,
227 "message": f"Quality rating of {rating_value.value} stars recorded",
228 }
230 def _validate_rating_value(self, rating_value: Any) -> None:
231 """Validate that rating is a valid QualityRating enum value."""
232 super()._validate_rating_value(rating_value)
234 if not isinstance(rating_value, QualityRating):
235 raise ValueError(
236 "Quality rating must be a QualityRating enum value (ONE_STAR through FIVE_STARS)"
237 )
239 def get_rating(
240 self, user_id: str, card_id: str
241 ) -> Optional[Dict[str, Any]]:
242 """Get a user's quality rating for a card."""
243 # Implementation depends on storage backend
244 if self.storage_backend:
245 # Storage query would go here
246 pass
247 return None
250class RelevanceRatingSystem(BaseRatingSystem):
251 """Rating system for personal relevance (thumbs up/down)."""
253 def get_rating_type(self) -> str:
254 return "relevance"
256 def rate(
257 self,
258 user_id: str,
259 card_id: str,
260 rating_value: RelevanceRating,
261 metadata: Optional[Dict[str, Any]] = None,
262 ) -> Dict[str, Any]:
263 """
264 Rate the relevance of an article.
266 Args:
267 rating_value: RelevanceRating.UP or RelevanceRating.DOWN
268 """
269 self._validate_rating_value(rating_value)
271 record = self._create_rating_record(
272 user_id, card_id, rating_value, metadata
273 )
275 # Store the rating (implementation depends on storage backend)
276 if self.storage_backend:
277 # Storage implementation would go here
278 pass
280 logger.info(
281 f"User {user_id} rated card {card_id} relevance: thumbs {rating_value.value}"
282 )
284 return {
285 "success": True,
286 "rating": record,
287 "message": f"Relevance rating of thumbs {rating_value.value} recorded",
288 }
290 def _validate_rating_value(self, rating_value: Any) -> None:
291 """Validate that rating is a valid RelevanceRating enum value."""
292 super()._validate_rating_value(rating_value)
294 if not isinstance(rating_value, RelevanceRating):
295 raise ValueError(
296 "Relevance rating must be RelevanceRating.UP or RelevanceRating.DOWN"
297 )
299 def get_rating(
300 self, user_id: str, card_id: str
301 ) -> Optional[Dict[str, Any]]:
302 """Get a user's relevance rating for a card."""
303 # Implementation depends on storage backend
304 if self.storage_backend:
305 # Storage query would go here
306 pass
307 return None