tfm_ainventory/backend/services/web_scraper.py

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