151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
"""
|
|
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 {}
|