diff --git a/backend/services/spec_extractor.py b/backend/services/spec_extractor.py new file mode 100644 index 00000000..1236a482 --- /dev/null +++ b/backend/services/spec_extractor.py @@ -0,0 +1,222 @@ +""" +Specification extractor service for search results parsing. + +Extracts product specifications (manufacturer, model, capacity, specs) from search +results and maps them to Item model fields for pre-population in onboarding UI. +""" + +import re +from dataclasses import dataclass, asdict +from typing import Optional, Dict, Any +import logging + +log = logging.getLogger("ainventory") + + +@dataclass +class ExtractedSpecs: + """Extracted specifications from search results.""" + + manufacturer: Optional[str] = None + model: Optional[str] = None + capacity: Optional[str] = None # e.g., "16GB" + memory_type: Optional[str] = None # e.g., "DDR4" + speed: Optional[str] = None # e.g., "3200MHz" + latency: Optional[str] = None # e.g., "CAS 16" + storage_type: Optional[str] = None # e.g., "SSD", "HDD" + processor_brand: Optional[str] = None # e.g., "Intel" + processor_model: Optional[str] = None # e.g., "Core i7-12700K" + power_rating: Optional[str] = None # e.g., "850W" + description: str = "" # Full snippet/details from search + confidence: float = 0.0 # 0.0-1.0 score + + def to_item_fields(self, category: str) -> Dict[str, str]: + """ + Map extracted specs to Item model fields. + + Args: + category: Item category (e.g., "Memory", "Storage", "Processor") + + Returns: + Dict with keys: type, description, notes + """ + type_str = "" + notes_parts = [] + + if category.lower() in ["memory", "ram"]: + if self.memory_type: + type_str = self.memory_type + if self.capacity: + notes_parts.append(f"Capacity: {self.capacity}") + if self.speed: + notes_parts.append(f"Speed: {self.speed}") + if self.latency: + notes_parts.append(f"Latency: {self.latency}") + + elif category.lower() in ["storage", "ssd", "hdd"]: + if self.storage_type: + type_str = self.storage_type + if self.capacity: + notes_parts.append(f"Capacity: {self.capacity}") + if self.model: + notes_parts.append(f"Model: {self.model}") + + elif category.lower() in ["processor", "cpu", "gpu"]: + if self.processor_brand: + type_str = self.processor_brand + if self.processor_model: + notes_parts.append(f"Model: {self.processor_model}") + + elif category.lower() in ["power", "psu"]: + if self.power_rating: + type_str = self.power_rating + if self.model: + notes_parts.append(f"Model: {self.model}") + + if self.manufacturer and self.manufacturer not in type_str: + notes_parts.insert(0, f"Manufacturer: {self.manufacturer}") + + return { + "type": type_str or "Generic", + "description": self.description[:200] if self.description else "", + "notes": " | ".join(notes_parts) if notes_parts else "" + } + + +def extract_specs_from_search(title: str, snippet: str, url: str = "") -> ExtractedSpecs: + """ + Extract specifications from a search result (title + snippet). + + Args: + title: Search result title + snippet: Search result snippet/description + url: Source URL (optional) + + Returns: + ExtractedSpecs object with extracted fields and confidence score + """ + full_text = f"{title} {snippet}".upper() + specs = ExtractedSpecs(description=snippet or title) + confidence_score = 0.0 + + # Manufacturer extraction + manufacturers = ["KINGSTON", "SAMSUNG", "CORSAIR", "INTEL", "AMD", "NVIDIA", + "CRUCIAL", "ADATA", "WESTERN DIGITAL", "SEAGATE", "HP", "DELL", + "LENOVO", "ASUS", "GIGABYTE", "MSI", "EVGA", "SAPPHIRE"] + for mfg in manufacturers: + if mfg in full_text: + specs.manufacturer = mfg.title() + confidence_score += 15 + break + + # Memory type extraction (DDR3/4/5, SODIMM, DIMM) + memory_patterns = { + r"DDR5?(\s|-)?LP?": "DDR5", + r"DDR4\b": "DDR4", + r"DDR3\b": "DDR3", + r"SODIMM\b": "SODIMM", + r"DIMM\b": "DIMM", + } + for pattern, memory_type in memory_patterns.items(): + if re.search(pattern, full_text): + specs.memory_type = memory_type + confidence_score += 20 + break + + # Capacity extraction (16GB, 512GB, 1TB, etc.) + capacity_match = re.search(r"(\d+(?:\.\d+)?)\s*(GB|TB|MB)", full_text) + if capacity_match: + specs.capacity = f"{capacity_match.group(1)}{capacity_match.group(2)}" + confidence_score += 25 + + # Speed extraction (3200MHz, 6400MT/s, etc.) + speed_match = re.search(r"(\d+)\s*(MHz|MT/S)", full_text) + if speed_match: + specs.speed = f"{speed_match.group(1)}{speed_match.group(2)}" + confidence_score += 10 + + # Latency extraction (CAS 16, CAS 18, etc.) + latency_match = re.search(r"CAS\s*(\d+)", full_text) + if latency_match: + specs.latency = f"CAS {latency_match.group(1)}" + confidence_score += 10 + + # Storage type extraction (SSD, HDD, NVMe, M.2) + storage_patterns = { + r"NVME\b|NVMe\b": "NVMe", + r"SSD\b": "SSD", + r"HDD\b|HARD DRIVE": "HDD", + r"M\.2\b": "M.2", + } + for pattern, storage_type in storage_patterns.items(): + if re.search(pattern, full_text): + specs.storage_type = storage_type + confidence_score += 20 + break + + # Processor extraction + processor_patterns = { + r"INTEL\s+(CORE\s+)?(I[3579]|PENTIUM|CELERON|XEON)": "Intel", + r"AMD\s+(RYZEN|EPYC|FX)": "AMD", + r"NVIDIA\s+GEFORCE": "NVIDIA", + } + for pattern, brand in processor_patterns.items(): + if re.search(pattern, full_text): + specs.processor_brand = brand + confidence_score += 15 + break + + # Model number extraction (alphanumeric patterns after brand) + if specs.manufacturer: + # Look for patterns like "Kingston KF466C40RS-16" or "Samsung 870 EVO" + model_match = re.search(rf"{specs.manufacturer.upper()}\s+([A-Z0-9\-]+)", full_text) + if model_match: + specs.model = model_match.group(1) + confidence_score += 20 + + # Power rating extraction (850W, 1000W, etc.) + power_match = re.search(r"(\d+)(\s*)W(?=\s|$|\D)", full_text) + if power_match: + specs.power_rating = f"{power_match.group(1)}W" + confidence_score += 15 + + # Normalize confidence to 0.0-1.0 range + specs.confidence = min(100.0, confidence_score) / 100.0 + + return specs + + +def extract_specs_from_multiple_results( + results: list[Dict[str, str]], + category: str +) -> Dict[str, str]: + """ + Extract specs from multiple search results and return best candidate fields. + + Aggregates specifications across multiple results, preferring specs with + highest confidence and deduplication. + + Args: + results: List of search result dicts with 'title', 'snippet', 'url' + category: Item category for field mapping + + Returns: + Dict with keys: type, description, notes + """ + if not results: + return {"type": "", "description": "", "notes": ""} + + # Extract from all results and pick highest confidence + all_specs = [] + for result in results: + specs = extract_specs_from_search( + result.get("title", ""), + result.get("snippet", ""), + result.get("url", "") + ) + all_specs.append(specs) + + # Pick spec set with highest confidence + best_specs = max(all_specs, key=lambda s: s.confidence) if all_specs else ExtractedSpecs() + + return best_specs.to_item_fields(category) diff --git a/backend/services/web_scraper.py b/backend/services/web_scraper.py new file mode 100644 index 00000000..1e1748c2 --- /dev/null +++ b/backend/services/web_scraper.py @@ -0,0 +1,213 @@ +""" +Web scraping service for spare-parts search with rate limiting and fallback engines. + +Provides HTTP request handling with User-Agent rotation and resilient search across +Google and Bing with graceful fallback and rate limiting (1 request per 5 seconds). +""" + +import asyncio +import random +import time +import urllib.parse +import logging +from typing import Optional, List, Dict, Any +import aiohttp +from bs4 import BeautifulSoup + +log = logging.getLogger("ainventory") + +USER_AGENT_POOL = [ + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (Chrome/120.0.0.0) Safari/537.36", + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (Chrome/120.0.0.0) Safari/537.36", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (Chrome/120.0.0.0) Safari/537.36", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0", + "Mozilla/5.0 (X11; Linux x86_64; rv:121.0) Gecko/20100101 Firefox/121.0", + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (Chrome/119.0.0.0) Safari/537.36 Edg/119.0.0.0", + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (Chrome/119.0.0.0) Safari/537.36", + "Mozilla/5.0 (iPad; CPU OS 17_1_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1.2 Mobile/15E148 Safari/604.1", + "Mozilla/5.0 (iPhone; CPU iPhone OS 17_1_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1.2 Mobile/15E148 Safari/604.1", + "Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36 (Chrome/120.0.0.0) Mobile Safari/537.36", +] + + +class SearchRateLimiter: + """ + Token bucket rate limiter for search requests. + + Ensures maximum request rate to avoid IP blocking. + Default: 1 request per 5 seconds (0.2 req/sec). + """ + + def __init__(self, requests_per_second: float = 0.2): + """ + Initialize rate limiter. + + Args: + requests_per_second: Rate limit (default 0.2 = 1 request per 5 seconds) + """ + self.capacity = 1.0 + self.refill_rate = requests_per_second + self.tokens = 1.0 + self.last_refill = time.time() + + async def acquire(self): + """ + Block until rate quota is available (token bucket algorithm). + + Uses time-based token refill without asyncio.sleep loops. + """ + while self.tokens < 1.0: + now = time.time() + elapsed = now - self.last_refill + self.tokens = min( + self.capacity, + self.tokens + elapsed * self.refill_rate + ) + self.last_refill = now + + if self.tokens < 1.0: + await asyncio.sleep(0.1) + + self.tokens -= 1.0 + + +async def search_google(query: str, timeout: int = 10) -> Optional[List[Dict[str, str]]]: + """ + Search Google for spare-parts information. + + Args: + query: Search query string + timeout: Request timeout in seconds + + Returns: + List of dicts with 'title', 'url', 'snippet' keys, or None on error + + Example: + >>> results = await search_google("Kingston DDR4 16GB RAM") + >>> results[0]['title'] # Product name + """ + url = f"https://www.google.com/search?q={urllib.parse.quote(query)}" + headers = {"User-Agent": random.choice(USER_AGENT_POOL)} + + try: + async with aiohttp.ClientSession() as session: + async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=timeout)) as response: + if response.status in (429, 403): + log.warning(f"Google blocked request: {response.status}") + return None + + html = await response.text() + soup = BeautifulSoup(html, "html.parser") + results = [] + + # Google search result container: div.g + for result_div in soup.find_all("div", class_="g")[:5]: # Top 5 results + title_elem = result_div.find("h3") + url_elem = result_div.find("a") + snippet_elem = result_div.find("span", class_="VwiC3b") + + if title_elem and url_elem: + results.append({ + "title": title_elem.get_text(), + "url": url_elem.get("href", ""), + "snippet": snippet_elem.get_text() if snippet_elem else "" + }) + + return results if results else None + + except asyncio.TimeoutError: + log.warning(f"Google search timed out: {query}") + raise + except aiohttp.ClientError as e: + log.warning(f"Google search failed: {e}") + return None + except Exception as e: + log.error(f"Google search error: {e}") + return None + + +async def search_bing(query: str, timeout: int = 10) -> Optional[List[Dict[str, str]]]: + """ + Search Bing for spare-parts information (fallback from Google). + + More stable than Google with less blocking. + + Args: + query: Search query string + timeout: Request timeout in seconds + + Returns: + List of dicts with 'title', 'url', 'snippet' keys, or None on error + """ + url = f"https://www.bing.com/search?q={urllib.parse.quote(query)}" + headers = {"User-Agent": random.choice(USER_AGENT_POOL)} + + try: + async with aiohttp.ClientSession() as session: + async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=timeout)) as response: + if response.status in (429, 403): + log.warning(f"Bing blocked request: {response.status}") + return None + + html = await response.text() + soup = BeautifulSoup(html, "html.parser") + results = [] + + # Bing search result container: li.b_algo + for result_li in soup.find_all("li", class_="b_algo")[:5]: # Top 5 results + title_elem = result_li.find("h2") + url_elem = result_li.find("a") + snippet_elem = result_li.find("p") + + if title_elem and url_elem: + results.append({ + "title": title_elem.get_text(), + "url": url_elem.get("href", ""), + "snippet": snippet_elem.get_text() if snippet_elem else "" + }) + + return results if results else None + + except asyncio.TimeoutError: + log.warning(f"Bing search timed out: {query}") + raise + except aiohttp.ClientError as e: + log.warning(f"Bing search failed: {e}") + return None + except Exception as e: + log.error(f"Bing search error: {e}") + return None + + +async def fetch_and_parse_html(url: str, timeout: int = 10) -> Optional[str]: + """ + Fetch and parse HTML from an arbitrary URL. + + Args: + url: Target URL + timeout: Request timeout in seconds + + Returns: + HTML content string or None on error + """ + headers = {"User-Agent": random.choice(USER_AGENT_POOL)} + + try: + async with aiohttp.ClientSession() as session: + async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=timeout)) as response: + if response.status == 200: + return await response.text() + else: + log.warning(f"Failed to fetch {url}: {response.status}") + return None + + except asyncio.TimeoutError: + log.warning(f"HTML fetch timed out: {url}") + raise + except aiohttp.ClientError as e: + log.warning(f"HTML fetch failed: {e}") + return None + except Exception as e: + log.error(f"HTML fetch error: {e}") + return None