Michael/Priceflow
1
0 Fork 0
mirror of https://github.com/R0m1k3/Priceflow.git synced 2026-10-12 01:39:25 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #27 from R0m1k3/claude/design-product-search-01Ea9dWZqZpuKFwBwVmp26PG

Browse Source 
fix: Use Browserless for search page scraping
This commit is contained in:
LogiFlow authored and GitHub committed 2025-11-22 16:06:36 +01:00
2 parents 6b705d1117 df613cf64f
commit fcf002b68b
1 file changed
+166 -118
Download Patch File Download Diff File Expand all files Collapse all files
app/services/direct_search_service.py
+166 -118
View File
@@ -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