""" 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)