Coverage for src / local_deep_research / news / core / search_integration.py: 40%

1""" 

2Integration with LDR's search system to track searches and manage priorities. 

3""" 

4 

5import uuid 

6from typing import Optional, Dict, Any, Callable 

7from datetime import datetime, timezone 

8from loguru import logger 

9 

10 

11class NewsSearchCallback: 

12 """ 

13 Callback handler for search system integration. 

14 Tracks searches for news personalization. 

15 """ 

16 

17 def __init__(self): 

18 self._tracking_enabled = None 

19 

20 @property 

21 def tracking_enabled(self) -> bool: 

22 """Check if search tracking is enabled.""" 

23 if self._tracking_enabled is None: 23 ↛ 26line 23 didn't jump to line 26 because the condition on line 23 was always true

24 # TODO: Per-user settings will be handled later 

25 self._tracking_enabled = False # Default: news.search_tracking 

26 return self._tracking_enabled 

27 

28 def __call__( 

29 self, 

30 query: str, 

31 result: Dict[str, Any], 

32 context: Optional[Dict[str, Any]] = None, 

33 ) -> None: 

34 """ 

35 Process a search completion. 

36 

37 Args: 

38 query: The search query 

39 result: The search results 

40 context: Optional context (includes is_user_search, user_id, etc.) 

41 """ 

42 

43 # Extract context 

44 context = context or {} 

45 is_user_search = context.get("is_user_search", True) 

46 user_id = context.get("user_id", "anonymous") 

47 search_id = context.get("search_id", str(uuid.uuid4())) 

48 

49 # Log the search if tracking is enabled 

50 if is_user_search and self.tracking_enabled: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true

51 self._track_user_search( 

52 search_id=search_id, user_id=user_id, query=query, result=result 

53 ) 

54 

55 def _track_user_search( 

56 self, search_id: str, user_id: str, query: str, result: Dict[str, Any] 

57 ) -> None: 

58 """ 

59 Track a user search for personalization. 

60 

61 Args: 

62 search_id: Unique search ID 

63 user_id: User who performed the search 

64 query: The search query 

65 result: The search results 

66 """ 

67 try: 

68 # Import here to avoid circular imports 

69 from ..preference_manager.search_tracker import SearchTracker 

70 

71 tracker = SearchTracker() 

72 tracker.track_search( 

73 user_id=user_id, 

74 query=query, 

75 search_id=search_id, 

76 result_quality=self._calculate_quality(result), 

77 result_count=len(result.get("findings", [])), 

78 strategy_used=result.get("strategy", "unknown"), 

79 ) 

80 

81 logger.debug(f"Tracked user search: {query[:50]}...") 

82 

83 except Exception: 

84 logger.exception("Error tracking search") 

85 

86 def _calculate_quality(self, result: Dict[str, Any]) -> float: 

87 """ 

88 Calculate quality score for search results. 

89 

90 Args: 

91 result: Search results 

92 

93 Returns: 

94 Quality score between 0 and 1 

95 """ 

96 # Simple heuristic based on result count and content 

97 findings = result.get("findings", []) 

98 if not findings: 

99 return 0.0 

100 

101 # More findings generally means better results 

102 count_score = min(len(findings) / 10, 1.0) 

103 

104 # Check if we have actual content 

105 has_content = any(f.get("content") for f in findings[:5]) 

106 content_score = 1.0 if has_content else 0.5 

107 

108 return (count_score + content_score) / 2 

109 

110 

111def create_search_wrapper(original_search_method: Callable) -> Callable: 

112 """ 

113 Create a wrapper for search methods that integrates news tracking. 

114 

115 Args: 

116 original_search_method: The original search method to wrap 

117 

118 Returns: 

119 Wrapped method with news tracking 

120 """ 

121 callback = NewsSearchCallback() 

122 

123 def wrapped_search(self, query: str, **kwargs) -> Dict[str, Any]: 

124 """Wrapped search with news tracking.""" 

125 

126 # Determine if this is a user search 

127 is_user_search = kwargs.pop("is_user_search", True) 

128 is_news_search = kwargs.pop("is_news_search", False) 

129 user_id = kwargs.pop("user_id", "anonymous") 

130 

131 # Generate search ID 

132 search_id = str(uuid.uuid4()) 

133 

134 # Build context 

135 context = { 

136 "is_user_search": is_user_search and not is_news_search, 

137 "is_news_search": is_news_search, 

138 "user_id": user_id, 

139 "search_id": search_id, 

140 "timestamp": datetime.now(timezone.utc), 

141 } 

142 

143 # Perform the search 

144 result = original_search_method(self, query, **kwargs) 

145 

146 # Call callback if available 

147 try: 

148 callback(query, result, context) 

149 except Exception: 

150 logger.exception("Error in news callback") 

151 

152 return result 

153 

154 # Preserve method metadata 

155 wrapped_search.__name__ = original_search_method.__name__ 

156 wrapped_search.__doc__ = original_search_method.__doc__ 

157 

158 return wrapped_search