Coverage for src / local_deep_research / news / core / relevance_service.py: 100%

1""" 

2Relevance and trending calculation service. 

3Handles business logic for calculating relevance scores and trending metrics. 

4""" 

5 

6from typing import Optional, Dict, List 

7from .base_card import BaseCard 

8 

9 

10class RelevanceService: 

11 """Service for calculating relevance and trending scores.""" 

12 

13 def calculate_relevance( 

14 self, card: BaseCard, user_prefs: Optional[Dict] 

15 ) -> float: 

16 """ 

17 Calculate relevance score for a card based on user preferences. 

18 

19 Args: 

20 card: The card to score 

21 user_prefs: User preferences dictionary 

22 

23 Returns: 

24 Relevance score between 0 and 1 

25 """ 

26 if not user_prefs: 

27 # Default relevance based on impact 

28 return float(getattr(card, "impact_score", 5)) / 10.0 

29 

30 score = 0.5 # Base score 

31 

32 # Category preference 

33 if hasattr(card, "category"): 

34 if card.category in user_prefs.get("liked_categories", []): 

35 score += 0.2 

36 elif card.category in user_prefs.get("disliked_categories", []): 

37 score -= 0.2 

38 

39 # Impact threshold 

40 if hasattr(card, "impact_score"): 

41 threshold = user_prefs.get("impact_threshold", 5) 

42 if card.impact_score >= threshold: 

43 score += 0.1 

44 else: 

45 score -= 0.1 

46 

47 # Topic matching (simplified without embeddings) 

48 if hasattr(card, "topics"): 

49 liked_topics = user_prefs.get("liked_topics", []) 

50 for topic in card.topics: 

51 if any(liked in topic.lower() for liked in liked_topics): 

52 score += 0.1 

53 break 

54 

55 # Ensure score is in valid range 

56 return max(0.0, min(1.0, score)) 

57 

58 def calculate_trending_score(self, card: BaseCard) -> float: 

59 """ 

60 Calculate trending score based on impact and engagement. 

61 

62 Args: 

63 card: The card to score 

64 

65 Returns: 

66 Trending score 

67 """ 

68 if not hasattr(card, "impact_score"): 

69 return 0.0 

70 

71 # Calculate engagement 

72 engagement = ( 

73 card.interaction.get("views", 0) 

74 + card.interaction.get("votes_up", 0) * 2 

75 - card.interaction.get("votes_down", 0) 

76 ) 

77 

78 # Combine impact and engagement 

79 trending_score: float = float(getattr(card, "impact_score", 0)) + ( 

80 engagement / 10 

81 ) 

82 

83 return trending_score 

84 

85 def filter_trending( 

86 self, cards: List[BaseCard], min_impact: int = 7, limit: int = 10 

87 ) -> List[BaseCard]: 

88 """ 

89 Filter and sort cards by trending score. 

90 

91 Args: 

92 cards: List of cards to filter 

93 min_impact: Minimum impact score required 

94 limit: Maximum number of cards to return 

95 

96 Returns: 

97 List of trending cards sorted by score 

98 """ 

99 trending = [] 

100 

101 for card in cards: 

102 if ( 

103 hasattr(card, "impact_score") 

104 and card.impact_score >= min_impact 

105 ): 

106 setattr( 

107 card, "trending_score", self.calculate_trending_score(card) 

108 ) 

109 trending.append(card) 

110 

111 # Sort by trending score 

112 trending.sort( 

113 key=lambda c: getattr(c, "trending_score", 0.0), reverse=True 

114 ) 

115 

116 return trending[:limit] 

117 

118 def personalize_feed( 

119 self, 

120 cards: List[BaseCard], 

121 user_prefs: Optional[Dict], 

122 include_seen: bool = True, 

123 ) -> List[BaseCard]: 

124 """ 

125 Personalize a feed of cards based on user preferences. 

126 

127 Args: 

128 cards: List of cards to personalize 

129 user_prefs: User preferences 

130 include_seen: Whether to include already viewed cards 

131 

132 Returns: 

133 Personalized list of cards 

134 """ 

135 personalized = [] 

136 

137 for card in cards: 

138 # Calculate relevance 

139 setattr( 

140 card, 

141 "relevance_score", 

142 self.calculate_relevance(card, user_prefs), 

143 ) 

144 

145 # Filter seen if requested 

146 if not include_seen and card.interaction.get("viewed"): 

147 continue 

148 

149 personalized.append(card) 

150 

151 # Sort by relevance 

152 personalized.sort( 

153 key=lambda c: getattr(c, "relevance_score", 0.0), reverse=True 

154 ) 

155 

156 return personalized 

157 

158 

159# Singleton instance 

160_relevance_service = None 

161 

162 

163def get_relevance_service() -> RelevanceService: 

164 """Get or create the global RelevanceService instance.""" 

165 global _relevance_service 

166 if _relevance_service is None: 

167 _relevance_service = RelevanceService() 

168 return _relevance_service