From 9a6ebf6f0637687fa5778378007d0c36d664872d Mon Sep 17 00:00:00 2001 From: Michael SCHAL Date: Sun, 23 Nov 2025 15:45:09 +0100 Subject: [PATCH] feat: Implement direct product search service using Browserless for e-commerce sites and add a search flow verification script. --- app/services/direct_search_service.py | 152 ++++++++++++++++++-------- app/services/search_service.py | 27 +++-- verify_search.py | 111 +++++++++++++++++++ 3 files changed, 234 insertions(+), 56 deletions(-) create mode 100644 verify_search.py diff --git a/app/services/direct_search_service.py b/app/services/direct_search_service.py index 4821dac..62d512d 100644 --- a/app/services/direct_search_service.py +++ b/app/services/direct_search_service.py @@ -28,11 +28,23 @@ USER_AGENT = ( class SearchResult: """Résultat de recherche""" - def __init__(self, url: str, title: str, snippet: str, source: str): + def __init__( + self, + url: str, + title: str, + snippet: str, + source: str, + price: float | None = None, + currency: str = "EUR", + in_stock: bool | None = None, + ): self.url = url self.title = title self.snippet = snippet self.source = source + self.price = price + self.currency = currency + self.in_stock = in_stock def __repr__(self): return f"SearchResult(url={self.url}, title={self.title[:30]}...)" @@ -423,19 +435,55 @@ async def search( all_results = [] results_per_site = max(5, max_results // len(sites)) - # Rechercher sur chaque site en séquence (pour éviter de surcharger Browserless) - for site in sites: - 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}") + logger.info(f"Démarrage recherche parallèle sur {len(sites)} sites pour '{query}'") - if len(all_results) >= max_results: - break + try: + async with async_playwright() as p: + # Se connecter à Browserless UNE SEULE FOIS pour tous les sites + logger.info(f"Connexion à Browserless: {BROWSERLESS_URL}") + browser = await p.chromium.connect_over_cdp(BROWSERLESS_URL) - logger.info(f"Recherche Browserless: {len(all_results)} résultats totaux") + try: + # Créer des tâches pour chaque site + tasks = [] + for site in sites: + task = _search_site_browserless( + query=query, + site=site, + max_results=results_per_site, + timeout=timeout, + browser=browser, # Passer l'instance du navigateur + ) + tasks.append(task) + + # Exécuter en parallèle avec asyncio.gather + # return_exceptions=True permet de continuer même si un site plante + results_list = await asyncio.gather(*tasks, return_exceptions=True) + + # Agréger les résultats + for i, result in enumerate(results_list): + site_domain = sites[i].get("domain", "inconnu") + + if isinstance(result, Exception): + logger.error(f"Erreur fatale recherche {site_domain}: {result}") + continue + + if result: + all_results.extend(result) + logger.info(f"Site {site_domain}: {len(result)} résultats") + else: + logger.info(f"Site {site_domain}: 0 résultat") + + finally: + # Fermer le navigateur à la fin + if browser: + await browser.close() + logger.info("Navigateur Browserless fermé") + + except Exception as e: + logger.error(f"Erreur globale recherche Browserless: {e}") + + logger.info(f"Recherche Browserless terminée: {len(all_results)} résultats totaux") return all_results[:max_results] @@ -444,6 +492,7 @@ async def _search_site_browserless( # noqa: PLR0912, PLR0915 site: dict, max_results: int, timeout: float, + browser, # Instance de navigateur partagée ) -> list[SearchResult]: """Recherche sur un site via Browserless (Playwright)""" raw_domain = site.get("domain", "").lower() @@ -464,47 +513,42 @@ async def _search_site_browserless( # noqa: PLR0912, PLR0915 logger.info(f"Recherche Browserless sur {domain}: {final_url}") + context = None + page = None + try: - async with async_playwright() as p: - # Se connecter à Browserless - browser = await p.chromium.connect_over_cdp(BROWSERLESS_URL) + # Créer un nouveau contexte isolé pour ce site + context = await browser.new_context( + viewport={"width": 1920, "height": 1080}, + user_agent=USER_AGENT, + locale="fr-FR", + ) + + # Bloquer les ressources inutiles + await context.route("**/*.{png,jpg,jpeg,gif,webp,svg,ico}", lambda route: route.abort()) + await context.route("**/analytics*", lambda route: route.abort()) + await context.route("**/tracking*", lambda route: route.abort()) + + page = await context.new_page() + # Naviguer vers la page de recherche + await page.goto(final_url, wait_until="domcontentloaded", timeout=int(timeout * 1000)) + + # Accepter les cookies si nécessaire + await _accept_cookies(page, domain) + + # Attendre que les résultats se chargent + if wait_selector: try: - context = await browser.new_context( - viewport={"width": 1920, "height": 1080}, - user_agent=USER_AGENT, - locale="fr-FR", - ) - page = await context.new_page() + await page.wait_for_selector(wait_selector, timeout=10000) + except Exception: + logger.debug(f"Selector {wait_selector} non trouvé sur {domain}, on continue...") - # 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()) + # Attendre un peu pour le JS (réduit pour parallélisation) + await asyncio.sleep(1.5) - # Naviguer vers la page de recherche - await page.goto(final_url, wait_until="domcontentloaded", timeout=int(timeout * 1000)) - - # Accepter les cookies si nécessaire - await _accept_cookies(page, domain) - - # 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() + # Extraire le HTML + html_content = await page.content() # Parser le HTML avec BeautifulSoup soup = BeautifulSoup(html_content, "lxml") @@ -568,6 +612,8 @@ async def _search_site_browserless( # noqa: PLR0912, PLR0915 title=title, snippet="", source=domain, + # On ne peut pas extraire le prix facilement ici sans scraper chaque page + # Le scraping détaillé se fera dans la phase 2 )) if len(results) >= max_results: @@ -579,6 +625,16 @@ async def _search_site_browserless( # noqa: PLR0912, PLR0915 except Exception as e: logger.error(f"Erreur Browserless sur {domain}: {e}") return [] + + finally: + # Fermer le contexte et la page, mais PAS le navigateur + try: + if page: + await page.close() + if context: + await context.close() + except Exception as e: + logger.debug(f"Erreur fermeture contexte {domain}: {e}") def _get_default_search_url(domain: str) -> str | None: diff --git a/app/services/search_service.py b/app/services/search_service.py index ed3ce94..2e4a14d 100644 --- a/app/services/search_service.py +++ b/app/services/search_service.py @@ -89,18 +89,29 @@ async def search_products( results=[], ) - search_results = await direct_search_service.search( - query=query, - sites=sites_data, - max_results=max_results, - ) - - if not search_results: + try: + search_results = await direct_search_service.search( + query=query, + sites=sites_data, + max_results=max_results, + ) + except Exception as e: + logger.error(f"Erreur critique lors de la recherche directe: {e}") yield SearchProgress( status="error", total=0, completed=0, - message="Aucun résultat trouvé", + message=f"Erreur de recherche: {str(e)}", + results=[], + ) + return + + if not search_results: + yield SearchProgress( + status="completed", # Changed from error to completed to avoid scary red alerts for just 0 results + total=0, + completed=0, + message="Aucun résultat trouvé sur les sites configurés", results=[], ) return diff --git a/verify_search.py b/verify_search.py new file mode 100644 index 0000000..f40aa86 --- /dev/null +++ b/verify_search.py @@ -0,0 +1,111 @@ +import asyncio +import logging +import sys +from unittest.mock import MagicMock, AsyncMock + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Mock dependencies +sys.modules["sqlalchemy"] = MagicMock() +sys.modules["sqlalchemy.orm"] = MagicMock() +sys.modules["app.database"] = MagicMock() +sys.modules["app.models"] = MagicMock() +sys.modules["app.services.ai_service"] = MagicMock() +sys.modules["app.services.scraper_service"] = MagicMock() +sys.modules["app.services.light_scraper_service"] = MagicMock() +sys.modules["app.services.settings_service"] = MagicMock() + +# Define dummy classes for schemas +class MockSearchProgress: + def __init__(self, status, total, completed, message, results, current_site=None): + self.status = status + self.total = total + self.completed = completed + self.message = message + self.results = results + self.current_site = current_site + + def model_dump_json(self): + return "json" + +class MockSearchResultItem: + def __init__(self, **kwargs): + for k, v in kwargs.items(): + setattr(self, k, v) + +# Setup schema mocks +schemas_mock = MagicMock() +schemas_mock.SearchProgress = MockSearchProgress +schemas_mock.SearchResultItem = MockSearchResultItem +sys.modules["app.schemas"] = schemas_mock + +# Import services after mocking +# We need to mock direct_search_service before importing search_service +# because search_service imports it. +direct_search_mock = MagicMock() +sys.modules["app.services.direct_search_service"] = direct_search_mock + +# Now we can import search_service +# We might need to mock other things that search_service imports +from app.services import search_service + +# Define a mock SearchResult class matching the one in direct_search_service +class MockSearchResult: + def __init__(self, url, title, source, price=None, currency="EUR", in_stock=None): + self.url = url + self.title = title + self.source = source + self.snippet = "snippet" + self.price = price + self.currency = currency + self.in_stock = in_stock + +async def test_search_flow(): + print("--- Starting Search Flow Verification ---") + + # Setup mocks + db = MagicMock() + + # Mock SettingsService + search_service.SettingsService.get_setting_value.side_effect = lambda db, key, default: default + + # Mock direct_search_service.search + mock_results = [ + MockSearchResult("http://site1.com/p1", "Product 1", "site1.com"), + MockSearchResult("http://site2.com/p2", "Product 2", "site2.com"), + ] + direct_search_mock.search = AsyncMock(return_value=mock_results) + direct_search_mock.SearchResult = MockSearchResult + + # Mock light_scraper_service + search_service.light_scraper_service.scrape_url = AsyncMock(return_value=MagicMock(success=False)) + + # Mock _scrape_with_browserless (to avoid actual scraping) + # We need to patch it in the module + original_scrape = search_service._scrape_with_browserless + search_service._scrape_with_browserless = AsyncMock(return_value=MagicMock( + url="http://site1.com/p1", + title="Product 1", + price=10.0, + site_name="Site 1", + site_domain="site1.com" + )) + + # Mock _get_sites to return some dummy sites + mock_site1 = MagicMock(domain="site1.com", name="Site 1", requires_js=True) + mock_site2 = MagicMock(domain="site2.com", name="Site 2", requires_js=True) + search_service._get_sites = MagicMock(return_value=[mock_site1, mock_site2]) + + # Run the search + print("Running search_products...") + async for progress in search_service.search_products("test query", db): + print(f"Event: {progress.status} - {progress.message}") + if progress.results: + print(f" Results: {len(progress.results)}") + + print("--- Verification Complete ---") + +if __name__ == "__main__": + asyncio.run(test_search_flow())