214 lines
7.7 KiB
Python
214 lines
7.7 KiB
Python
"""
|
|
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
|