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

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 getattr(card, "impact_score", 5) / 10.0 

29 

30 score = 0.5 # Base score 

31 

32 # Category preference 

33 if hasattr(card, "category"): 33 ↛ 40line 33 didn't jump to line 40 because the condition on line 33 was always true

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"): 40 ↛ 48line 40 didn't jump to line 48 because the condition on line 40 was always true

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

42 if card.impact_score >= threshold: 42 ↛ 45line 42 didn't jump to line 45 because the condition on line 42 was always true

43 score += 0.1 

44 else: 

45 score -= 0.1 

46 

47 # Topic matching (simplified without embeddings) 

48 if hasattr(card, "topics"): 48 ↛ 56line 48 didn't jump to line 56 because the condition on line 48 was always true

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): 51 ↛ 50line 51 didn't jump to line 50 because the condition on line 51 was always true

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 = card.impact_score + (engagement / 10) 

80 

81 return trending_score 

82 

83 def filter_trending( 

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

85 ) -> List[BaseCard]: 

86 """ 

87 Filter and sort cards by trending score. 

88 

89 Args: 

90 cards: List of cards to filter 

91 min_impact: Minimum impact score required 

92 limit: Maximum number of cards to return 

93 

94 Returns: 

95 List of trending cards sorted by score 

96 """ 

97 trending = [] 

98 

99 for card in cards: 

100 if ( 

101 hasattr(card, "impact_score") 

102 and card.impact_score >= min_impact 

103 ): 

104 card.trending_score = self.calculate_trending_score(card) 

105 trending.append(card) 

106 

107 # Sort by trending score 

108 trending.sort(key=lambda c: c.trending_score, reverse=True) 

109 

110 return trending[:limit] 

111 

112 def personalize_feed( 

113 self, 

114 cards: List[BaseCard], 

115 user_prefs: Optional[Dict], 

116 include_seen: bool = True, 

117 ) -> List[BaseCard]: 

118 """ 

119 Personalize a feed of cards based on user preferences. 

120 

121 Args: 

122 cards: List of cards to personalize 

123 user_prefs: User preferences 

124 include_seen: Whether to include already viewed cards 

125 

126 Returns: 

127 Personalized list of cards 

128 """ 

129 personalized = [] 

130 

131 for card in cards: 

132 # Calculate relevance 

133 card.relevance_score = self.calculate_relevance(card, user_prefs) 

134 

135 # Filter seen if requested 

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

137 continue 

138 

139 personalized.append(card) 

140 

141 # Sort by relevance 

142 personalized.sort(key=lambda c: c.relevance_score, reverse=True) 

143 

144 return personalized 

145 

146 

147# Singleton instance 

148_relevance_service = None 

149 

150 

151def get_relevance_service() -> RelevanceService: 

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

153 global _relevance_service 

154 if _relevance_service is None: 

155 _relevance_service = RelevanceService() 

156 return _relevance_service