File size: 15,388 Bytes
b30f068 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 | """
API matching system using fuzzy string matching and semantic similarity.
"""
from typing import Dict, List, Any, Optional, Tuple, NamedTuple
from dataclasses import dataclass
from enum import Enum
import json
import re
import concurrent.futures
try:
from rapidfuzz import fuzz, process
RAPIDFUZZ_AVAILABLE = True
except ImportError:
RAPIDFUZZ_AVAILABLE = False
from config.settings import settings
from src.utils.logging_config import logger
class MatchType(Enum):
"""Types of API matches."""
EXACT = "exact"
FUZZY = "fuzzy"
SEMANTIC = "semantic"
PATTERN = "pattern"
@dataclass
class APIOperation:
"""Represents an API operation."""
name: str
method: str
endpoint: str
description: str
parameters: List[Dict[str, Any]]
tags: List[str]
summary: Optional[str] = None
operation_id: Optional[str] = None
@dataclass
class APIMatch:
"""Represents a match between keywords and API operations."""
keyword: str
operation: APIOperation
match_type: MatchType
similarity_score: float
confidence: float
reasoning: str
class APIMatcher:
"""Matches extracted keywords to available API operations."""
def __init__(self, config: Optional[Dict[str, Any]] = None):
"""Initialize the API matcher."""
self.config = config or {}
self.operations: List[APIOperation] = []
self.operation_index: Dict[str, APIOperation] = {}
# Configuration
self.fuzzy_threshold = self.config.get('fuzzy_threshold', settings.fuzzy_match_threshold)
self.similarity_threshold = self.config.get('similarity_threshold', settings.similarity_threshold)
self.max_matches = self.config.get('max_matches', 10)
if not RAPIDFUZZ_AVAILABLE:
logger.warning("RapidFuzz not available. Fuzzy matching will be limited.")
logger.info("Initialized APIMatcher")
def load_openapi_spec(self, spec: Dict[str, Any]) -> None:
"""
Load API operations from OpenAPI specification.
Args:
spec: OpenAPI specification dictionary
"""
operations = []
try:
paths = spec.get('paths', {})
for path, path_item in paths.items():
for method, operation in path_item.items():
if method.upper() in ['GET', 'POST', 'PUT', 'DELETE', 'PATCH']:
# Extract operation details
op = APIOperation(
name=operation.get('operationId', f"{method}_{path.replace('/', '_')}"),
method=method.upper(),
endpoint=path,
description=operation.get('description', ''),
summary=operation.get('summary', ''),
parameters=operation.get('parameters', []),
tags=operation.get('tags', []),
operation_id=operation.get('operationId')
)
operations.append(op)
self.operations = operations
self._build_operation_index()
logger.info(f"Loaded {len(operations)} API operations from OpenAPI spec")
except Exception as e:
logger.error(f"Error loading OpenAPI spec: {e}")
raise
def add_operation(self, operation: APIOperation) -> None:
"""Add a single API operation."""
self.operations.append(operation)
self.operation_index[operation.name] = operation
logger.debug(f"Added operation: {operation.name}")
def add_operations(self, operations: List[APIOperation]) -> None:
"""Add multiple API operations."""
self.operations.extend(operations)
self._build_operation_index()
logger.info(f"Added {len(operations)} operations")
def _build_operation_index(self) -> None:
"""Build index for fast operation lookup."""
self.operation_index = {op.name: op for op in self.operations}
def match_keywords(self, keywords: List[str]) -> List[APIMatch]:
"""
Match keywords to API operations.
Args:
keywords: List of extracted keywords
Returns:
List of API matches sorted by confidence
"""
if not self.operations:
logger.warning("No API operations loaded")
return []
all_matches = []
# Match each keyword against all operations in parallel
with concurrent.futures.ThreadPoolExecutor() as executor:
for matches in executor.map(self._match_single_keyword, keywords):
all_matches.extend(matches)
# Remove duplicates and sort by confidence
unique_matches = self._deduplicate_matches(all_matches)
sorted_matches = sorted(unique_matches, key=lambda x: x.confidence, reverse=True)
return sorted_matches[:self.max_matches]
def _match_single_keyword(self, keyword: str) -> List[APIMatch]:
"""Match a single keyword against all operations."""
matches = []
for operation in self.operations:
# Try different matching strategies
match_results = [
self._exact_match(keyword, operation),
self._fuzzy_match(keyword, operation),
self._pattern_match(keyword, operation),
self._semantic_match(keyword, operation)
]
# Keep the best match for this operation
best_match = max(match_results, key=lambda x: x.confidence if x else 0)
if best_match and best_match.confidence > 0.3: # Minimum confidence threshold
matches.append(best_match)
return matches
def _exact_match(self, keyword: str, operation: APIOperation) -> Optional[APIMatch]:
"""Check for exact matches."""
keyword_lower = keyword.lower()
# Check operation name
if keyword_lower == operation.name.lower():
return APIMatch(
keyword=keyword,
operation=operation,
match_type=MatchType.EXACT,
similarity_score=1.0,
confidence=1.0,
reasoning="Exact match with operation name"
)
# Check tags
for tag in operation.tags:
if keyword_lower == tag.lower():
return APIMatch(
keyword=keyword,
operation=operation,
match_type=MatchType.EXACT,
similarity_score=1.0,
confidence=0.9,
reasoning=f"Exact match with tag: {tag}"
)
# Check if keyword appears in description
if keyword_lower in operation.description.lower():
return APIMatch(
keyword=keyword,
operation=operation,
match_type=MatchType.EXACT,
similarity_score=1.0,
confidence=0.8,
reasoning="Exact match in description"
)
return None
def _fuzzy_match(self, keyword: str, operation: APIOperation) -> Optional[APIMatch]:
"""Perform fuzzy string matching."""
if not RAPIDFUZZ_AVAILABLE:
return None
# Prepare search targets
targets = [
operation.name,
operation.summary or "",
operation.description,
" ".join(operation.tags),
operation.endpoint
]
best_score = 0
best_target = ""
for target in targets:
if target:
score = fuzz.WRatio(keyword.lower(), target.lower())
if score > best_score:
best_score = score
best_target = target
if best_score >= self.fuzzy_threshold:
confidence = min(0.9, best_score / 100.0)
return APIMatch(
keyword=keyword,
operation=operation,
match_type=MatchType.FUZZY,
similarity_score=best_score / 100.0,
confidence=confidence,
reasoning=f"Fuzzy match with '{best_target}' (score: {best_score})"
)
return None
def _pattern_match(self, keyword: str, operation: APIOperation) -> Optional[APIMatch]:
"""Match using patterns and heuristics."""
keyword_lower = keyword.lower()
# HTTP method patterns
method_patterns = {
'get': ['get', 'fetch', 'retrieve', 'find', 'search', 'list'],
'post': ['create', 'add', 'new', 'insert', 'submit'],
'put': ['update', 'modify', 'change', 'edit', 'replace'],
'delete': ['delete', 'remove', 'destroy', 'drop'],
'patch': ['patch', 'partial', 'modify']
}
# Check if keyword matches operation method pattern
for method, patterns in method_patterns.items():
if operation.method.lower() == method and keyword_lower in patterns:
return APIMatch(
keyword=keyword,
operation=operation,
match_type=MatchType.PATTERN,
similarity_score=0.8,
confidence=0.7,
reasoning=f"Pattern match: '{keyword}' suggests {method.upper()} operation"
)
# Resource name patterns
endpoint_parts = [part for part in operation.endpoint.split('/') if part and not part.startswith('{')]
for part in endpoint_parts:
if keyword_lower in part.lower() or part.lower() in keyword_lower:
return APIMatch(
keyword=keyword,
operation=operation,
match_type=MatchType.PATTERN,
similarity_score=0.7,
confidence=0.6,
reasoning=f"Pattern match with endpoint resource: {part}"
)
return None
def _semantic_match(self, keyword: str, operation: APIOperation) -> Optional[APIMatch]:
"""Perform semantic matching (placeholder for now)."""
# This would use embeddings/transformers for semantic similarity
# For now, implement basic word overlap
keyword_words = set(keyword.lower().split())
# Combine operation text
operation_text = " ".join([
operation.name,
operation.summary or "",
operation.description,
" ".join(operation.tags)
]).lower()
operation_words = set(operation_text.split())
# Calculate Jaccard similarity
intersection = keyword_words.intersection(operation_words)
union = keyword_words.union(operation_words)
if union:
similarity = len(intersection) / len(union)
if similarity > 0.2: # Minimum semantic similarity
return APIMatch(
keyword=keyword,
operation=operation,
match_type=MatchType.SEMANTIC,
similarity_score=similarity,
confidence=similarity * 0.6, # Lower confidence for basic semantic matching
reasoning=f"Semantic similarity based on word overlap: {intersection}"
)
return None
def _deduplicate_matches(self, matches: List[APIMatch]) -> List[APIMatch]:
"""Remove duplicate matches, keeping the best one for each operation."""
operation_matches = {}
for match in matches:
op_key = f"{match.operation.method}_{match.operation.endpoint}"
if op_key not in operation_matches or match.confidence > operation_matches[op_key].confidence:
operation_matches[op_key] = match
return list(operation_matches.values())
def get_operation_by_name(self, name: str) -> Optional[APIOperation]:
"""Get operation by name."""
return self.operation_index.get(name)
def get_operations_by_tag(self, tag: str) -> List[APIOperation]:
"""Get operations by tag."""
return [op for op in self.operations if tag.lower() in [t.lower() for t in op.tags]]
def get_operations_by_method(self, method: str) -> List[APIOperation]:
"""Get operations by HTTP method."""
return [op for op in self.operations if op.method.upper() == method.upper()]
def search_operations(self, query: str) -> List[APIOperation]:
"""Search operations by query string."""
if not RAPIDFUZZ_AVAILABLE:
# Fallback to simple text search
query_lower = query.lower()
results = []
for op in self.operations:
search_text = f"{op.name} {op.description} {' '.join(op.tags)}".lower()
if query_lower in search_text:
results.append(op)
return results
# Use fuzzy search
search_targets = []
for op in self.operations:
search_text = f"{op.name} {op.description} {' '.join(op.tags)}"
search_targets.append((search_text, op))
matches = process.extract(
query,
[target[0] for target in search_targets],
scorer=fuzz.WRatio,
limit=10
)
results = []
for match, score, _ in matches:
if score >= 60: # Minimum score for search results
# Find corresponding operation
for text, op in search_targets:
if text == match:
results.append(op)
break
return results
def get_stats(self) -> Dict[str, Any]:
"""Get matcher statistics."""
method_counts = {}
tag_counts = {}
for op in self.operations:
method_counts[op.method] = method_counts.get(op.method, 0) + 1
for tag in op.tags:
tag_counts[tag] = tag_counts.get(tag, 0) + 1
return {
'total_operations': len(self.operations),
'methods': method_counts,
'tags': tag_counts,
'fuzzy_matching_available': RAPIDFUZZ_AVAILABLE,
'configuration': {
'fuzzy_threshold': self.fuzzy_threshold,
'similarity_threshold': self.similarity_threshold,
'max_matches': self.max_matches
}
}
|