---
wave: 2
depends_on: ["4.1-PLAN-01.md"]
files_modified:
- path: "backend/services/spare_parts_search.py"
- path: "backend/services/web_scraper.py"
- path: "backend/services/spec_extractor.py"
- path: "backend/routers/items.py"
- path: "tests/test_spare_parts_search.py"
- path: "backend/requirements.txt"
autonomous: true
---
# Phase 4.1 Wave 2: Web Scraping Service & Backend Integration
**Objective:** Implement web scraping and spec extraction services, integrate into `/api/onboarding/extract` endpoint, and add comprehensive backend tests.
**Prerequisites:** Wave 1 must be complete (spare_parts_whitelist.py, AI prompt enhancements, classification tests passing).
---
## Task 1: Create Web Scraper Service
```xml
Build HTTP request handler with rate limiting, User-Agent rotation, and fallback search engines (Google → Bing) for resilient spare-parts searching.
- 4.1-RESEARCH.md sections 1 (Web Scraping Best Practices) and 5 (Backend Integration — Rate Limiting Implementation)
- PROJECT_ARCHITECTURE.md section 2.1 (Python 3.12+, FastAPI, async patterns)
- No existing scraper — new file
Create file: backend/services/web_scraper.py
**Module Structure:**
1. Import statements:
```python
import asyncio
import random
import time
from typing import Optional, List
import aiohttp
from bs4 import BeautifulSoup
```
(Note: add aiohttp and beautifulsoup4 to requirements.txt)
2. Create constant: USER_AGENT_POOL (list of 10+ realistic User-Agent strings)
```python
USER_AGENT_POOL = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (Chrome/120.0.0.0)",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (Firefox/121.0)",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (Safari/537.36)",
# ... add 7+ more variations across Windows, Linux, macOS and Chrome/Firefox/Safari
]
```
3. Implement class: `SearchRateLimiter`
- Method `__init__(self, requests_per_second: float = 0.2)` → (1 request per 5 seconds)
- Method `async acquire(self)` → blocks until rate quota available, using token bucket algorithm
- Attributes: `capacity`, `refill_rate`, `tokens`, `last_refill` (time-based)
- Logic from 4.1-RESEARCH.md section 5 (Rate Limiting Implementation) pseudocode
4. Implement async function: `search_google(query: str, timeout: int = 10) -> Optional[List[dict]]`
- Builds Google search URL: `f"https://www.google.com/search?q={urllib.parse.quote(query)}"`
- Uses aiohttp with rotating User-Agent
- Parses HTML with BeautifulSoup using CSS selector `div.g` (Google result container)
- Returns list of dicts: `[{"title": str, "url": str, "snippet": str}, ...]` or None
- On 429/403 error: log warning, return None
- On timeout: raise asyncio.TimeoutError
- Default timeout: 10 seconds
5. Implement async function: `search_bing(query: str, timeout: int = 10) -> Optional[List[dict]]`
- Builds Bing search URL: `f"https://www.bing.com/search?q={urllib.parse.quote(query)}"`
- Parses HTML with CSS selector `li.b_algo` (Bing result container)
- Returns same format as search_google()
- More stable than Google (less blocking)
6. Implement async function: `fetch_and_parse_html(url: str, timeout: int = 10) -> Optional[str]`
- Fetches HTML from arbitrary URL
- Returns HTML string or None on error
- Timeout: 10 seconds default
**Code Quality:**
- All functions are async (use `async def`)
- Type hints on all parameters and returns
- Docstrings with example usage
- Exception handling: catch aiohttp.ClientError, asyncio.TimeoutError, and BeautifulSoup parse errors
- No blocking I/O in async functions
- Rate limiter uses time.time() for token bucket (not asyncio.sleep loops)
- File exists: backend/services/web_scraper.py
- Grep finds: `class SearchRateLimiter:` in file
- Grep finds: `async def search_google(` in file
- Grep finds: `async def search_bing(` in file
- Grep finds: `USER_AGENT_POOL` with 10+ entries
- Module imports without error: `python3 -c "from backend.services.web_scraper import SearchRateLimiter, search_google, search_bing"`
- Rate limiter has `acquire()` async method
- Both search functions accept `query: str, timeout: int` parameters
- Both search functions return `Optional[List[dict]]`
- aiohttp and beautifulsoup4 added to backend/requirements.txt
```
---
## Task 2: Create Spec Extractor Service
```xml
Extract product specifications, manufacturer, model, and description from search results using regex patterns and data mapping to Item fields.
- 4.1-RESEARCH.md sections 4 (Search Result Parsing) with regex patterns and data extraction pipeline
- 4.1-RESEARCH.md section 4 (Mapping to Item Fields) for spec extraction rules
- PROJECT_ARCHITECTURE.md section 3 (Item model fields: Category, Type, Notes)
Create file: backend/services/spec_extractor.py
**Module Structure:**
1. Import statements:
```python
import re
from typing import Optional, Dict, Any
from dataclasses import dataclass
```
2. Create dataclass: `ExtractedSpecs`
```python
@dataclass
class ExtractedSpecs:
manufacturer: Optional[str]
model: Optional[str]
capacity: Optional[str] # e.g., "16GB"
memory_type: Optional[str] # e.g., "DDR4"
speed: Optional[str] # e.g., "3200MHz"
latency: Optional[str] # e.g., "CAS 16"
storage_type: Optional[str] # e.g., "SSD", "HDD"
processor_brand: Optional[str] # e.g., "Intel"
processor_model: Optional[str] # e.g., "Core i7-12700K"
power_rating: Optional[str] # e.g., "850W"
description: str # Full snippet/details from search
confidence: float # 0.0-1.0 score
def to_item_fields(self, category: str) -> Dict[str, str]:
"""Map extracted specs to Item model fields."""
# Implementation: see action below
```
3. Implement function: `extract_specs_from_snippet(snippet: str, title: str, category: str) -> ExtractedSpecs`
- Input: search result title, snippet, and item category
- Uses regex patterns from 4.1-RESEARCH.md section 4:
- Memory: `r'\b(\d+)\s*(GB|TB)\s*(DDR\d|DRAM|RAM|SDRAM)'`
- Storage: `r'\b(\d+)\s*(GB|TB)\s*(SSD|NVME|NVMe|M\.2|HDD|SATA)'`
- Processor: `r'(Intel|AMD)\s+([A-Z0-9-]+)\s*(\d+\.\d+\s*GHz)?'`
- Power: `r'\b(\d+)\s*(W|watts?|watt)\s*(power|supply|PSU)'`
- Speed/Latency: `r'(\d+)\s*(MHz|GHz|CAS|Latency)'`
- Extract manufacturer: check title/snippet for known brands (Kingston, Samsung, Intel, AMD, Corsair, etc.)
- Build confidence score:
- Exact part match in snippet: +0.2
- All major specs found: +0.3
- Manufacturer + model: +0.2
- Consistency checks (e.g., DDR4 with GHz speed): +0.25
- Return ExtractedSpecs dataclass
4. Implement method: `ExtractedSpecs.to_item_fields(category: str) -> Dict[str, str]`
- Maps specs to Item model fields:
- **Item.Category**: category (from whitelist)
- **Item.Type**: formatted as "[manufacturer] [model] [capacity/speed]" or specific type (e.g., "DDR4", "NVMe")
- **Item.Notes**: full description including all extracted specs
- Example output:
```python
{
"category": "RAM",
"item_type": "Kingston Fury 16GB DDR4-3200",
"notes": "Kingston Fury 16GB DDR4-3200MHz CAS Latency 16 - High-performance RAM module"
}
```
5. Implement function: `normalize_variations(text: str) -> str`
- Normalizes common abbreviations:
- "DDR4" ↔ "DDR 4" ↔ "DDR-4" → "DDR4"
- "3200 MHz" ↔ "3200MHz" → "3200MHz"
- "Intel i7" ↔ "Intel Core i7" → standardized format
- Used in regex extraction for consistency
**Code Quality:**
- All functions have type hints
- Docstrings with example input/output
- Regex patterns are compiled once as module constants (not in loop)
- Error handling: gracefully handle missing fields (return None/default)
- Confidence scoring is deterministic (no randomness)
- File exists: backend/services/spec_extractor.py
- Grep finds: `class ExtractedSpecs:` in file
- Grep finds: `def extract_specs_from_snippet(` in file
- Grep finds: `def to_item_fields(` in file
- Module imports without error: `python3 -c "from backend.services.spec_extractor import ExtractedSpecs, extract_specs_from_snippet"`
- ExtractedSpecs dataclass has all fields: manufacturer, model, capacity, memory_type, speed, latency, storage_type, processor_brand, processor_model, power_rating, description, confidence
- to_item_fields() returns Dict[str, str] with keys: category, item_type, notes
- Example test: `extract_specs_from_snippet("Kingston Fury 16GB DDR4-3200 RAM", "...", "RAM")` returns ExtractedSpecs with confidence > 0.5
```
---
## Task 3: Create Spare-Parts Search Orchestrator Service
```xml
Build main orchestrator service that combines web scraping, rate limiting, and spec extraction with automatic fallback and timeout handling.
- 4.1-RESEARCH.md section 5 (Backend Integration Architecture) — Search Service Pseudocode
- Task 1 output: backend/services/web_scraper.py
- Task 2 output: backend/services/spec_extractor.py
- backend/ai/spare_parts_whitelist.py (from Wave 1)
Create file: backend/services/spare_parts_search.py
**Module Structure:**
1. Import statements:
```python
import asyncio
import logging
from typing import Optional
from dataclasses import dataclass
from backend.services.web_scraper import search_google, search_bing, SearchRateLimiter
from backend.services.spec_extractor import ExtractedSpecs, extract_specs_from_snippet
from backend.ai.spare_parts_whitelist import classify_as_spare_part
```
2. Create dataclass: `SparePartSearchResult`
```python
@dataclass
class SparePartSearchResult:
status: str # "success", "timeout", "error", "no_results", "not_spare_part"
specs: Optional[ExtractedSpecs] = None
error: Optional[str] = None
confidence: float = 0.0
```
3. Create module-level rate limiter (singleton pattern):
```python
_rate_limiter = SearchRateLimiter(requests_per_second=0.2) # 1 request per 5 seconds
```
4. Implement async function: `search_and_extract(part_number: str, category: str, manufacturer: Optional[str] = None, timeout: int = 20) -> SparePartSearchResult`
- Algorithm (from 4.1-RESEARCH.md section 5):
a. Check: is category in spare-parts whitelist? If not → return `SparePartSearchResult(status="not_spare_part", ...)`
b. Build search query: `f"{part_number} {category} {manufacturer or ''}".strip()`
c. Wrap in asyncio.timeout(timeout) block:
- Acquire rate limiter: `await _rate_limiter.acquire()`
- Try Google search: `results = await search_google(query)`
- If no results → fallback to Bing: `results = await search_bing(query)`
- If still no results → return `SparePartSearchResult(status="no_results", error="...")`
- Parse best result (index 0): `specs = extract_specs_from_snippet(results[0]["title"], results[0]["snippet"], category)`
- Return `SparePartSearchResult(status="success", specs=specs, confidence=specs.confidence)`
d. On asyncio.TimeoutError → return `SparePartSearchResult(status="timeout", error="Search exceeded {timeout}s timeout")`
e. On Exception → return `SparePartSearchResult(status="error", error=str(e))`
5. Implement logging:
- Log all search attempts with query and category
- Log timeouts, errors, and fallbacks (INFO level)
- Log rate limiter waits (DEBUG level)
- Use logger: `logging.getLogger(__name__)`
**Code Quality:**
- All functions are async
- Type hints on all parameters and returns
- Docstrings with example usage
- No external API calls to Google/Bing in unit tests (use mocks)
- Graceful error handling for all network failures
- Timeout is enforced by asyncio.timeout() context manager (exact timeout from parameter)
- File exists: backend/services/spare_parts_search.py
- Grep finds: `class SparePartSearchResult:` in file
- Grep finds: `async def search_and_extract(` in file
- Module imports without error: `python3 -c "from backend.services.spare_parts_search import search_and_extract, SparePartSearchResult"`
- SparePartSearchResult has fields: status, specs, error, confidence
- search_and_extract accepts parameters: part_number: str, category: str, manufacturer: Optional[str], timeout: int
- Function returns SparePartSearchResult with appropriate status values
- Rate limiter is module-level singleton: `_rate_limiter = SearchRateLimiter(...)`
- Timeout is enforced via asyncio.timeout() context manager
```
---
## Task 4: Integrate Search into `/api/onboarding/extract` Endpoint
```xml
Modify backend router to call spare-parts search service after AI extraction when category matches whitelist and part number exists.
- backend/routers/items.py (locate `/api/onboarding/extract` endpoint)
- 4.1-CONTEXT.md decisions D-05, D-06, D-07, D-08 (search trigger and user flow)
- 4.1-RESEARCH.md section 5 (Integration Flow in `/api/onboarding/extract`)
- Tasks 1-3 output (all search services)
Modify file: backend/routers/items.py
**Action Steps:**
1. Add imports at top of file:
```python
from backend.services.spare_parts_search import search_and_extract as search_spare_parts
from backend.ai.spare_parts_whitelist import classify_as_spare_part
import asyncio
```
2. Locate the `/api/onboarding/extract` POST endpoint (should return extracted item data from AI)
3. Modify endpoint logic AFTER AI extraction step (Gemini or Claude):
```python
# Existing AI extraction code...
ai_data = await extract_with_gemini_or_claude(...) # Returns: {name, category, item_type, part_number, ...}
# NEW: Check if search should be triggered
search_results = None
search_status = "skipped"
search_error = None
category = ai_data.get("category", "").strip()
part_number = ai_data.get("part_number", "").strip()
if classify_as_spare_part(category) and part_number:
# Trigger spare-parts search
try:
manufacturer = ai_data.get("manufacturer", "")
search_result = await search_spare_parts(
part_number=part_number,
category=category,
manufacturer=manufacturer,
timeout=20 # 20-30 seconds from RESEARCH.md
)
search_status = search_result.status
search_error = search_result.error
if search_result.status == "success" and search_result.specs:
search_results = search_result.specs.to_item_fields(category)
except asyncio.TimeoutError:
search_status = "timeout"
search_error = "Search exceeded 20 second timeout"
except Exception as e:
search_status = "error"
search_error = str(e)
# Return combined response
return {
"ai_data": ai_data,
"search_results": search_results,
"search_status": search_status,
"search_error": search_error
}
```
4. Response schema should include:
- `ai_data`: dict with original AI-extracted fields
- `search_results`: dict with `{category, item_type, notes}` or null if skipped/failed
- `search_status`: string enum ["success", "timeout", "error", "no_results", "skipped", "not_spare_part"]
- `search_error`: error message string or null
5. Ensure endpoint remains async and doesn't block other requests
**Code Quality:**
- No changes to existing AI extraction logic
- Search is called conditionally (only if category matches AND part_number exists)
- Timeout is enforced (20 seconds from RESEARCH.md)
- Errors are caught and returned in response (not raising exceptions)
- Response structure matches frontend expectations (from RESEARCH.md section 6)
- File backend/routers/items.py modified
- Grep finds: `from backend.services.spare_parts_search import search_spare_parts` in file
- Grep finds: `from backend.ai.spare_parts_whitelist import classify_as_spare_part` in file
- Grep finds: `classify_as_spare_part(category) and part_number:` in file
- Grep finds: `search_status = search_result.status` in file
- Endpoint returns dict with keys: ai_data, search_results, search_status, search_error
- Endpoint is still async function (no blocking calls)
- Timeout is set to 20 seconds: `timeout=20`
- Search is conditional: only triggered if category is spare part AND part_number exists
```
---
## Task 5: Create Backend Tests for Search Services
```xml
Write comprehensive pytest tests for search orchestrator, web scraper, and spec extractor with mocked HTTP responses.
- 4.1-RESEARCH.md section 8 (Testing & Validation Strategy) — Unit Tests and Integration Tests subsections
- Tasks 1-4 output (all services)
- PROJECT_ARCHITECTURE.md section 2.1 (Testing: Pytest)
Create file: tests/test_spare_parts_search.py
**Test Structure (Pytest with pytest-asyncio for async tests):**
```python
import pytest
from unittest.mock import AsyncMock, patch
from backend.services.spare_parts_search import search_and_extract, SparePartSearchResult
from backend.services.spec_extractor import extract_specs_from_snippet
from backend.ai.spare_parts_whitelist import classify_as_spare_part
class TestSparePartsSearch:
"""Test spare-parts search orchestrator."""
@pytest.mark.asyncio
async def test_search_and_extract_not_spare_part(self):
"""Non-spare-parts category should skip search."""
result = await search_and_extract(
part_number="6ft Cable",
category="Cable",
timeout=20
)
assert result.status == "not_spare_part"
assert result.specs is None
@pytest.mark.asyncio
async def test_search_and_extract_no_part_number(self):
"""Missing part number should skip search."""
result = await search_and_extract(
part_number="",
category="RAM",
timeout=20
)
assert result.status == "skipped"
@pytest.mark.asyncio
@patch('backend.services.spare_parts_search.search_google')
async def test_search_and_extract_success(self, mock_google):
"""Successful search should return specs."""
mock_google.return_value = [
{
"title": "Kingston Fury 16GB DDR4-3200",
"snippet": "Kingston Fury 16GB DDR4-3200MHz CAS Latency 16",
"url": "https://example.com"
}
]
result = await search_and_extract(
part_number="Kingston Fury 16GB",
category="RAM",
timeout=20
)
assert result.status == "success"
assert result.specs is not None
assert result.confidence > 0.5
@pytest.mark.asyncio
async def test_search_and_extract_timeout(self):
"""Timeout should return timeout status."""
result = await search_and_extract(
part_number="Kingston Fury 16GB",
category="RAM",
timeout=0.001 # Force timeout
)
assert result.status == "timeout"
assert result.specs is None
assert "timeout" in result.error.lower()
@pytest.mark.asyncio
@patch('backend.services.spare_parts_search.search_google')
@patch('backend.services.spare_parts_search.search_bing')
async def test_search_fallback_to_bing(self, mock_bing, mock_google):
"""Should fallback to Bing if Google returns no results."""
mock_google.return_value = None
mock_bing.return_value = [
{
"title": "Samsung 970 EVO 1TB NVMe",
"snippet": "Samsung 970 EVO 1TB NVMe SSD",
"url": "https://example.com"
}
]
result = await search_and_extract(
part_number="Samsung 970 EVO",
category="SSD",
timeout=20
)
assert result.status == "success"
mock_bing.assert_called_once()
class TestSpecExtractor:
"""Test specification extraction from search results."""
def test_extract_specs_from_snippet_ram(self):
"""Extract RAM specifications."""
specs = extract_specs_from_snippet(
snippet="Kingston Fury 16GB DDR4-3200MHz CAS Latency 16",
title="Kingston Fury 16GB DDR4-3200",
category="RAM"
)
assert specs.manufacturer == "Kingston"
assert specs.capacity == "16GB"
assert specs.memory_type == "DDR4"
assert specs.speed == "3200"
def test_extract_specs_from_snippet_ssd(self):
"""Extract SSD specifications."""
specs = extract_specs_from_snippet(
snippet="Samsung 970 EVO 1TB NVMe SSD",
title="Samsung 970 EVO 1TB",
category="SSD"
)
assert specs.manufacturer == "Samsung"
assert specs.capacity == "1TB"
assert specs.storage_type == "NVMe"
def test_to_item_fields_mapping(self):
"""Test mapping specs to Item model fields."""
specs = extract_specs_from_snippet(
snippet="Kingston Fury 16GB DDR4-3200MHz",
title="Kingston Fury 16GB DDR4-3200",
category="RAM"
)
item_fields = specs.to_item_fields("RAM")
assert item_fields["category"] == "RAM"
assert "Kingston" in item_fields["item_type"]
assert "16GB" in item_fields["notes"]
class TestWhitelistIntegration:
"""Test whitelist integration with search."""
def test_classify_spare_part_enables_search(self):
"""Spare parts should enable search."""
assert classify_as_spare_part("RAM") is True
assert classify_as_spare_part("SSD") is True
def test_consumable_disables_search(self):
"""Consumables should skip search."""
assert classify_as_spare_part("Cable") is False
assert classify_as_spare_part("Thermal Paste") is False
```
**Test Execution:**
- All tests must pass: `pytest tests/test_spare_parts_search.py -v`
- Async tests use `@pytest.mark.asyncio` decorator
- Mock external HTTP calls (don't make real requests to Google/Bing)
- Use `pytest-asyncio` package for async support
**Code Quality:**
- Descriptive test names
- Docstrings on each test
- Clear assertions with expected values
- Minimum 10 test cases
- File exists: tests/test_spare_parts_search.py
- Test suite runs without errors: `pytest tests/test_spare_parts_search.py -v`
- Minimum 10 test cases implemented
- Test passes: `test_search_and_extract_not_spare_part`
- Test passes: `test_search_and_extract_timeout`
- Test passes: `test_search_fallback_to_bing` (using mocked Bing)
- Test passes: `test_extract_specs_from_snippet_ram`
- Test passes: `test_to_item_fields_mapping`
- pytest-asyncio added to backend/requirements.txt
- All external HTTP calls are mocked (no real requests in tests)
```
---
## Task 6: Update Backend Dependencies
```xml
Add new Python packages to requirements.txt with version constraints for all services created in Wave 2.
- backend/requirements.txt (current state)
- AI_RULES.md section 2 (DEPENDENCIES: Update requirements.txt with version constraints)
- Tasks 1-5 (all new services)
Modify file: backend/requirements.txt
**Action Steps:**
1. Add these lines (in alphabetical order if file is sorted):
```
aiohttp==3.9.1
beautifulsoup4==4.12.2
fuzzywuzzy==0.18.0
python-Levenshtein==0.21.1
pytest-asyncio==0.23.2
```
2. Verify no duplicate entries exist in file
3. Ensure all existing dependencies remain unchanged (only ADD new ones)
**Rationale:**
- **aiohttp**: Async HTTP client for web scraping in web_scraper.py
- **beautifulsoup4**: HTML parsing for search results
- **fuzzywuzzy**: Fuzzy string matching for spare-parts classification (added in Wave 1)
- **python-Levenshtein**: Fast Levenshtein distance for fuzzywuzzy
- **pytest-asyncio**: Async test support for pytest
**Code Quality:**
- Use specific version pinning (major.minor.patch) for stability
- No pre-release versions (no alpha/beta)
- Versions chosen from stable releases as of 2026-04
- File backend/requirements.txt modified
- Grep finds: `aiohttp==3.9.1` in file
- Grep finds: `beautifulsoup4==4.12.2` in file
- Grep finds: `fuzzywuzzy==0.18.0` in file
- Grep finds: `pytest-asyncio==0.23.2` in file
- No duplicate entries in file
- All existing dependencies remain unchanged
- File has no syntax errors (can run `pip install -r backend/requirements.txt` without parsing errors)
```
---
## Wave 2 Summary
**What this wave accomplishes:**
- Creates resilient web scraping service with fallback engines and rate limiting
- Builds spec extraction service with regex patterns and confidence scoring
- Implements orchestrator service combining all search logic with timeout handling
- Integrates search into backend API endpoint for automatic spare-parts lookup
- Provides comprehensive backend tests with mocked HTTP
**Completion Criteria:**
- All 6 tasks pass acceptance criteria
- Backend tests pass: `pytest tests/test_spare_parts_search.py -v` → all tests pass
- All services import without error:
```bash
python3 -c "from backend.services.web_scraper import search_google, search_bing"
python3 -c "from backend.services.spec_extractor import extract_specs_from_snippet"
python3 -c "from backend.services.spare_parts_search import search_and_extract"
```
- `/api/onboarding/extract` endpoint returns search results in expected format
- Dependencies installed: `pip install -r backend/requirements.txt` → no errors
**Dependencies for Wave 3:**
- All search services (Tasks 1-3)
- Backend API integration (Task 4)
- Backend tests passing (Task 5)
- Dependencies installed (Task 6)
---