Coverage for src/qdrant_loader_mcp_server/search/enhanced/cdi/legacy_adapters.py: 67%

1from __future__ import annotations 

2 

3import re 

4from typing import Any 

5 

6from ....utils.logging import LoggingConfig 

7 

8logger = LoggingConfig.get_logger(__name__) 

9 

10 

11class LegacyConflictDetectorAdapter: 

12 """Adapter that hosts legacy/test-specific compatibility methods. 

13 

14 This wraps a current `ConflictDetector` instance and provides 

15 the old compatibility helpers used by tests without bloating 

16 the main detector implementation. 

17 """ 

18 

19 def __init__(self, detector: Any): 

20 self.detector = detector 

21 self.logger = LoggingConfig.get_logger(__name__) 

22 

23 def _find_contradiction_patterns(self, doc1, doc2): 

24 try: 

25 text1 = getattr(doc1, "text", getattr(doc1, "content", "")) 

26 text2 = getattr(doc2, "text", getattr(doc2, "content", "")) 

27 

28 patterns = [] 

29 

30 # Version conflicts 

31 version_pattern = r"version\s+(\d+\.\d+\.\d+)" 

32 versions1 = re.findall(version_pattern, text1.lower()) 

33 versions2 = re.findall(version_pattern, text2.lower()) 

34 

35 if versions1 and versions2 and versions1 != versions2: 

36 patterns.append( 

37 { 

38 "type": "version_conflict", 

39 "reason": f"Version mismatch: {versions1[0]} vs {versions2[0]}", 

40 "confidence": 0.8, 

41 } 

42 ) 

43 

44 # Procedural conflicts 

45 conflict_indicators = [ 

46 ("should not", "should"), 

47 ("avoid", "use"), 

48 ("deprecated", "recommended"), 

49 ("wrong", "correct"), 

50 ("never", "always"), 

51 ("must not", "must"), 

52 ("don't", "do"), 

53 ] 

54 

55 for negative, positive in conflict_indicators: 

56 if (negative in text1.lower() and positive in text2.lower()) or ( 

57 negative in text2.lower() and positive in text1.lower() 

58 ): 

59 patterns.append( 

60 { 

61 "type": "procedural_conflict", 

62 "reason": f"Conflicting advice: '{negative}' vs '{positive}'", 

63 "confidence": 0.8, 

64 } 

65 ) 

66 

67 return patterns 

68 except Exception as e: 

69 self.logger.warning( 

70 f"Error in compatibility method _find_contradiction_patterns: {e}" 

71 ) 

72 return [] 

73 

74 def _detect_version_conflicts(self, doc1, doc2): 

75 try: 

76 text1 = getattr(doc1, "text", getattr(doc1, "content", "")) 

77 text2 = getattr(doc2, "text", getattr(doc2, "content", "")) 

78 

79 import re 

80 

81 version_pattern = r"(?:python|node|java|version)\s*(\d+\.\d+(?:\.\d+)?)" 

82 versions1 = re.findall(version_pattern, text1.lower()) 

83 versions2 = re.findall(version_pattern, text2.lower()) 

84 

85 if versions1 and versions2: 

86 for v1 in versions1: 

87 for v2 in versions2: 

88 if v1 != v2: 

89 return [ 

90 { 

91 "type": "version_conflict", 

92 "reason": f"Version mismatch: {v1} vs {v2}", 

93 "summary": f"Version mismatch: {v1} vs {v2}", 

94 "confidence": 0.8, 

95 } 

96 ] 

97 

98 # Fallback to metadata analysis from detector 

99 try: 

100 has_conflict, reason, confidence = ( 

101 self.detector._analyze_metadata_conflicts(doc1, doc2) 

102 ) 

103 if ( 

104 has_conflict 

105 and isinstance(reason, str) 

106 and "version" in reason.lower() 

107 ): 

108 return [ 

109 { 

110 "type": "version_conflict", 

111 "reason": reason, 

112 "confidence": confidence, 

113 } 

114 ] 

115 except Exception as exc: 

116 self.logger.exception( 

117 "Metadata conflict analysis failed during version conflict detection", 

118 exc_info=exc, 

119 ) 

120 return [] 

121 except Exception as e: 

122 self.logger.warning(f"Error in version conflict detection: {e}") 

123 return [] 

124 

125 def _detect_procedural_conflicts(self, doc1, doc2): 

126 try: 

127 text1 = getattr(doc1, "text", getattr(doc1, "content", "")) 

128 text2 = getattr(doc2, "text", getattr(doc2, "content", "")) 

129 

130 procedural_conflicts = [ 

131 ("should", "should not"), 

132 ("must", "must not"), 

133 ("manually", "automated"), 

134 ("always", "never"), 

135 ] 

136 

137 for positive, negative in procedural_conflicts: 

138 if (positive in text1.lower() and negative in text2.lower()) or ( 

139 negative in text1.lower() and positive in text2.lower() 

140 ): 

141 return [ 

142 { 

143 "type": "procedural_conflict", 

144 "reason": f"Conflicting procedures: '{positive}' vs '{negative}'", 

145 "confidence": 0.8, 

146 } 

147 ] 

148 

149 # Fallback to text analysis if available 

150 try: 

151 has_conflict, reason, confidence = ( 

152 self.detector._analyze_text_conflicts(doc1, doc2) 

153 ) 

154 if has_conflict and any( 

155 k in str(reason).lower() 

156 for k in ["procedure", "process", "steps", "workflow"] 

157 ): 

158 return [ 

159 { 

160 "type": "procedural_conflict", 

161 "reason": reason, 

162 "confidence": confidence, 

163 } 

164 ] 

165 except Exception as exc: 

166 self.logger.exception( 

167 "Text conflict analysis failed during procedural conflict detection", 

168 exc_info=exc, 

169 ) 

170 

171 return [] 

172 except Exception as e: 

173 self.logger.warning(f"Error in procedural conflict detection: {e}") 

174 return [] 

175 

176 

177__all__ = [ 

178 "LegacyConflictDetectorAdapter", 

179]