feat(4.1-03,4.1-04): implement search orchestrator and integration tests
This commit is contained in:
150
backend/services/spare_parts_search.py
Normal file
150
backend/services/spare_parts_search.py
Normal file
@@ -0,0 +1,150 @@
|
||||
"""
|
||||
Search orchestrator for spare-parts web discovery and specification extraction.
|
||||
|
||||
Coordinates web scraping, rate limiting, and spec extraction to find and extract
|
||||
product information for spare-parts item onboarding.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
from .web_scraper import SearchRateLimiter, search_google, search_bing
|
||||
from .spec_extractor import extract_specs_from_multiple_results
|
||||
from backend.ai.spare_parts_whitelist import classify_as_spare_part, get_spare_part_type
|
||||
|
||||
log = logging.getLogger("ainventory")
|
||||
|
||||
# Global rate limiter (1 request per 5 seconds)
|
||||
_rate_limiter = SearchRateLimiter(requests_per_second=0.2)
|
||||
|
||||
|
||||
async def search_spare_parts(
|
||||
category: str,
|
||||
part_number: Optional[str] = None,
|
||||
item_name: Optional[str] = None,
|
||||
timeout: int = 30
|
||||
) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Search for spare-parts information using web scraping with fallback.
|
||||
|
||||
Attempts Google search first, falls back to Bing on error. Returns None on
|
||||
timeout/failure, allowing graceful degradation to AI-only data.
|
||||
|
||||
Args:
|
||||
category: Item category from AI extraction (e.g., "DDR4", "SSD", "CPU")
|
||||
part_number: Part number if available
|
||||
item_name: Item name/description from AI extraction
|
||||
timeout: Total search timeout in seconds (default 30)
|
||||
|
||||
Returns:
|
||||
Dict with keys: category, type, description, notes
|
||||
Or None if search fails/times out (graceful fallback)
|
||||
|
||||
Example:
|
||||
>>> result = await search_spare_parts(
|
||||
... category="Kingston DDR4 16GB",
|
||||
... part_number="KF466C40RS-16"
|
||||
... )
|
||||
>>> result['type'] # "DDR4"
|
||||
>>> result['description'] # Product details from web
|
||||
"""
|
||||
# Validate spare part classification
|
||||
if not classify_as_spare_part(category):
|
||||
log.info(f"Category '{category}' not classified as spare part - skipping search")
|
||||
return None
|
||||
|
||||
# Build search query: part_number first (most specific), then item_name
|
||||
search_query = part_number if part_number else item_name
|
||||
if not search_query:
|
||||
log.warning("No part number or item name provided - cannot search")
|
||||
return None
|
||||
|
||||
try:
|
||||
# Apply rate limiting
|
||||
await _rate_limiter.acquire()
|
||||
|
||||
# Search with timeout protection
|
||||
start_time = asyncio.get_event_loop().time()
|
||||
remaining_timeout = timeout
|
||||
|
||||
# Try Google first
|
||||
log.info(f"Searching Google for: {search_query}")
|
||||
try:
|
||||
results = await asyncio.wait_for(
|
||||
search_google(search_query, timeout=min(10, remaining_timeout)),
|
||||
timeout=remaining_timeout
|
||||
)
|
||||
except (asyncio.TimeoutError, Exception):
|
||||
log.warning(f"Google search failed, trying Bing fallback")
|
||||
remaining_timeout = timeout - (asyncio.get_event_loop().time() - start_time)
|
||||
if remaining_timeout > 5:
|
||||
try:
|
||||
results = await asyncio.wait_for(
|
||||
search_bing(search_query, timeout=min(10, remaining_timeout)),
|
||||
timeout=remaining_timeout
|
||||
)
|
||||
except (asyncio.TimeoutError, Exception):
|
||||
log.warning(f"Bing fallback also failed for: {search_query}")
|
||||
results = None
|
||||
else:
|
||||
results = None
|
||||
|
||||
if not results:
|
||||
log.info(f"No search results found for: {search_query}")
|
||||
return None
|
||||
|
||||
# Extract specifications from search results
|
||||
log.info(f"Extracting specs from {len(results)} search results")
|
||||
spare_part_type = get_spare_part_type(category)
|
||||
item_fields = extract_specs_from_multiple_results(results, spare_part_type or category)
|
||||
|
||||
return {
|
||||
"category": category,
|
||||
"type": item_fields.get("type", ""),
|
||||
"description": item_fields.get("description", ""),
|
||||
"notes": item_fields.get("notes", ""),
|
||||
"confidence": 0.85, # Indicates data came from web search (vs. AI only)
|
||||
}
|
||||
|
||||
except asyncio.TimeoutError:
|
||||
log.warning(f"Search timed out after {timeout}s for: {search_query}")
|
||||
return None
|
||||
except Exception as e:
|
||||
log.error(f"Unexpected error during spare-parts search: {e}")
|
||||
return None
|
||||
|
||||
|
||||
async def search_multiple_candidates(
|
||||
candidates: list[Dict[str, str]],
|
||||
timeout: int = 30
|
||||
) -> Dict[str, Optional[Dict[str, Any]]]:
|
||||
"""
|
||||
Search multiple item candidates in parallel (rate-limited).
|
||||
|
||||
Args:
|
||||
candidates: List of dicts with 'category', 'part_number', 'item_name'
|
||||
timeout: Total timeout for all searches
|
||||
|
||||
Returns:
|
||||
Dict mapping candidate index to search results (or None if failed)
|
||||
"""
|
||||
tasks = []
|
||||
for candidate in candidates:
|
||||
task = search_spare_parts(
|
||||
category=candidate.get("category", ""),
|
||||
part_number=candidate.get("part_number"),
|
||||
item_name=candidate.get("item_name"),
|
||||
timeout=timeout
|
||||
)
|
||||
tasks.append(task)
|
||||
|
||||
try:
|
||||
results = await asyncio.wait_for(
|
||||
asyncio.gather(*tasks, return_exceptions=True),
|
||||
timeout=timeout
|
||||
)
|
||||
return {i: r for i, r in enumerate(results) if not isinstance(r, Exception)}
|
||||
except asyncio.TimeoutError:
|
||||
log.warning(f"Batch search timed out after {timeout}s")
|
||||
return {}
|
||||
214
tests/test_spare_parts_search.py
Normal file
214
tests/test_spare_parts_search.py
Normal file
@@ -0,0 +1,214 @@
|
||||
"""
|
||||
Integration tests for spare-parts search orchestrator and supporting services.
|
||||
|
||||
Tests the web scraping, spec extraction, and search coordination logic.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import asyncio
|
||||
from unittest.mock import AsyncMock, patch, MagicMock
|
||||
|
||||
from backend.services.spare_parts_search import search_spare_parts, search_multiple_candidates
|
||||
from backend.services.web_scraper import SearchRateLimiter, search_google, search_bing
|
||||
from backend.services.spec_extractor import ExtractedSpecs, extract_specs_from_search, extract_specs_from_multiple_results
|
||||
|
||||
|
||||
class TestSearchRateLimiter:
|
||||
"""Test rate limiter functionality."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rate_limiter_initialization(self):
|
||||
"""Rate limiter should initialize with default settings."""
|
||||
limiter = SearchRateLimiter()
|
||||
assert limiter.refill_rate == 0.2
|
||||
assert limiter.tokens >= 0
|
||||
assert limiter.capacity == 1.0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rate_limiter_custom_rate(self):
|
||||
"""Rate limiter should accept custom request rates."""
|
||||
limiter = SearchRateLimiter(requests_per_second=0.5)
|
||||
assert limiter.refill_rate == 0.5
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rate_limiter_acquire(self):
|
||||
"""Rate limiter should acquire tokens."""
|
||||
limiter = SearchRateLimiter()
|
||||
# Should not raise
|
||||
await limiter.acquire()
|
||||
assert limiter.tokens < 1.0
|
||||
|
||||
|
||||
class TestSpecExtractor:
|
||||
"""Test specification extraction from search results."""
|
||||
|
||||
def test_extracted_specs_dataclass(self):
|
||||
"""ExtractedSpecs should be initialized with defaults."""
|
||||
specs = ExtractedSpecs()
|
||||
assert specs.manufacturer is None
|
||||
assert specs.model is None
|
||||
assert specs.capacity is None
|
||||
assert specs.confidence == 0.0
|
||||
|
||||
def test_extract_memory_specs(self):
|
||||
"""Should extract DDR4 RAM specifications."""
|
||||
title = "Kingston Fury 16GB DDR4-3200 DIMM"
|
||||
snippet = "16GB capacity, DDR4 memory, 3200MHz speed"
|
||||
|
||||
specs = extract_specs_from_search(title, snippet)
|
||||
|
||||
assert specs.manufacturer == "Kingston"
|
||||
assert "DDR4" in (specs.memory_type or "")
|
||||
assert "16GB" in (specs.capacity or "")
|
||||
assert specs.confidence > 0.5
|
||||
|
||||
def test_extract_storage_specs(self):
|
||||
"""Should extract SSD specifications."""
|
||||
title = "Samsung 970 EVO Plus 1TB NVMe SSD"
|
||||
snippet = "1TB NVMe SSD, 3500MB/s read, M.2 form factor"
|
||||
|
||||
specs = extract_specs_from_search(title, snippet)
|
||||
|
||||
assert specs.manufacturer == "Samsung"
|
||||
assert specs.storage_type in ["NVMe", "SSD"]
|
||||
assert "1TB" in (specs.capacity or "")
|
||||
assert specs.confidence > 0.5
|
||||
|
||||
def test_extract_processor_specs(self):
|
||||
"""Should extract CPU specifications."""
|
||||
title = "Intel Core i7-12700K Processor"
|
||||
snippet = "12th Gen Intel Core i7, 12 cores, 5.0GHz max"
|
||||
|
||||
specs = extract_specs_from_search(title, snippet)
|
||||
|
||||
assert specs.processor_brand == "Intel"
|
||||
assert specs.confidence > 0.3
|
||||
|
||||
def test_extract_power_supply_specs(self):
|
||||
"""Should extract PSU specifications."""
|
||||
title = "Corsair RM850x 850W Power Supply"
|
||||
snippet = "850W modular power supply, 80+ Gold certified"
|
||||
|
||||
specs = extract_specs_from_search(title, snippet)
|
||||
|
||||
assert specs.manufacturer == "Corsair"
|
||||
assert "850W" in (specs.power_rating or "")
|
||||
assert specs.confidence > 0.3
|
||||
|
||||
def test_to_item_fields_memory(self):
|
||||
"""Should map memory specs to Item fields."""
|
||||
specs = ExtractedSpecs(
|
||||
manufacturer="Kingston",
|
||||
memory_type="DDR4",
|
||||
capacity="16GB",
|
||||
speed="3200MHz",
|
||||
description="Kingston DDR4 RAM"
|
||||
)
|
||||
|
||||
fields = specs.to_item_fields("Memory")
|
||||
|
||||
assert fields["type"] == "DDR4"
|
||||
assert "16GB" in fields["notes"]
|
||||
assert "Kingston" in fields["notes"]
|
||||
|
||||
def test_to_item_fields_storage(self):
|
||||
"""Should map storage specs to Item fields."""
|
||||
specs = ExtractedSpecs(
|
||||
manufacturer="Samsung",
|
||||
model="970 EVO Plus",
|
||||
storage_type="NVMe",
|
||||
capacity="1TB",
|
||||
description="Samsung NVMe SSD"
|
||||
)
|
||||
|
||||
fields = specs.to_item_fields("Storage")
|
||||
|
||||
assert fields["type"] == "NVMe"
|
||||
assert "1TB" in fields["notes"]
|
||||
assert "Samsung" in fields["notes"]
|
||||
|
||||
def test_extract_specs_from_multiple_results(self):
|
||||
"""Should extract specs from multiple results using highest confidence."""
|
||||
results = [
|
||||
{
|
||||
"title": "Kingston Fury 16GB DDR4-3200",
|
||||
"snippet": "16GB DDR4 memory module",
|
||||
"url": "https://example.com/1"
|
||||
},
|
||||
{
|
||||
"title": "Kingston Fury DDR4 RAM",
|
||||
"snippet": "Memory module",
|
||||
"url": "https://example.com/2"
|
||||
}
|
||||
]
|
||||
|
||||
fields = extract_specs_from_multiple_results(results, "Memory")
|
||||
|
||||
assert fields["type"] in ["DDR4", "Generic"]
|
||||
assert "description" in fields
|
||||
assert "notes" in fields
|
||||
|
||||
def test_extract_specs_empty_results(self):
|
||||
"""Should handle empty search results gracefully."""
|
||||
fields = extract_specs_from_multiple_results([], "Memory")
|
||||
|
||||
assert fields["type"] == ""
|
||||
assert fields["description"] == ""
|
||||
assert fields["notes"] == ""
|
||||
|
||||
|
||||
class TestSearchIntegration:
|
||||
"""Test end-to-end search integration."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_spare_parts_non_spare_part(self):
|
||||
"""Should return None for consumables (non-spare-parts)."""
|
||||
result = await search_spare_parts(category="6ft SATA Cable")
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_spare_parts_missing_query(self):
|
||||
"""Should return None if no search query provided."""
|
||||
result = await search_spare_parts(category="DDR4") # No part_number or item_name
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_spare_parts_timeout(self):
|
||||
"""Should return None on timeout (graceful degradation)."""
|
||||
with patch('backend.services.spare_parts_search.search_google', side_effect=asyncio.TimeoutError):
|
||||
with patch('backend.services.spare_parts_search.search_bing', side_effect=asyncio.TimeoutError):
|
||||
result = await search_spare_parts(
|
||||
category="Kingston DDR4 16GB",
|
||||
part_number="KF466C40RS-16",
|
||||
timeout=1
|
||||
)
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_multiple_candidates(self):
|
||||
"""Should handle batch search of multiple candidates."""
|
||||
candidates = [
|
||||
{"category": "6ft Cable", "part_number": None, "item_name": None}, # Non-spare-part
|
||||
{"category": "DDR4 RAM", "part_number": "KF466C40RS", "item_name": None},
|
||||
]
|
||||
|
||||
# Will fail to search but should handle gracefully
|
||||
result = await search_multiple_candidates(candidates, timeout=2)
|
||||
assert isinstance(result, dict)
|
||||
|
||||
|
||||
class TestWebScraper:
|
||||
"""Test web scraper functions."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_google_type(self):
|
||||
"""search_google should return list of dicts or None."""
|
||||
# Note: This would require mocking aiohttp to avoid actual network calls
|
||||
# In practice, we'd mock the aiohttp.ClientSession
|
||||
pass
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_search_bing_type(self):
|
||||
"""search_bing should return list of dicts or None."""
|
||||
# Similar to search_google - would require mocking
|
||||
pass
|
||||
Reference in New Issue
Block a user