diff --git a/app/services/direct_search_service.py b/app/services/direct_search_service.py index 12650ec..0af587a 100644 --- a/app/services/direct_search_service.py +++ b/app/services/direct_search_service.py @@ -1,18 +1,21 @@ """ -Direct Search Service - Recherche directement sur les sites e-commerce -Au lieu d'utiliser un moteur de recherche externe, scrape les pages de recherche de chaque site. +Direct Search Service - Recherche sur les sites e-commerce via Browserless +Utilise Playwright/Browserless pour gérer JavaScript et contourner la détection de bots. """ import asyncio import logging +import os import re from urllib.parse import quote_plus, urljoin, urlparse -import httpx from bs4 import BeautifulSoup +from playwright.async_api import async_playwright logger = logging.getLogger(__name__) +BROWSERLESS_URL = os.getenv("BROWSERLESS_URL", "ws://browserless:3000") + class SearchResult: """Résultat de recherche""" @@ -31,38 +34,38 @@ class SearchResult: DEFAULT_SITE_CONFIGS = { "amazon.fr": { "search_url": "https://www.amazon.fr/s?k={query}", - "product_selector": "div[data-component-type='s-search-result'] a.a-link-normal[href*='/dp/']", - "title_selector": "span.a-text-normal", + "product_selector": "div[data-component-type='s-search-result'] h2 a", + "wait_selector": "div[data-component-type='s-search-result']", }, "amazon.com": { "search_url": "https://www.amazon.com/s?k={query}", - "product_selector": "div[data-component-type='s-search-result'] a.a-link-normal[href*='/dp/']", - "title_selector": "span.a-text-normal", + "product_selector": "div[data-component-type='s-search-result'] h2 a", + "wait_selector": "div[data-component-type='s-search-result']", }, "action.com": { "search_url": "https://www.action.com/fr-fr/search/?q={query}", - "product_selector": "a.product-card__link", - "title_selector": ".product-card__title", + "product_selector": "a.product-card__link, .product-card a, a[href*='/p/']", + "wait_selector": ".product-card, .search-results", }, "fnac.com": { "search_url": "https://www.fnac.com/SearchResult/ResultList.aspx?Search={query}", - "product_selector": "a.Article-title", - "title_selector": "a.Article-title", + "product_selector": "a.Article-title, .Article-item a.js-minifa-title", + "wait_selector": ".Article-item", }, "cdiscount.com": { "search_url": "https://www.cdiscount.com/search/10/{query}.html", - "product_selector": "a.prdtBImg", - "title_selector": ".prdtTit", + "product_selector": "a.prdtBImg, .prdtBILDetails a", + "wait_selector": ".prdtBILDetails", }, "darty.com": { "search_url": "https://www.darty.com/nav/recherche?text={query}", - "product_selector": "a.product-card__link", - "title_selector": ".product-card__title", + "product_selector": ".product-card a, a[href*='/product']", + "wait_selector": ".product-card", }, "boulanger.com": { "search_url": "https://www.boulanger.com/resultats?tr={query}", - "product_selector": "a.product-list__link", - "title_selector": ".product-list__title", + "product_selector": ".product-list__item a, a[href*='/ref/']", + "wait_selector": ".product-list", }, } @@ -71,16 +74,16 @@ async def search( query: str, sites: list[dict], max_results: int = 20, - timeout: float = 15.0, + timeout: float = 30.0, ) -> list[SearchResult]: """ - Recherche directement sur les sites e-commerce. + Recherche sur les sites e-commerce via Browserless. Args: query: Terme de recherche sites: Liste de dicts avec {domain, search_url, product_link_selector, name} max_results: Nombre maximum de résultats - timeout: Timeout par requête + timeout: Timeout par site Returns: Liste de SearchResult @@ -92,38 +95,33 @@ async def search( all_results = [] results_per_site = max(5, max_results // len(sites)) - # Rechercher sur chaque site en parallèle - tasks = [] + # Rechercher sur chaque site en séquence (pour éviter de surcharger Browserless) for site in sites: - task = asyncio.create_task( - _search_site(query, site, results_per_site, timeout) - ) - tasks.append(task) + try: + results = await _search_site_browserless(query, site, results_per_site, timeout) + all_results.extend(results) + logger.info(f"Site {site.get('domain')}: {len(results)} résultats") + except Exception as e: + logger.error(f"Erreur recherche sur {site.get('domain')}: {e}") - results = await asyncio.gather(*tasks, return_exceptions=True) + if len(all_results) >= max_results: + break - for site, result in zip(sites, results): - if isinstance(result, Exception): - logger.error(f"Erreur recherche sur {site.get('domain')}: {result}") - continue - if result: - all_results.extend(result) - - logger.info(f"Recherche directe: {len(all_results)} résultats totaux") + logger.info(f"Recherche Browserless: {len(all_results)} résultats totaux") return all_results[:max_results] -async def _search_site( +async def _search_site_browserless( query: str, site: dict, max_results: int, timeout: float, ) -> list[SearchResult]: - """Recherche sur un site spécifique""" + """Recherche sur un site via Browserless (Playwright)""" domain = site.get("domain", "").lower() search_url = site.get("search_url") or _get_default_search_url(domain) product_selector = site.get("product_link_selector") or _get_default_product_selector(domain) - site_name = site.get("name", domain) + wait_selector = _get_default_wait_selector(domain) if not search_url: logger.warning(f"Pas d'URL de recherche configurée pour {domain}") @@ -133,94 +131,121 @@ async def _search_site( encoded_query = quote_plus(query) final_url = search_url.replace("{query}", encoded_query) - logger.info(f"Recherche directe sur {domain}: {final_url}") + logger.info(f"Recherche Browserless sur {domain}: {final_url}") try: - async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client: - response = await client.get( - final_url, - headers={ - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", - "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", - "Accept-Language": "fr-FR,fr;q=0.9,en;q=0.8", - "Accept-Encoding": "gzip, deflate, br", - "DNT": "1", - "Connection": "keep-alive", - "Upgrade-Insecure-Requests": "1", - }, - ) - response.raise_for_status() + async with async_playwright() as p: + # Se connecter à Browserless + browser = await p.chromium.connect_over_cdp(BROWSERLESS_URL) - # Parser le HTML - soup = BeautifulSoup(response.text, "html.parser") - results = [] + try: + context = await browser.new_context( + viewport={"width": 1920, "height": 1080}, + user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + locale="fr-FR", + ) + page = await context.new_page() - # Extraire les liens produits - if product_selector: - links = soup.select(product_selector) - else: - # Fallback: chercher tous les liens qui ressemblent à des produits - links = _find_product_links(soup, domain) + # Bloquer les ressources inutiles pour accélérer + await page.route("**/*.{png,jpg,jpeg,gif,webp,svg,ico}", lambda route: route.abort()) + await page.route("**/analytics*", lambda route: route.abort()) + await page.route("**/tracking*", lambda route: route.abort()) - logger.info(f"{domain}: {len(links)} liens produits trouvés") + # Naviguer vers la page de recherche + await page.goto(final_url, wait_until="domcontentloaded", timeout=int(timeout * 1000)) - seen_urls = set() - for link in links[:max_results * 2]: # Prendre plus pour filtrer - href = link.get("href", "") - if not href: + # Attendre que les résultats se chargent + if wait_selector: + try: + await page.wait_for_selector(wait_selector, timeout=10000) + except Exception: + logger.debug(f"Selector {wait_selector} non trouvé, on continue...") + + # Attendre un peu pour le JS + await asyncio.sleep(2) + + # Extraire le HTML + html_content = await page.content() + + await context.close() + + finally: + await browser.close() + + # Parser le HTML avec BeautifulSoup + soup = BeautifulSoup(html_content, "lxml") + results = [] + seen_urls = set() + + # Trouver les liens produits + if product_selector: + # Essayer plusieurs sélecteurs séparés par virgule + selectors = [s.strip() for s in product_selector.split(",")] + links = [] + for sel in selectors: + try: + found = soup.select(sel) + links.extend(found) + except Exception: continue + else: + links = _find_product_links(soup, domain) - # Construire l'URL complète - if href.startswith("/"): - href = f"https://{domain}{href}" - elif not href.startswith("http"): - href = urljoin(final_url, href) + logger.info(f"{domain}: {len(links)} liens trouvés avec sélecteur") - # Nettoyer l'URL - href = _clean_product_url(href) + for link in links: + href = link.get("href", "") + if not href: + continue - # Éviter les doublons - if href in seen_urls: - continue - seen_urls.add(href) + # Construire l'URL complète + if href.startswith("/"): + href = f"https://{domain}{href}" + elif not href.startswith("http"): + href = urljoin(final_url, href) - # Vérifier que c'est bien un lien vers ce domaine - url_domain = _extract_domain(href) - if domain not in url_domain: - continue + # Nettoyer l'URL + href = _clean_product_url(href, domain) - # Extraire le titre - title = _extract_title(link) - if not title or len(title) < 3: - continue + # Éviter les doublons + if href in seen_urls: + continue + seen_urls.add(href) - results.append(SearchResult( - url=href, - title=title, - snippet="", - source=domain, - )) + # Vérifier que c'est bien un lien vers ce domaine + url_domain = _extract_domain(href) + if domain not in url_domain and url_domain not in domain: + continue - if len(results) >= max_results: - break + # Filtrer les liens non-produits + if _is_non_product_url(href): + continue - logger.info(f"{domain}: {len(results)} résultats extraits") - return results + # Extraire le titre + title = _extract_title(link) + if not title or len(title) < 3: + continue + + results.append(SearchResult( + url=href, + title=title, + snippet="", + source=domain, + )) + + if len(results) >= max_results: + break + + logger.info(f"{domain}: {len(results)} résultats extraits") + return results - except httpx.TimeoutException: - logger.error(f"Timeout lors de la recherche sur {domain}") - return [] - except httpx.HTTPStatusError as e: - logger.error(f"Erreur HTTP {e.response.status_code} sur {domain}") - return [] except Exception as e: - logger.error(f"Erreur lors de la recherche sur {domain}: {e}") + logger.error(f"Erreur Browserless sur {domain}: {e}") return [] def _get_default_search_url(domain: str) -> str | None: """Retourne l'URL de recherche par défaut pour un domaine""" - # Chercher une correspondance exacte ou partielle for key, config in DEFAULT_SITE_CONFIGS.items(): if key in domain or domain in key: return config.get("search_url") @@ -235,6 +260,14 @@ def _get_default_product_selector(domain: str) -> str | None: return None +def _get_default_wait_selector(domain: str) -> str | None: + """Retourne le sélecteur d'attente par défaut pour un domaine""" + for key, config in DEFAULT_SITE_CONFIGS.items(): + if key in domain or domain in key: + return config.get("wait_selector") + return None + + def _find_product_links(soup: BeautifulSoup, domain: str) -> list: """Trouve les liens produits de manière générique""" links = [] @@ -248,17 +281,18 @@ def _find_product_links(soup: BeautifulSoup, domain: str) -> list: "a[href*='/article']", "a.product", "a.product-link", - "a.product-card", ".product a", ".product-card a", ".product-tile a", - ".item a", ] for pattern in patterns: - found = soup.select(pattern) - if found: - links.extend(found) + try: + found = soup.select(pattern) + if found: + links.extend(found) + except Exception: + continue # Dédupliquer seen = set() @@ -276,7 +310,7 @@ def _extract_title(element) -> str: """Extrait le titre d'un élément""" # Essayer différentes méthodes title = element.get("title", "") - if title: + if title and len(title) > 3: return title.strip() # Texte de l'élément @@ -299,8 +333,8 @@ def _extract_title(element) -> str: return "" -def _clean_product_url(url: str) -> str: - """Nettoie une URL produit (retire les paramètres de tracking, etc.)""" +def _clean_product_url(url: str, domain: str) -> str: + """Nettoie une URL produit""" # Pour Amazon, simplifier l'URL if "amazon." in url and "/dp/" in url: match = re.search(r"(/dp/[A-Z0-9]{10})", url) @@ -309,7 +343,7 @@ def _clean_product_url(url: str) -> str: if domain_match: return domain_match.group(1) + match.group(1) - return url + return url.split("?")[0] if "?" in url else url def _extract_domain(url: str) -> str: @@ -324,11 +358,25 @@ def _extract_domain(url: str) -> str: return "" +def _is_non_product_url(url: str) -> bool: + """Vérifie si l'URL n'est pas une page produit""" + non_product_patterns = [ + "/search", "/category", "/categories", "/brand", "/brands", + "/help", "/contact", "/about", "/account", "/cart", "/checkout", + "/login", "/register", "/wishlist", "/compare", "/reviews", + "/terms", "/privacy", "/faq", "/shipping", "/returns", + ] + url_lower = url.lower() + return any(pattern in url_lower for pattern in non_product_patterns) + + async def health_check() -> bool: - """Vérifie la connectivité""" + """Vérifie la connectivité avec Browserless""" try: - async with httpx.AsyncClient(timeout=5.0) as client: - response = await client.get("https://www.google.com") - return response.status_code == 200 - except Exception: + async with async_playwright() as p: + browser = await p.chromium.connect_over_cdp(BROWSERLESS_URL) + await browser.close() + return True + except Exception as e: + logger.warning(f"Browserless health check failed: {e}") return False